Skip to content

Using FastCRUD with SQLModel

Since SQLModel is just a combination of SQLAlchemy and Pydantic, the process simplifies as SQLModel combines the model and schema definitions.

Wherever in the docs you see a SQLAlchemy model or Pydantic schema being used, you may just replace it with SQLModel and it will work. For the quick start:

Define your SQLModel model

setup.py
from sqlmodel import SQLModel, Field
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.asyncio import AsyncSession


class Item(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str | None = None
    description: str | None = None
    category: str | None = None
    price: float | None = None
    last_sold: datetime.datetime | None = None
    created_at: datetime.datetime = Field(
        nullable=False, sa_column_kwargs={"default": func.now()}
    )


class ItemSchema(SQLModel):
    name: str | None = None
    description: str | None = None
    category: str | None = None
    price: float | None = None
    last_sold: datetime.datetime | None = None


DATABASE_URL = "sqlite+aiosqlite:///./test.db"
engine = create_async_engine(DATABASE_URL, echo=True)
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

Then your schemas

setup.py
from sqlmodel import SQLModel, Field
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.asyncio import AsyncSession


class Item(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str | None = None
    description: str | None = None
    category: str | None = None
    price: float | None = None
    last_sold: datetime.datetime | None = None
    created_at: datetime.datetime = Field(
        nullable=False, sa_column_kwargs={"default": func.now()}
    )


class ItemSchema(SQLModel):
    name: str | None = None
    description: str | None = None
    category: str | None = None
    price: float | None = None
    last_sold: datetime.datetime | None = None


DATABASE_URL = "sqlite+aiosqlite:///./test.db"
engine = create_async_engine(DATABASE_URL, echo=True)
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

And, finally, your database connection

setup.py
from sqlmodel import SQLModel, Field
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.asyncio import AsyncSession


class Item(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str | None = None
    description: str | None = None
    category: str | None = None
    price: float | None = None
    last_sold: datetime.datetime | None = None
    created_at: datetime.datetime = Field(
        nullable=False, sa_column_kwargs={"default": func.now()}
    )


class ItemSchema(SQLModel):
    name: str | None = None
    description: str | None = None
    category: str | None = None
    price: float | None = None
    last_sold: datetime.datetime | None = None


DATABASE_URL = "sqlite+aiosqlite:///./test.db"
engine = create_async_engine(DATABASE_URL, echo=True)
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

Use crud_router and include it in your FastAPI application

main.py
from fastcrud import crud_router

# Database session dependency
async def get_session() -> AsyncGenerator[AsyncSession, None]:
    async with async_session() as session:
        yield session

# Create tables before the app start
async def lifespan(app: FastAPI):
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield

# FastAPI app
app = FastAPI(lifespan=lifespan)

item_router = crud_router(
    session=get_session,
    model=Item,
    create_schema=ItemSchema,
    update_schema=ItemSchema,
    path="/items",
    tags=["Items"]
)

app.include_router(item_router)

By following the above setup, FastCRUD auto-generates CRUD endpoints for your model, accessible through the /docs route of your FastAPI application.