Plane API 导出工具utils.exporters技术解析基于 Schema 的 CSV/JSON/XLSX 数据导出实现【免费下载链接】plane Open-source Jira, Linear, Monday, and ClickUp alternative. Plane is a modern project management platform to manage tasks, sprints, docs, and triage.项目地址: https://gitcode.com/GitHub_Trending/pl/planePlane 的apps/api/plane/utils/exporters模块提供了一套 Schema 驱动的数据导出工具支持将 Django QuerySet 序列化为 CSV、JSON、XLSX 三种格式内置七种类型安全字段、点路径访问嵌套模型、prepare_*自定义转换与上下文预取机制。读完本文你将掌握该模块的完整 API、字段语义与格式差异并能参照IssueExportSchema的真实实现为自己的模型写出可复制、可运行的导出 Schema。模块定位与整体架构该模块位于 apps/api/plane/utils/exporters是一个灵活且可扩展的数据导出工具核心能力包括多格式CSV、JSON、XLSXExcel类型安全字段StringField、NumberField、DateField、DateTimeField、BooleanField、ListField、JSONField自定义转换字段级转换与prepare_*自定义预备方法preparer点路径dotted path记法便捷访问嵌套属性与关联模型按格式差异化处理同一字段在不同格式下自动转换例如列表在 JSON 中保持数组在 CSV 中拼接为逗号分隔字符串。从源码结构看模块由四个层次构成层实现文件职责入口/门面exporter.pyExporter类选择格式、触发序列化、调用 Formatter格式化器formatters.pyBaseFormatter及CSVFormatter/JSONFormatter/XLSXFormatter字段与 Schemaschemas/base.pyExportField基类、七种字段类型、ExportSchema及元类业务 Schemaschemas/issue.pyIssueExportSchemaIssue 模型完整导出的参考实现模块的__init__.py统一对外导出Exporter、全部字段类型、三种 Formatter 与IssueExportSchema因此所有 API 都可以通过from plane.utils.exporters import ...直接获取。快速上手基本用法定义一个继承ExportSchema的类声明字段再把 QuerySet 直接交给Exporterfrom plane.utils.exporters import Exporter, ExportSchema, StringField, NumberField # 定义 Schema class UserExportSchema(ExportSchema): name StringField(sourceusername, labelUser Name) email StringField(sourceemail, labelEmail Address) posts_count NumberField(labelTotal Posts) def prepare_posts_count(self, obj): return obj.posts.count() # 导出数据——只需传入 queryset users User.objects.all() exporter Exporter(format_typecsv, schema_classUserExportSchema) filename, content exporter.export(users_export, users)export()的返回约定在 Exporter.export 中有明确定义返回(filename_with_extension, content)其中 CSV/JSON 的content是strXLSX 是bytes。传入 QuerySet 时Exporter 内部会先调用schema_class.serialize_queryset(data, fieldsfields)完成序列化再把fields合并进options交给 Formatter。导出 Issue使用现成的 IssueExportSchemafrom plane.utils.exporters import Exporter, IssueExportSchema # 获取带预取关联的 issues issues Issue.objects.filter(project_idproject_id).prefetch_related( assignee_details, label_details, issue_module, # ... 其他关联 ) # 导出为 XLSX——直接传入 queryset exporter Exporter(format_typexlsx, schema_classIssueExportSchema) filename, content exporter.export(issues, issues) # 只导出自定义字段 exporter Exporter(format_typejson, schema_classIssueExportSchema) filename, content exporter.export(issues_filtered, issues, fields[id, name, state_name, assignees])分项目单独导出# 每个项目导出为单独文件 for project_id in project_ids: project_issues issues.filter(project_idproject_id) exporter Exporter(format_typecsv, schema_classIssueExportSchema) filename, content exporter.export(fissues-{project_id}, project_issues) # 保存或上传文件这种对 QuerySet 做 filter 再逐个导出的模式与 Plane 真实的 Issue 导出后台任务完全一致bgtasks/export_task.py 中的issue_export_task在multipleTrue时同样按project_id过滤 QuerySet 逐个导出随后打包为 ZIP 上传对象存储并将ExporterHistory记录状态更新为completed/failed。Schema 定义字段类型七种字段类型全部定义在 schemas/base.py均为ExportField的dataclass子类通过覆写_format_value()实现各自的类型转换。StringField转换为字符串。name StringField(sourcename, labelName, defaultN/A)NumberField处理数值int、float。count NumberField(sourceitems_count, labelCount, default0)DateField将 date 对象格式化为%a, %d %b %Y如 Mon, 01 Jan 2024。start_date DateField(sourcestart_date, labelStart Date)源码实现见 DateField._format_value对带strftime的对象使用%a, %d %b %Y格式化否则退回str(raw)。DateTimeField将 datetime 对象格式化为%a, %d %b %Y %I:%M:%S %Z%z。created_at DateTimeField(sourcecreated_at, labelCreated At)BooleanField转换为布尔值。is_active BooleanField(sourceis_active, labelActive, defaultFalse)ListField处理列表/数组值。在 CSV/XLSX 中列表按分隔符默认, 拼接在 JSON 中保持数组。tags ListField(sourcetags, labelTags) assignees ListField(labelAssignees) # 可由自定义 preparer 填充从 ListField._format_value 的源码看列表值会被规整为list单个非列表值会被自动包成单元素列表None则回退到default默认空列表——因此 Formatter 拿到的一定是可迭代的列表。JSONField处理复杂 JSON 可序列化对象dict、dict 列表。CSV/XLSX 中会被序列化为 JSON 字符串JSON 格式中保持为对象结构。metadata JSONField(sourcemetadata, labelMetadata) comments JSONField(labelComments)字段参数所有字段类型支持这三个参数见 ExportFieldsource属性点路径字符串如project.namedefault字段为 None 时的默认值label导出表头中显示的列名。未指定 label 时BaseFormatter._get_field_info 会将字段名按下划线转为 Title Case 作为表头如created_at→ Created At。点路径Dotted Path记法使用点号访问嵌套属性project_name StringField(sourceproject.name, labelProject) owner_email StringField(sourcecreated_by.email, labelOwner Email)实现位于 ExportField._resolve_dotted_path逐段getattr取值中途遇到None或取不到的属性时安全返回None随后按default回退中间节点若是 dict则按 key 取值。这让project.name、created_by.email这类跨外键的取值无需额外配置。自定义 Preparer复杂逻辑通过定义prepare_{field_name}方法实现class MySchema(ExportSchema): assignees ListField(labelAssignees) def prepare_assignees(self, obj): return [f{u.first_name} {u.last_name} for u in obj.assignee_details]Preparer 的优先级高于字段定义。这在 ExportSchema.serialize 中体现得很直接每个字段名先查找prepare_{field_name}若存在且可调用就直接执行否则才走export_field.get_value(obj, self.context)。另外注意若fields过滤列表中请求了 Schema 未声明的字段名该字段会被静默跳过而不是报错。使用 Preparer 做自定义转换任何自定义逻辑或转换都可以用prepare_field_name方法表达class MySchema(ExportSchema): name StringField(sourcename, labelName (Uppercase)) status StringField(labelStatus) def prepare_name(self, obj): 将 name 字段转换为大写。 return obj.name.upper() if obj.name else def prepare_status(self, obj): 根据模型状态计算 status。 return Active if obj.is_active else Inactive导出格式详解CSV 格式字段使用QUOTE_ALL引号包裹列表按, 拼接可用list_joiner选项自定义JSON 对象序列化为 JSON 字符串文件扩展名.csv。exporter Exporter( format_typecsv, schema_classMySchema, options{list_joiner: ; } # 自定义分隔符 )实现见 CSVFormatter_format_field_value负责单值转换None→ 空串、list→ 拼接、dict→json.dumps_create_csv_file使用csv.writer(buf, delimiter,, quotingcsv.QUOTE_ALL)写入StringIO。一个容易忽视的行为是records 为空时format直接返回(f{filename}.csv, )即空内容字符串。JSON 格式列表保持数组对象保持嵌套结构保留数据类型文件扩展名.json。exporter Exporter(format_typejson, schema_classMySchema) filename, content exporter.export(data, records) # content 为 JSON 字符串[{field: value}, ...]从 JSONFormatter 的源码看每条记录的 key 使用的是字段的label而非字段名即 JSON 输出的键就是导出表头文本records 为空时返回[]。XLSX 格式使用 openpyxl 生成 Excel 兼容文件列表按, 拼接可用list_joiner自定义JSON 对象序列化为 JSON 字符串文件扩展名.xlsx返回二进制内容bytes。exporter Exporter(format_typexlsx, schema_classMySchema) filename, content exporter.export(data, records) # content 为 bytesXLSXFormatter 内部创建Workbook后逐行append再经io.BytesIO存回字节流records 为空时仍会生成一个合法的空工作簿而非空字符串这一点与 CSV 不同。用 Context 预取数据规避 N1 查询这是该模块最有价值的机制之一Schema 可以覆写get_context_data()在序列化开始前对整个 QuerySet 做一次批量预取把结果放进self.context供各 preparer 复用避免逐行触发关联查询。class MySchema(ExportSchema): attachment_count NumberField(labelAttachments) def prepare_attachment_count(self, obj): attachments_dict self.context.get(attachments_dict, {}) return len(attachments_dict.get(obj.id, [])) classmethod def get_context_data(cls, queryset): 用一次查询预取所有附件。 attachments_dict get_attachments_dict(queryset) return {attachments_dict: attachments_dict} # Exporter 序列化时会自动使用 get_context_data() queryset MyModel.objects.all() exporter Exporter(format_typecsv, schema_classMySchema) filename, content exporter.export(data, queryset)调用链在 ExportSchema.serialize_queryset 中先调用一次cls.get_context_data(queryset)再用同一个 context 实例化 Schema然后逐对象serialize。默认实现返回空 dict子类按需覆写。真实示例IssueExportSchema 的 Context 用法IssueExportSchema 展示了完整实现。它的get_context_data预取两块数据classmethod def get_context_data(cls, queryset: QuerySet) - Dict[str, Any]: 获取 issue 序列化的 context 数据。 return { attachments_dict: get_issue_attachments_dict(queryset), cycles_dict: get_issue_last_cycles_dict(queryset), }其中 get_issue_attachments_dict 用一条FileAsset.objects.filter(issue_id__in...)查询把所有附件按 issue ID 归组成 dictget_issue_last_cycles_dict 用select_related(cycle)加上按issue_id, -created_at排序在 Python 侧只保留每个 issue 最近的一条 CycleIssue。对应的 preparer 直接从 context 读字典不再发查询def prepare_attachment_count(self, i): return len((self.context.get(attachments_dict) or {}).get(i.id, [])) def prepare_cycle_name(self, i): cycles_dict self.context.get(cycles_dict) or {} last_cycle cycles_dict.get(i.id) return last_cycle.cycle.name if last_cycle else IssueExportSchema同时是七种字段类型的综合示范约 28 个字段声明id、project_identifier、state_name、labels、comments、estimate、assignees、attachment_links、cycle_name、relations等见 字段定义区点路径取值如project.identifier、preparer 转换如prepare_id返回f{i.project.identifier}-{i.sequence_id}、context 预取附件与循环全部齐备可作为编写新 Schema 的模板。高级用法注册自定义 Formatter为新的导出格式添加支持from plane.utils.exporters import Exporter, BaseFormatter class XMLFormatter(BaseFormatter): def format(self, filename, records, schema_class, optionsNone): # 实现 return (f{filename}.xml, xml_content) # 注册 formatter Exporter.register_formatter(xml, XMLFormatter) # 使用 exporter Exporter(format_typexml, schema_classMySchema)Exporter 内部维护一个类级FORMATTERS字典默认csv/json/xlsx三项register_formatter直接写入该字典——注册后Exporter(format_typexml, ...)即可生效未注册的格式在构造时抛出ValueError并列出可用格式。查询可用格式formats Exporter.get_available_formats() # 返回[csv, json, xlsx]字段过滤传入fields参数只导出指定字段# 只导出特定字段 exporter Exporter(format_typecsv, schema_classMySchema) filename, content exporter.export( filtered_data, queryset, fields[id, name, email] )从 CSVFormatter.format 看过滤是在 Formatter 层完成的按 Schema 声明顺序保留requested_fields中出现的字段field_order [f for f in field_order if f in requested_fields]因此列顺序始终遵循 Schema 声明顺序而非调用方传入的顺序。扩展 Schema通过继承已有 Schema 并覆写get_context_data()创建扩展 Schemaclass ExtendedIssueExportSchema(IssueExportSchema): custom_field JSONField(labelCustom Data) def prepare_custom_field(self, obj): # 使用 context 中预取的数据 return self.context.get(custom_data, {}).get(obj.id, {}) classmethod def get_context_data(cls, queryset): # 获取父类 context附件等 context super().get_context_data(queryset) # 追加自定义预取数据 context[custom_data] fetch_custom_data(queryset) return context字段继承由元类 ExportSchemaMeta 支撑它在类创建时收集当前类声明的ExportField实例并沿 MRO 合并父类的_declared_fields所以子类可以新增字段、也可以仅覆写 preparer 而复用全部父类字段。手动序列化需要只序列化而不导出文件时直接使用 Schema# 将 queryset 序列化为 dict 列表 data MySchema.serialize_queryset(queryset, fields[id, name]) # 或序列化单个对象 schema MySchema() obj_data schema.serialize(obj)源码级补充CSV 注入防护这是 README 未展开、但生产导出必须知道的一点。CSV/XLSX 导出中若用户可控的字符串以、、-、、Tab、CR、LF 开头会在电子表格应用中被当作公式执行即 CSV 注入攻击。Plane 在 apps/api/plane/utils/csv_utils.py 中实现sanitize_csv_value对以公式触发字符开头的字符串前置单引号使其被当作纯文本sanitize_csv_row对整行批量处理。在 Formatter 中可以看到这条防护线的位置CSVFormatter._create_csv_file 与 XLSXFormatter._create_xlsx_file 在写每一行前都会调用sanitize_csv_row(row)表头同样经过处理。对应的回归测试在 apps/api/plane/tests/unit/utils/test_xlsx_export_sanitization.py它验证HYPERLINK(https://example.com/poc,click)这类 payload 在 XLSX 单元格中data_type ! f不是公式单元格且值被加前缀单引号、普通字符串保持原样、非字符串值如数值5保持数字类型、列表拼接后触发的公式前缀同样被转义、表头也会被消毒。若你要扩展新的 Formatter建议沿用同样的sanitize_csv_row调用并补充类似断言。最佳实践继承自模块文档避免 N1 查询覆写get_context_data()预取关联数据classmethod def get_context_data(cls, queryset): return { attachments: get_attachments_dict(queryset), comments: get_comments_dict(queryset), }使用 label为字段提供描述性 label让导出表头更友好created_at DateTimeField(sourcecreated_at, labelCreated At)处理 None 值为可能为 None 的字段设置合适的默认值count NumberField(sourcecount, default0)复杂逻辑用 Preparer保持字段定义简单把复杂转换放进 preparerdef prepare_assignees(self, obj): return [f{u.first_name} {u.last_name} for u in obj.assignee_details]直接传 QuerySet让 Exporter 处理序列化# 推荐——Exporter 负责序列化 exporter.export(data, queryset) # 避免——除非确有必要才手动序列化 data MySchema.serialize_queryset(queryset) exporter.export(data, data)过滤 QuerySet而非过滤数据多文件导出时对 QuerySet 做 filter而不是对已序列化数据切片# 推荐——高效只序列化需要的部分 for project_id in project_ids: project_issues issues.filter(project_idproject_id) exporter.export(fproject-{project_id}, project_issues) # 避免——提前序列化全部数据 all_data MySchema.serialize_queryset(issues) for project_id in project_ids: project_data [d for d in all_data if d[project_id] project_id] exporter.export(fproject-{project_id}, project_data)API 参考速查Exporter成员说明__init__(format_type, schema_class, optionsNone)format_type格式csv/json/xlsxschema_class定义字段的 Schema 类options可选的格式相关选项 dictexport(filename, data, fieldsNone)filename不含扩展名的文件名dataDjango QuerySet 或 dict 列表fields可选的字段名列表返回(filename_with_extension, content)CSV/JSON 的content为strXLSX 为bytesget_available_formats()类方法返回可用格式类型列表register_formatter(format_type, formatter_class)类方法注册自定义 FormatterExportSchema成员说明__init__(contextNone)context可选 dictpreparer 中可通过self.context访问的预取数据serialize(obj, fieldsNone)返回单个对象序列化后的字段值 dictserialize_queryset(queryset, fieldsNone)类方法返回序列化数据 dict 列表get_context_data(queryset)类方法覆写以预取关联数据返回 context dictExportField所有字段类型的基类继承它可创建自定义字段类型get_value(obj, context)返回该字段的格式化值_format_value(raw)在子类中覆写以实现类型特定格式化。测试写法模块文档推荐的测试方式——围绕导出结果与字段过滤做断言# 测试导出 queryset queryset MyModel.objects.all() exporter Exporter(format_typejson, schema_classMySchema) filename, content exporter.export(test, queryset) assert filename test.json assert isinstance(content, str) # 测试字段过滤 filename, content exporter.export(test, queryset, fields[id, name]) data json.loads(content) assert all(set(item.keys()) {id, name} for item in data) # 测试手动序列化 data MySchema.serialize_queryset(queryset) assert len(data) queryset.count()注意最后一类断言与 JSON 格式的一个细节相关联JSON 输出的 key 是字段 label 而非字段名因此如果你的 label 与字段名不同如labelID对应id断言集合应按 label 编写。小结plane.utils.exporters的价值在于把取数 → 转换 → 格式化三段解耦Schema 负责字段语义与转换逻辑get_context_data负责批量预取Formatter 负责格式差异与输出安全。配合IssueExportSchema这个完整参考实现与 导出后台任务 展示的分项目导出、打包上传流程该模块既可直接复用也提供了向其他模型扩展导出能力的清晰路径。【免费下载链接】plane Open-source Jira, Linear, Monday, and ClickUp alternative. Plane is a modern project management platform to manage tasks, sprints, docs, and triage.项目地址: https://gitcode.com/GitHub_Trending/pl/plane创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考