fastapi-import-export 0.2.0

A reusable base library for importing and exporting data with FastAPI.

fastapi-import-export

FastAPI-first import/export utilities that keep your domain model decoupled.

Other languages: README_CN.md | README_JP.md

Features

• Async-first lifecycle hooks for import and export.
• Explicit Resource mapping to avoid ORM coupling.
• Pluggable backends for parsing, storage, and validation.
• Streaming export payloads for large datasets.

Requirements

• Python 3.12-3.14
• FastAPI 0.128+

Compatibility Matrix

Component	Supported	Notes
Python	3.12-3.14	Tested with async-first workflows.
FastAPI	0.128+	Uses UploadFile and async endpoints.
Pydantic	2.x	Schemas rely on BaseModel.
polars	1.x	Included by default.
openpyxl	3.x	Included by default.

Why Not django-import-export

• Django-centric ORM/Admin coupling does not fit FastAPI async workflows.
• This library is async-first and designed as a toolkit, not a framework.
• Stable, composable APIs with explicit lifecycle hooks.

Core Boundaries

• Do not manage DB connections.
• Do not own ORM, only adapt.
• Do not handle auth.
• Do not own storage (pluggable backends only).

Install

Standard install (batteries included):

pip install fastapi-import-export
# or
uv add fastapi-import-export

Includes Polars/XLSX/storage helpers. ORM adapters are optional:

pip install fastapi-import-export[sqlalchemy]
# or
pip install fastapi-import-export[sqlmodel]
# or
pip install fastapi-import-export[tortoise]
# or
uv add fastapi-import-export[sqlalchemy]
# or
uv add fastapi-import-export[sqlmodel]
# or
uv add fastapi-import-export[tortoise]

Install your database driver separately if needed (e.g. asyncpg, aiomysql).

Development & unit-test deps:

pip install fastapi-import-export pytest pytest-asyncio pytest-cov anyio
# or
uv add --group dev fastapi-import-export pytest pytest-asyncio pytest-cov anyio

E2E integration-test deps (optional, for running example apps):

pip install fastapi-import-export[sqlalchemy] httpx python-multipart aiosqlite
# or
uv add --group e2e fastapi-import-export[sqlalchemy] httpx python-multipart aiosqlite

Quick Start (Easy)

1) Define a Resource

from fastapi_import_export import Resource


class UserResource(Resource):
    id: int | None
    username: str
    email: str

    field_aliases = {
        "Username": "username",
        "Email": "email",
    }

2) Import CSV/XLSX in One Call

from fastapi import APIRouter, UploadFile
from fastapi_import_export import import_csv

router = APIRouter()


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)


@router.post("/import")
async def import_data(file: UploadFile):
    return await import_csv(
        file,
        resource=UserResource,
        validate_fn=validate_fn,
        persist_fn=persist_fn,
    )

3) Export CSV/XLSX

from fastapi import StreamingResponse
from fastapi_import_export import export_csv


async def query_fn(*, resource, params=None):
    return [
        {"id": 1, "username": "alice"},
        {"id": 2, "username": "bob"},
    ]


payload = await export_csv(query_fn, resource=UserResource)
return StreamingResponse(payload.stream, media_type=payload.media_type)

ORM Adapters (Optional)

Adapters live under fastapi_import_export.contrib for SQLAlchemy/SQLModel/Tortoise. Install one of:

pip install fastapi-import-export[sqlalchemy]
# or
pip install fastapi-import-export[sqlmodel]
# or
pip install fastapi-import-export[tortoise]

What you get:

• Auto field inference from ORM models (column order and required fields).
• Auto type conversion via codecs (Enum/Date/Datetime/Decimal/Bool).
• Override per-field codecs with field_codecs / __import_export_codecs__.

Codecs (Widget System)

Codecs handle common type conversion for import/export. Built-ins include Enum/Date/Datetime/Decimal/Bool. You can register custom codecs per field. Easy layer automatically applies codecs before validate_fn/persist_fn, and formats values during export.

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(),
    }

Resource Model Binding (Lightweight)

If a Resource declares model but does not declare fields, the library infers fields from the ORM model:

• Source: model.__table__.columns (SQLAlchemy/SQLModel) or model._meta (Tortoise)
• Auto-exclude: primary key (id), created_at, updated_at, and soft-delete flags
• Configurable: exclude_fields = ["password"]
• Explicit wins: field_aliases always overrides auto mapping

class BookResource(Resource):
    model = Book
    exclude_fields = ["password"]
    field_aliases = {"Author": "author"}  # overrides auto mapping

Advanced (Hooks)

Advanced APIs live under fastapi_import_export.advanced and expose the full hook-based lifecycle.

1) Define a Resource

from fastapi_import_export import Resource


class UserResource(Resource):
    id: int | None
    username: str
    email: str

    field_aliases = {
        "Username": "username",
        "Email": "email",
    }

2) Build an Importer

from fastapi_import_export.advanced import Importer


importer = Importer(
    parser=parse_fn,
    validator=validate_fn,
    transformer=transform_fn,
    persister=persist_fn,
)

3) FastAPI Integration

from uuid import UUID

from fastapi import APIRouter, UploadFile


router = APIRouter()


@router.post("/import")
async def import_data(file: UploadFile):
    result = await importer.import_data(file=file, resource=UserResource)
    return result

Async Import Example

async def validate_fn(*, data, resource, allow_overwrite=False):
    return data, []


async def persist_fn(*, data, resource, allow_overwrite=False):
    return 100

Large Export (Streaming)

from fastapi import StreamingResponse


payload = await exporter.stream(
    resource=UserResource,
    fmt="csv",
    filename="users.csv",
    media_type="text/csv",
)
return StreamingResponse(payload.stream, media_type=payload.media_type)

Exporter Usage Example

If Excel shows garbled characters for CSV, emit UTF-8 BOM (use utf-8-sig).

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",
)

Pluggable Backend Facades

• parse/storage/validation/db_validation are lazy-loaded facades.
• Missing dependencies only occur if you remove bundled packages or use an optional adapter/backend/driver that is not installed (e.g. ORM adapters without installing the matching extra).

Upload Allowlist Configuration

Priority order: per-call override > resolve_config parameters > environment variables > defaults.

Per-call override:

await svc.upload_parse_validate(
    file=file,
    column_aliases=UserResource.field_mapping(),
    validate_fn=service_validate_fn,
    allowed_extensions=[".csv"],
    allowed_mime_types=["text/csv"],
)

Config-level override:

from fastapi_import_export.config import resolve_config


cfg = resolve_config(
    allowed_extensions=[".csv", ".xlsx"],
    allowed_mime_types=["text/csv", "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"],
)
svc = ImportExportService(db=object(), config=cfg)

Environment variable example:

export IMPORT_EXPORT_ALLOWED_EXTENSIONS=".csv,.xlsx"
export IMPORT_EXPORT_ALLOWED_MIME_TYPES="text/csv,application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"

Unique Constraint Detection

When committing imported data, the library automatically detects unique constraint violations from the database and returns a user-friendly error response. The constraint_parser module supports precise error parsing for five databases:

Database	Error Pattern	Extracted Info
PostgreSQL	Key (col)=(val) already exists.	columns, values, constraint
MySQL / MariaDB	Duplicate entry 'val' for key 'key_name'	values, constraint
SQLite	UNIQUE constraint failed: table.col	columns
SQL Server	Violation of UNIQUE KEY constraint 'name'	values, constraint
Oracle	ORA-00001: unique constraint (SCHEMA.NAME) violated	constraint

You can also use the parser directly in your own code:

from fastapi_import_export.advanced import ConstraintDetail, parse_unique_constraint_error

detail: ConstraintDetail | None = parse_unique_constraint_error(
    str(exc), detail_text=getattr(getattr(exc, "orig", None), "detail", "")
)
if detail:
    print(detail.db_type, detail.columns, detail.values)

End-to-End Example

Below is a minimal end-to-end flow that supports upload, validate, preview, and commit.

from uuid import UUID

from fastapi import APIRouter, UploadFile

from fastapi_import_export.advanced import Importer, Resource
from fastapi_import_export.schemas import ImportCommitRequest
from fastapi_import_export.advanced import ImportExportService


class UserResource(Resource):
    id: int | None
    username: str
    email: str

    field_aliases = {
        "Username": "username",
        "Email": "email",
    }


async def parse_fn(*, file: UploadFile, resource: type[Resource]):
    return await some_parse_impl(file=file, resource=resource)


async def validate_fn(*, data, resource: type[Resource], allow_overwrite: bool = False):
    return data, []


async def transform_fn(*, data, resource: type[Resource]):
    return data


async def persist_fn(*, data, resource: type[Resource], allow_overwrite: bool = False) -> int:
    return 100


async def service_validate_fn(db, df, *, allow_overwrite: bool = False):
    return df, []


async def service_persist_fn(db, valid_df, *, allow_overwrite: bool = False) -> int:
    return 100


importer = Importer(
    parser=parse_fn,
    validator=validate_fn,
    transformer=transform_fn,
    persister=persist_fn,
)

router = APIRouter()


@router.post("/import")
async def import_data(file: UploadFile):
    return await importer.import_data(file=file, resource=UserResource)


# Optional: use the service class for upload/preview/commit workflow
svc = ImportExportService(db=object())


@router.post("/import/validate")
async def import_validate(file: UploadFile):
    return await svc.upload_parse_validate(
        file=file,
        column_aliases=UserResource.field_mapping(),
        validate_fn=service_validate_fn,
    )


@router.get("/import/preview")
async def import_preview(import_id: UUID, checksum: str, page: int = 1, page_size: int = 50):
    return await svc.preview(
        import_id=import_id,
        checksum=checksum,
        page=page,
        page_size=page_size,
        kind="all",
    )


@router.post("/import/commit")
async def import_commit(body: ImportCommitRequest):
    return await svc.commit(body=body, persist_fn=service_persist_fn)

FAQ

Why do I get missing dependency errors?

You likely removed bundled dependencies or are using an optional adapter/driver that is not installed (e.g. ORM adapters without the matching extra). Reinstall the base package or add the missing adapter/driver.

Why are my rows filtered after validation?

The service removes rows that fail validation when generating valid.parquet. Use preview with kind=all to inspect the original parsed data.

Troubleshooting

• Upload too large: Increase max_upload_mb when creating ImportExportService.
• checksum mismatch: Ensure the client passes the checksum from upload_parse_validate.
• missing_dependency: Reinstall bundled dependencies or install the required adapter/driver.
• db_conflict errors: Check unique constraints and whether soft-deleted records exist.

Testing

Run the unit test suite (see Install for dependency setup):

pytest tests/ -v

Run the E2E integration tests:

pytest examples/ -v

The examples/ directory contains complete FastAPI applications for SQLAlchemy, SQLModel, and Tortoise ORM, each covering the full HTTP lifecycle: upload, preview, commit, and export against an in-memory SQLite database.

License

MIT