fastapi-crud-engine 0.1.4
FastAPI CRUD router and repository toolkit.
License: MIT
Requires Python: >=3.11

# 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.

[![PyPI version](https://img.shields.io/pypi/v/fastapi-crud-engine)](https://pypi.org/project/fastapi-crud-engine/)
[![Python versions](https://img.shields.io/pypi/pyversions/fastapi-crud-engine)](https://pypi.org/project/fastapi-crud-engine/)
[![License](https://img.shields.io/github/license/Lakeserl/fastapi_crud_engine)](https://github.com/Lakeserl/fastapi_crud_engine/LICENSE)

## Table of Contents

- [Features](#features)
- [Installation](#installation)
- [Quickstart](#quickstart)
- [Core Usage](#core-usage)
- [Feature-by-Feature Usage](#feature-by-feature-usage)
- [Configuration](#configuration)
- [Testing and Development](#testing-and-development)
- [Release and Contributing](#release-and-contributing)
- [License](#license)
- [Acknowledgements](#acknowledgements)
- [Security](#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

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

### Install from PyPI

```bash
python -m pip install fastapi-crud-engine
```

### Install optional extras

```bash
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)

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

## Quickstart

```python
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:

```bash
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

```python
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:

```http
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`.

```python
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

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

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

### 4. Cache

```python
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

```python
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

```python
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

```python
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

```python
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:

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

### 10. Bulk create

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

Disable if not needed:

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

### 11. Global exception handling

```python
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

```python
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