FastAPI Integration¶
Advanced Alchemy’s repository and service patterns work well within FastAPI applications.
Basic Setup¶
Configure SQLAlchemy with FastAPI:
from fastapi import FastAPI
from advanced_alchemy.extensions.fastapi import AdvancedAlchemy, AsyncSessionConfig, SQLAlchemyAsyncConfig
sqlalchemy_config = SQLAlchemyAsyncConfig(
connection_string="sqlite+aiosqlite:///test.sqlite",
session_config=AsyncSessionConfig(expire_on_commit=False),
create_all=True,
commit_mode="autocommit",
)
app = FastAPI()
alchemy = AdvancedAlchemy(config=sqlalchemy_config, app=app)
Models and Schemas¶
Define your SQLAlchemy models and Pydantic schemas:
import datetime
from uuid import UUID
from sqlalchemy import ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship
from pydantic import BaseModel as _BaseModel
from advanced_alchemy.base import UUIDAuditBase, UUIDBase
class BaseModel(_BaseModel):
"""Extend Pydantic's BaseModel to enable ORM mode"""
model_config = {"from_attributes": True}
class AuthorModel(UUIDBase):
__tablename__ = "author"
name: Mapped[str]
dob: Mapped[date | None]
books: Mapped[list[BookModel]] = relationship(back_populates="author", lazy="noload")
class BookModel(UUIDAuditBase):
__tablename__ = "book"
title: Mapped[str]
author_id: Mapped[UUID] = mapped_column(ForeignKey("author.id"))
author: Mapped[AuthorModel] = relationship(lazy="joined", innerjoin=True, viewonly=True)
class Author(BaseModel):
id: UUID | None
name: str
dob: datetime.date | None = None
class AuthorCreate(BaseModel):
name: str
dob: datetime.date | None = None
class AuthorUpdate(BaseModel):
name: str | None = None
dob: datetime.date | None = None
Repository and Service¶
Create repository and service classes:
from advanced_alchemy.extensions.fastapi import repository, service
class AuthorService(service.SQLAlchemyAsyncRepositoryService[AuthorModel]):
"""Author service."""
class Repo(repository.SQLAlchemyAsyncRepository[AuthorModel]):
"""Author repository."""
model_type = AuthorModel
repository_type = Repo
Dependency Injection¶
Set up dependency injected into the request context.
DatabaseSession = Annotated[AsyncSession, Depends(alchemy.provide_session())]
Authors = Annotated[AuthorService, Depends(provide_authors_service)]
async def provide_authors_service(db_session: DatabaseSession) -> AsyncGenerator[AuthorService, None]:
"""This provides the default Authors repository."""
async with AuthorService.new(session=db_session) as service:
yield service
Controllers¶
Create controllers using the service:
from fastapi import APIRouter, Depends
from uuid import UUID
from advanced_alchemy.extensions.fastapi import filters
author_router = APIRouter()
@author_router.get(path="/authors", response_model=filters.OffsetPagination[Author])
async def list_authors(
authors_service: Authors,
limit_offset: Annotated[filters.LimitOffset, Depends(provide_limit_offset_pagination)],
) -> filters.OffsetPagination[AuthorModel]:
"""List authors."""
results, total = await authors_service.list_and_count(limit_offset)
return authors_service.to_schema(results, total, filters=[limit_offset])
@author_router.post(path="/authors", response_model=Author)
async def create_author(
authors_service: Authors,
data: AuthorCreate,
) -> AuthorModel:
"""Create a new author."""
obj = await authors_service.create(data)
return authors_service.to_schema(obj)
@author_router.get(path="/authors/{author_id}", response_model=Author)
async def get_author(
authors_service: Authors,
author_id: UUID,
) -> AuthorModel:
"""Get an existing author."""
obj = await authors_service.get(author_id)
return authors_service.to_schema(obj)
@author_router.patch(path="/authors/{author_id}", response_model=Author)
async def update_author(
authors_service: Authors,
data: AuthorUpdate,
author_id: UUID,
) -> AuthorModel:
"""Update an author."""
obj = await authors_service.update(data, item_id=author_id)
return authors_service.to_schema(obj)
@author_router.delete(path="/authors/{author_id}")
async def delete_author(
authors_service: Authors,
author_id: UUID,
) -> None:
"""Delete an author from the system."""
_ = await authors_service.delete(author_id)
Application Configuration¶
Finally, configure your FastAPI application with the router:
app.include_router(author_router)