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.