Using BetterCRUD with SQLModel¶
Since SQLModel combines the functionality of sqlalchemy and pydantic, all you need to do is replace the model with SQLModel
Warning
Prerequisites: prepare your database. Only asynchronous mode is supported — use aiomysql or aiosqlite.
db.py
from typing import AsyncGenerator
from sqlmodel import SQLModel
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import NullPool
from sqlalchemy.ext.asyncio import create_async_engine
DATABASE_URL = "sqlite+aiosqlite:///crud.db"
engine = create_async_engine(
DATABASE_URL,
echo=False,
poolclass=NullPool
)
SessionLocal = sessionmaker(
autocommit=False,
autoflush=False,
bind=engine,
class_=AsyncSession,
expire_on_commit=False,
)
async def get_session() -> AsyncGenerator[AsyncSession, None]:
async with SessionLocal() as session:
yield session
async def init_db():
async with engine.begin() as conn:
await conn.run_sync(SQLModel.metadata.create_all)
First Define your SQLModel model
model.py
from typing import Optional, List
from sqlmodel import Field, SQLModel, Relationship
class PetBase(SQLModel):
name: Optional[str] = None
description: Optional[str] = Field(default=None)
class Pet(PetBase, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
class PetPublic(PetBase):
id: Optional[int]
class PetCreate(PetBase):
pass
class PetUpdate(PetBase):
pass
Next we need to create a service:
service.py
from better_crud.service.sqlalchemy import SqlalchemyCrudService
from .model import Pet
class PetService(SqlalchemyCrudService[Pet]):
def __init__(self):
super().__init__(Pet)
Next we need to define the controller and decorate it with the crud decorator. The controller is just a normal class — the crud decorator gives it super powers.
controller.py
from fastapi import APIRouter, Depends
from better_crud import crud
from .model import PetCreate, PetUpdate, PetPublic
from .service import PetService
pet_router = APIRouter()
@crud(
pet_router,
dto={
"create": PetCreate,
"update": PetUpdate
},
serialize={
"base": PetPublic,
}
)
class PetController():
service: PetService = Depends(PetService)
Next we can register router to the fastapi routing system
main.py
from better_crud import BetterCrudGlobalConfig
from fastapi import FastAPI
from contextlib import asynccontextmanager
from .db import get_session, init_db
BetterCrudGlobalConfig.init(
backend_config={
"sqlalchemy": {
"db_session": get_session
}
}
)
@asynccontextmanager
async def lifespan(_: FastAPI):
await init_db()
# Shutdown
yield
app = FastAPI(lifespan=lifespan)
def register_router():
from app.controller import pet_router
app.include_router(pet_router, prefix="/pet")
register_router()
And it's all done, just go to /docs and the crud endpoints are created.