auen 0.2.0

Auto-generate FastAPI CRUD endpoints from SQLModel models

auen

Auto-generate FastAPI CRUD endpoints from SQLModel models.

Installation

pip install auen

Quickstart

from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager

from fastapi import FastAPI
from sqlalchemy.ext.asyncio import create_async_engine
from sqlmodel import Field, SQLModel
from sqlmodel.ext.asyncio.session import AsyncSession

from auen import CrudRouterBuilder

engine = create_async_engine("sqlite+aiosqlite:///heroes.db")


class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    secret_name: str
    age: int | None = None


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)
    yield


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


app = FastAPI(lifespan=lifespan)
app.include_router(CrudRouterBuilder.for_model(Hero, get_session).build())

This gives you:

Method	Path	Description
POST	/heros/	Create a hero
GET	/heros/	List heroes
GET	/heros/{id}	Get a hero
PATCH	/heros/{id}	Update a hero
DELETE	/heros/{id}	Delete a hero

Explicit Schemas (Recommended)

For production, define explicit schemas instead of auto-derived ones:

from pydantic import BaseModel
from auen import CrudRouterBuilder, SchemaConfig


class HeroCreate(BaseModel):
    name: str
    secret_name: str
    age: int | None = None


class HeroRead(BaseModel):
    id: int
    name: str
    secret_name: str
    age: int | None = None


class HeroUpdate(BaseModel):
    name: str | None = None
    secret_name: str | None = None
    age: int | None = None


app.include_router(
    CrudRouterBuilder.for_model(Hero, get_session)
    .with_schemas(
        SchemaConfig(
            create=HeroCreate,
            read=HeroRead,
            update=HeroUpdate,
        )
    )
    .build()
)

Authentication

from fastapi import Header, HTTPException
from auen import AuthConfig, CrudRouterBuilder


def get_current_user(authorization: str = Header()) -> str:
    if not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401)
    return authorization[7:]  # return user identity


app.include_router(
    CrudRouterBuilder.for_model(Hero, get_session)
    .with_auth(AuthConfig(dependency=get_current_user))
    .build()
)

Row-Level Policies

from auen import AuthConfig, CrudRouterBuilder


class OwnerPolicy:
    def can_create(self, user, obj_in):
        return True

    def can_read(self, user, db_obj):
        return db_obj.owner_id == user

    def can_update(self, user, db_obj, obj_in):
        return db_obj.owner_id == user

    def can_delete(self, user, db_obj):
        return db_obj.owner_id == user

    def filter_list_query(self, user, query):
        return query.where(Hero.owner_id == user)


app.include_router(
    CrudRouterBuilder.for_model(Hero, get_session)
    .with_auth(AuthConfig(dependency=get_current_user))
    .with_policy(OwnerPolicy())
    .build()
)

Lifecycle Hooks

from auen import CrudRouterBuilder, HooksConfig


async def send_welcome_email(session, user_obj, current_user):
    ...  # send email, log event, publish to queue, etc.


async def validate_before_create(session, obj_in, current_user):
    ...  # raise HTTPException to abort


app.include_router(
    CrudRouterBuilder.for_model(Hero, get_session)
    .with_hooks(
        HooksConfig(
            before_create=validate_before_create,
            after_create=send_welcome_email,
        )
    )
    .build()
)

Pagination & Filtering

from auen import CrudRouterBuilder, PaginationConfig, FilterConfig, FilterFieldConfig

app.include_router(
    CrudRouterBuilder.for_model(Hero, get_session)
    .with_pagination(PaginationConfig(default_limit=20, max_limit=100))
    .with_filters(
        FilterConfig(
            fields={"name": FilterFieldConfig(ops=frozenset({"eq"}))},
            sort_fields=["name", "age"],
        )
    )
    .build()
)

Selecting Operations

from auen import CrudRouterBuilder, Operation

app.include_router(
    CrudRouterBuilder.for_model(Hero, get_session)
    .with_operations({Operation.LIST, Operation.READ})  # read-only API
    .build()
)

Development

# Install dependencies
uv sync

# Run tests
uv run pytest

# Format
uv run ruff format .

# Lint
uv run ruff check .

# Type check
uv run ty check

License

MIT