fastapi-crud-engine 0.1.2

FastAPI CRUD router and repository toolkit.

fastapi-crud-engine

fastapi-crud-engine là thư viện giúp tạo CRUD API cho FastAPI + SQLAlchemy (async) nhanh hơn, giảm lặp code và tích hợp sẵn nhiều tính năng thường dùng.

Tính năng chính

• Tự sinh CRUD endpoints từ model/schema:
  • list, get, create, update, patch, delete
• Soft delete + restore
• Filter/search/order/range/in/isnull
• Pagination chuẩn (page, size)
• Bulk create
• Export CSV/XLSX, import CSV/XLSX
• Audit trail
• Cache (memory/redis)
• Rate limit (memory/redis)
• Webhook (HTTP hoặc Celery)
• Hooks trước/sau create/update/delete/restore
• Field-level permissions theo role

Cài đặt

Cài local (khuyến nghị khi phát triển)

pip install -e .

Cài thêm optional dependencies

pip install -e ".[excel,redis,celery]"

Quick Start

from __future__ import annotations
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.filters import FilterSet
from fastapi_crud_engine.core.handlers import register_exception_handlers
from fastapi_crud_engine.core.mixins import SoftDeleteMixin, TimestampMixin
from fastapi_crud_engine.router import CRUDRouter

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


class Base(DeclarativeBase):
    pass


class User(SoftDeleteMixin, TimestampMixin, Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
    role: Mapped[str] = mapped_column(String(50), default="user")


class UserSchema(BaseModel):
    id: int | None = None
    email: str
    role: 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,
        filterset=FilterSet(
            fields=["role", "email"],
            search_fields=["email"],
            ordering_fields=["id", "email", "created_at"],
            nullable_fields=["deleted_at"],
            default_ordering="-id",
        ),
    )
)

Endpoints được tạo

Với 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 (khi soft_delete=True)
• POST /users/{pk}/restore (khi soft_delete=True)

Bạn có thể tắt endpoint bằng disable=["import", "bulk", "export", "deleted", "restore"].

Các tham số query phổ biến

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=2025-01-01&created_at__lte=2025-12-31
GET /users?id__in=1,2,3
GET /users?deleted_at__isnull=true

Bật các feature nâng cao

1. Cache

from fastapi_crud_engine.features.cache import Cache

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

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

2. Rate limit

from fastapi_crud_engine.features.rate_limiter import RateLimiter

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

3. 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,
    ),
)

4. Field 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,
)

Mặc định role được đọc từ request.state.role (fallback là "user").

5. Webhooks

from fastapi_crud_engine.features.webhooks import WebhookConfig, WebhookEndpoint

webhooks = WebhookConfig(
    delivery="http",  # hoặc "celery"
    endpoints=[
        WebhookEndpoint(
            url="https://example.com/webhook",
            events=["user.created", "user.updated"],
            secret="your-secret",
        )
    ],
)

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

Exception handling

CRUDRouter đã map custom exceptions của thư viện sang HTTP response. Nếu bạn gọi trực tiếp repository/service và muốn xử lý đồng nhất toàn app, đăng ký thêm:

from fastapi import FastAPI
from fastapi_crud_engine.core.handlers import register_exception_handlers

app = FastAPI()
register_exception_handlers(app)

Chạy test

pytest

Cấu trúc package

• fastapi_crud_engine/router.py: CRUD router + auto endpoint registration
• fastapi_crud_engine/repository.py: repository layer cho model
• fastapi_crud_engine/core/*: filters, pagination, mixins, permissions, exceptions, audit
• fastapi_crud_engine/features/*: cache, rate limiter, export/import, webhooks

License

MIT