القائمة
FastAPI-first import/export toolkit with composable workflows and pluggable backends.
يُحدَّد Prompt الذي يراجع المصدر أولًا بشكل افتراضي. يمكنك التبديل إلى أمر مباشر أو تنزيل نسخة محلية.
اقرأ SKILL.md وأي ملفات مرافقة يعرضها SkillsMP قبل أن تقرر التثبيت.
التثبيت باستخدام Codex أو Claude انسخ هذا Prompt والصقه في Codex أو Claude أو مساعد آخر ليراجع صفحة Skill ويثبّتها لك.
يتجاوز الأمر المباشر Prompt المخصّص للمراجعة. افحص المصدر قبل تشغيله.
npx skills add https://github.com/lijianqiao/fastapi-import-export --skill fastapi-import-exportيبقى الأمر في سطر واحد. مرّر أفقيًا لمراجعته كاملًا قبل النسخ.
تفضّل نسخة محلية؟ نزّل الملفات المتاحة حاليًا لدى SkillsMP.
عرض SKILL.md
استنادا إلى تصنيف SOC المهني
| name | fastapi-import-export |
| description | FastAPI-first import/export toolkit with composable workflows and pluggable backends. |
This skill helps you build and operate import/export workflows for FastAPI. It now includes stable overwrite semantics, unified error codes, template contracts, observability events, and performance regression gating.
本技能用于构建并运维 FastAPI 导入导出流程。 当前能力已覆盖稳定覆盖策略、统一错误码、模板契约、可观测事件与性能门禁。
export_* / import_* top-level functions.ExportOptions / ImportOptions (explicit configuration).reject / upsert / replace.schema_error / type_error / db_conflict.fastapi_import_export.advanced.fastapi_import_export.contrib.* (requires [sqlalchemy] / [sqlmodel] / [tortoise]).event_hook) with lifecycle events.get_book_template_contract).Easy Layer (Top-level)
export_csv(source, *, resource=None, params=None, options=None) -> ExportPayloadexport_xlsx(source, *, resource=None, params=None, options=None) -> ExportPayloadimport_csv(file, *, resource, validate_fn, persist_fn, options=None) -> ImportResultimport_xlsx(file, *, resource, validate_fn, persist_fn, options=None) -> ImportResultOptions Layer
ExportOptions
filename: str | Nonemedia_type: str | Noneinclude_bom: bool (default: False)line_ending: str (default: "\r\n")chunk_size: int (default: 64 * 1024)columns: list[str] | NoneImportOptions
db: Any | Noneallow_overwrite: bool (default: False)overwrite_mode: OverwriteMode | str | Noneunique_fields: list[str] | Nonedb_checks: list[DbCheckSpec] | Noneallowed_extensions: Iterable[str] | Noneallowed_mime_types: Iterable[str] | NoneCore Types
Resource
field_aliases: dict[str, str]field_mapping() -> dict[str, str]export_aliases: dict[str, str]export_mapping() -> dict[str, str]field_codecs: dict[str, Codec]model: Any | Noneexclude_fields: list[str]Importer
import_data(*, file, resource, allow_overwrite=False) -> ImportResultparse(*, file, resource) -> TTablevalidate(*, data, resource, allow_overwrite) -> tuple[TTable, list[TError]]transform(*, data, resource) -> TTablepersist(*, data, resource, allow_overwrite) -> intExporter
query(*, resource, params=None) -> TTableserialize(*, data, fmt) -> bytesrender(*, data, fmt) -> ByteStreamstream(*, resource, fmt, filename, media_type, params=None) -> ExportPayloadImportExportService
__init__(..., redis_client=None, event_hook=None, config=None, base_dir=None, max_upload_mb=20, lock_ttl_seconds=300)upload_parse_validate(*, file, column_aliases, validate_fn, allow_overwrite=False, overwrite_mode=None, unique_fields=None, db_checks=None, allowed_extensions=None, allowed_mime_types=None) -> ImportValidateResponsepreview(*, import_id, checksum, page, page_size, kind) -> ImportPreviewResponsecommit(*, body, persist_fn, lock_namespace="import") -> ImportCommitResponseConfig and Facades
resolve_config(...) -> ImportExportConfigparse_tabular_file(file_path, *, filename)normalize_columns(df, column_mapping)dataframe_to_preview_rows(df)collect_infile_duplicates(df, unique_fields)run_db_checks(*, db, df, specs, allow_overwrite=False)Errors
ImportExportError with message, status_code, details, error_codeParseError / ValidationError / PersistError / ExportErrorImportErrorItem includes row_number, field, type, messageUnified validation error code contract
schema_error: Contract/required/enum/in-file duplicate errors.type_error: Parse/coercion/type conversion errors.db_conflict: Database uniqueness/conflict errors.ImportExportError common error_code values
missing_dependency: Backend/adapter dependency is missing (bundled package removed or optional adapter not installed).unsupported_media_type: File extension or MIME type is not allowed.import_export_error: Default fallback code for generic errors.Easy Export
source can be Iterable[Mapping], polars.DataFrame, or query_fn.source is callable, it is treated as query_fn.options.columns > Resource field order > inferred from rows.Resource.export_mapping() is applied.media_type derived from format; CSV uses \\r\\n and no BOM.Easy Import
import_csv/import_xlsx runs upload -> parse -> validate -> commit.validate_fn and persist_fn only.overwrite_mode > allow_overwrite.ImportResult(status=VALIDATED, errors=...).ImportResult(status=COMMITTED, imported_rows=...).Importer.import_data
file, resourceallow_overwriteImportResult with status, imported_rows, errorsExporter.stream
resource, fmt, filename, media_typeparamsExportPayload with filename, media_type, streamImportExportService.upload_parse_validate
file, column_aliases, validate_fnallow_overwrite, overwrite_mode, unique_fields, db_checks, allowed_extensions, allowed_mime_typesImportValidateResponse with import_id, checksum, total_rows, valid_rows, error_rows, errorsmax_upload_mbImportExportService.preview
import_id (UUID), checksum, page, page_size, kindkind is all or validImportPreviewResponse with rows (list of ImportPreviewRow)ImportExportService.commit
body, persist_fnlock_namespaceImportCommitResponse with imported_rows, status, created_atImportExportService observability
event_hook(event: dict) -> None | awaitableupload_parse_validate.started/completed/failedpreview.started/completed/failedcommit.started/completed/failedImportErrorItem
{
"row_number": 12,
"field": "email",
"type": "schema_error",
"message": "Duplicate value for field email: a@b.com"
}
ImportValidateResponse (partial)
{
"import_id": "uuid",
"checksum": "sha256",
"total_rows": 100,
"valid_rows": 95,
"error_rows": 5,
"errors": [
{"row_number": 12, "field": "email", "message": "Duplicate value for field email: a@b.com"}
]
}
resolve_config(allowed_extensions, allowed_mime_types) or per-call override.openpyxl in easy layer.field_codecs.Resource.model is set and no fields are declared, fields are inferred from the ORM model.validate_fn/persist_fn and formats values during export.ImportOptions.overwrite_mode.field_codecs / __import_export_codecs__.Example (Enum/Date/Decimal)
from enum import Enum
from fastapi_import_export import Resource
from fastapi_import_export.codecs import DateCodec, DecimalCodec, EnumCodec
class Status(Enum):
DRAFT = "draft"
PUBLISHED = "published"
class BookResource(Resource):
field_codecs = {
"status": EnumCodec(Status),
"published_at": DateCodec(),
"price": DecimalCodec(),
}
Rules when Resource.model is set and no fields are declared:
model.__table__.columns (SQLAlchemy/SQLModel) or model._meta (Tortoise)id), created_at, updated_at, soft-delete flagsexclude_fields = ["password"]field_aliases overrides auto mapping (merge: auto mapping then update with field_aliases)class BookResource(Resource):
model = Book
exclude_fields = ["password"]
field_aliases = {"Author": "author"}
0) Easy Export (Top-level)
from fastapi import StreamingResponse
from fastapi_import_export import export_csv, Resource
class UserResource(Resource):
id: int | None
username: str
async def query_fn(*, resource, params=None):
return [{"id": 1, "username": "alice"}]
payload = await export_csv(query_fn, resource=UserResource)
return StreamingResponse(payload.stream, media_type=payload.media_type)
0.1) Easy Import (Top-level)
from fastapi_import_export import import_csv
async def validate_fn(db, df, *, allow_overwrite: bool = False):
return df, []
async def persist_fn(db, valid_df, *, allow_overwrite: bool = False) -> int:
return int(valid_df.height)
result = await import_csv(file, resource=UserResource, validate_fn=validate_fn, persist_fn=persist_fn)
0.2) Easy Import with overwrite mode
from fastapi_import_export import import_csv
from fastapi_import_export.options import ImportOptions
result = await import_csv(
file,
resource=UserResource,
validate_fn=validate_fn,
persist_fn=persist_fn,
options=ImportOptions(overwrite_mode="upsert"),
)
1) Define Resource and Importer
from fastapi_import_export.advanced import Importer, Resource
class UserResource(Resource):
id: int | None
username: str
email: str
field_aliases = {"Username": "username", "Email": "email"}
importer = Importer(
parser=parse_fn,
validator=validate_fn,
transformer=transform_fn,
persister=persist_fn,
)
2) Streaming export
payload = await exporter.stream(
resource=UserResource,
fmt="csv",
filename="users.csv",
media_type="text/csv",
)
2.1) Exporter module usage
import csv
import io
from collections.abc import AsyncIterator
from fastapi_import_export.advanced import Exporter, Resource
class UserResource(Resource):
id: int | None
username: str
async def query_fn(*, resource: type[Resource], params: dict | None = None):
return [
{"id": 1, "username": "alice"},
{"id": 2, "username": "bob"},
]
async def serialize_fn(*, data: list[dict], fmt: str) -> bytes:
buf = io.StringIO()
writer = csv.DictWriter(buf, fieldnames=["id", "username"])
writer.writeheader()
writer.writerows(data)
return buf.getvalue().encode("utf-8-sig")
async def render_fn(*, data: bytes, fmt: str) -> AsyncIterator[bytes]:
async def _stream() -> AsyncIterator[bytes]:
yield data
return _stream()
exporter = Exporter(query_fn=query_fn, serialize_fn=serialize_fn, render_fn=render_fn)
payload = await exporter.stream(
resource=UserResource,
fmt="csv",
filename="users.csv",
media_type="text/csv",
)
3) Service workflow
from fastapi_import_export.advanced import ImportExportService
svc = ImportExportService(db=object())
resp = await svc.upload_parse_validate(
file=file,
column_aliases=UserResource.field_mapping(),
validate_fn=service_validate_fn,
)
commit = await svc.commit(body=commit_body, persist_fn=service_persist_fn)
4) Upload allowlist configuration
from fastapi_import_export.config import resolve_config
cfg = resolve_config(allowed_extensions=[".csv"], allowed_mime_types=["text/csv"])
svc = ImportExportService(db=object(), config=cfg)
fastapi_import_export.advanced.Importer for custom parse/validate/transform/persist.fastapi_import_export.advanced.Exporter for custom query/serialize/render.fastapi_import_export.advanced.ImportExportService for upload/preview/commit workflows.parse, validation, db_validation, storage to plug backends.get_book_template_contract() returns a built-in book import contract.BOOK_TEMPLATE_COLUMNSBOOK_STATUS_ENUMUse these presets to keep header/value contracts stable across front-end, ops, and backend teams.
Benchmark script:
benchmarks/benchmark_import_service.pyKey options:
--kind csv|xlsx--rows <int>--seed <int>--warmup <int>--rounds <int>--export-json <path>--baseline-json <path>--regression-threshold <float>--fail-on-regressionCI workflow:
.github/workflows/performance-gate.yml.perf/baseline.json as baseline.bytesadvanced.Exporter.serialize or your own wrapper in the easy layer.AsyncIterator[bytes].
parse -> validate -> transform -> persist
validate returns (valid_data, errors). If errors is non-empty, skip transform/persist.allow_overwrite is passed through to validation and persistence.query -> serialize -> render
query returns table-like data.serialize returns bytes.render returns an async byte stream.parse and validation default to Polars backends (bundled by default).storage defaults to filesystem storage (bundled by default).\\r\\n line endings.options.columns > Resource field order > inferred.export_aliases > invertible field_aliases > identity.