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

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": "a@b.com"})
# 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