fastapi-crud-engine 0.1.3

FastAPI CRUD router and repository toolkit.

fastapi-crud-engine

Async CRUD engine for FastAPI + SQLAlchemy with built-in filtering, pagination, soft delete, audit logs, cache, rate limiting, import/export, and webhooks.

fastapi-crud-engine helps you ship consistent CRUD APIs faster with less boilerplate and more production-ready defaults.

Table of Contents

• Features
• Installation
• Quickstart
• Core Usage
• Feature-by-Feature Usage
• Configuration
• Testing and Development
• Release and Contributing
• License
• Acknowledgements
• Security

Features

• Automatic CRUD router:
  • GET /, GET /{pk}, POST /, PUT /{pk}, PATCH /{pk}, DELETE /{pk}
• Soft delete and restore endpoints
• Advanced FilterSet support:
  • exact, search, icontains, ordering, range, in, isnull
• Pagination with consistent response schema
• Bulk create endpoint (/bulk)
• CSV/XLSX import and export (/import, /export)
• Audit trail for create/update/delete/restore
• Cache backend (in-memory or Redis)
• Rate limiting (in-memory or Redis)
• Webhook delivery (http or celery)
• Lifecycle hooks (before_*, after_*)
• Field-level permissions by role
• Built-in exception mapping for FastAPI

Installation

Requirements: Python >=3.11

Recommended: install in a virtual environment

python -m venv .venv
source .venv/bin/activate
python -m pip install -U pip

Install from PyPI

python -m pip install fastapi-crud-engine

Install optional extras

python -m pip install "fastapi-crud-engine[excel,redis,celery]"

• excel: enables XLSX import/export via openpyxl
• redis: enables Redis cache and Redis rate limiter
• celery: enables async webhook delivery through Celery workers

Install from source (local development)

git clone https://github.com/Lakeserl/auto-crud.git
cd auto-crud
python -m pip install -e ".[dev]"

Install from TestPyPI (if you publish test builds)

python -m pip install "fastapi<1" "pydantic<3" "SQLAlchemy<3" "httpx<1" "python-dateutil<3"
python -m pip install --index-url https://test.pypi.org/simple --no-deps fastapi-crud-engine==0.1.3

Quickstart

from contextlib import asynccontextmanager
from typing import AsyncGenerator

from fastapi import FastAPI
from pydantic import BaseModel
from sqlalchemy import String
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

from fastapi_crud_engine.core.handlers import register_exception_handlers
from fastapi_crud_engine.core.mixins import SoftDeleteMixin
from fastapi_crud_engine.router import CRUDRouter

engine = create_async_engine("sqlite+aiosqlite:///./app.db")
SessionLocal = async_sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)


class Base(DeclarativeBase):
    pass


class User(SoftDeleteMixin, Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)


class UserSchema(BaseModel):
    id: int | None = None
    email: str
    model_config = {"from_attributes": True}


async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with SessionLocal() as session:
        yield session


@asynccontextmanager
async def lifespan(app: FastAPI):
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield


app = FastAPI(lifespan=lifespan)
register_exception_handlers(app)

app.include_router(
    CRUDRouter(
        model=User,
        schema=UserSchema,
        db=get_db,
        prefix="/users",
        soft_delete=True,
    )
)

Run:

uvicorn main:app --reload

Core Usage

Endpoints generated by CRUDRouter

For prefix="/users":

• GET /users
• GET /users/{pk}
• POST /users
• PUT /users/{pk}
• PATCH /users/{pk}
• DELETE /users/{pk}
• POST /users/bulk
• GET /users/export?fmt=csv|xlsx
• POST /users/import
• GET /users/deleted (when soft_delete=True)
• POST /users/{pk}/restore (when soft_delete=True)

Feature-by-Feature Usage

1. Filtering and pagination

from fastapi_crud_engine.core.filters import FilterSet

router = CRUDRouter(
    ...,
    filterset=FilterSet(
        fields=["role", "status"],
        search_fields=["email", "name"],
        ordering_fields=["id", "created_at", "email"],
        range_fields=["created_at"],
        in_fields=["role"],
        nullable_fields=["deleted_at"],
        default_ordering="-id",
    ),
)

Example queries:

GET /users?page=1&size=20
GET /users?role=admin
GET /users?search=john
GET /users?ordering=-created_at,email
GET /users?created_at__gte=2026-01-01&created_at__lte=2026-12-31
GET /users?role__in=admin,editor
GET /users?deleted_at__isnull=true

2. Soft delete and restore

Your model must inherit SoftDeleteMixin.

from fastapi_crud_engine.core.mixins import SoftDeleteMixin

class User(SoftDeleteMixin, Base):
    ...

router = CRUDRouter(..., soft_delete=True)

When enabled:

• DELETE performs soft delete (deleted_at is set)
• GET /{prefix}/deleted lists soft-deleted records
• POST /{prefix}/{pk}/restore restores a record

3. Audit trail

router = CRUDRouter(..., audit_trail=True)

This logs create/update/delete/restore operations to the audit model generated by build_audit_log_model.

4. Cache

from fastapi_crud_engine.features.cache import Cache

cache = Cache(ttl=60, backend="memory")
# or Cache(ttl=60, backend="redis", redis_url="redis://localhost:6379/0")

router = CRUDRouter(
    ...,
    cache=cache,
    cache_endpoints=["list", "get"],
)

Write operations automatically invalidate model cache keys.

5. Rate limiting

from fastapi_crud_engine.features.rate_limiter import RateLimiter

router = CRUDRouter(
    ...,
    rate_limit=RateLimiter(requests=100, window=60),
)

Default key strategy is client IP. If limit is exceeded, API returns 429 with Retry-After.

6. Field-level permissions

from fastapi_crud_engine.core.permissions import FieldPermissions

permissions = FieldPermissions(
    hidden_by_default=["password_hash"],
    read={"admin": "__all__", "user": ["id", "email", "role"]},
    write={"admin": "__all__", "user": ["email"]},
)

router = CRUDRouter(..., field_permissions=permissions)

Role is read from request.state.role (fallback: "user").

7. Lifecycle hooks

from fastapi_crud_engine.router import CRUDHooks

async def before_create(db, payload):
    ...

async def after_create(db, obj):
    ...

router = CRUDRouter(
    ...,
    hooks=CRUDHooks(
        before_create=before_create,
        after_create=after_create,
    ),
)

Available hooks:

• before_create, after_create
• before_update, after_update
• before_delete, after_delete
• before_restore, after_restore

8. Webhooks

from fastapi_crud_engine.features.webhooks import WebhookConfig, WebhookEndpoint

webhooks = WebhookConfig(
    delivery="http",  # or "celery"
    max_retries=3,
    timeout=10,
    endpoints=[
        WebhookEndpoint(
            url="https://example.com/webhook",
            events=["user.created", "user.updated"],
            secret="super-secret",
            headers={"X-App": "my-service"},
        )
    ],
)

router = CRUDRouter(..., webhooks=webhooks)

Event names look like: modelname.created, modelname.updated, modelname.deleted, modelname.restored.

9. Import and export

• Export: GET /{prefix}/export?fmt=csv|xlsx
• Import: POST /{prefix}/import with a CSV/XLSX file

Disable them if you do not need them:

router = CRUDRouter(..., disable=["import", "export"])

10. Bulk create

• Endpoint: POST /{prefix}/bulk
• Payload: list of create schema objects

Disable if not needed:

router = CRUDRouter(..., disable=["bulk"])

11. Global exception handling

from fastapi_crud_engine.core.handlers import register_exception_handlers

register_exception_handlers(app)

This handles library exceptions consistently (not found, permission denied, lock conflict, rate limit, bulk errors).

12. Using repository directly

from fastapi_crud_engine.repository import CRUDRepository

repo = CRUDRepository(User, soft_delete=True)

# Inside your service/endpoint:
# obj = await repo.create(db, {"email": "[e-mail]"})
# page = await repo.list(db, params=PageParams(page=1, size=20), filter_params=request.query_params)

Configuration

Common router options

• soft_delete=True
• audit_trail=True
• filterset=FilterSet(...)
• cache=Cache(...)
• rate_limit=RateLimiter(...)
• webhooks=WebhookConfig(...)
• hooks=CRUDHooks(...)
• field_permissions=FieldPermissions(...)
• disable=["import", "bulk", "export", "deleted", "restore"]

Environment variables

• REDIS_URL
  • Used by Cache(backend="auto") and RateLimiter(redis_url=None)
• CELERY_BROKER_URL
  • Used when WebhookConfig(delivery="celery")

Contributing

• Read CONTRIBUTING.md
• Create a branch from main
• Add tests for any behavior change
• Open a pull request with clear scope and rationale

License

MIT License. See LICENSE.

Acknowledgements

• Inspired by fastapi-crudrouter: https://github.com/awtkns/fastapi-crudrouter
• Built on top of FastAPI, SQLAlchemy, Pydantic, HTTPX, and the open-source ecosystem