Async FastAPI Project Setup with SQLAlchemy, Alembic, Pydantic, UV & Pyright

Scaffold Project

# create project
uv init fastapi_play
# navigate to project
cd ./fastapi_play
# create virtual env
# search for uv instalation process
uv venv
# activate venv
. ./.venv/bin/activate
# install dependencies
uv add fastapi[standard] pydantic pydantic-settings
# create app dir
mkdir app
# add init file, so app is recognized as module
touch app/__init__.py
# move main into app module 
mv ./main.py ./app/main.py

from typing import Union
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
async def read_root():
 return {"Hello": "World"}
@app.get("/items/{item_id}")
async def read_item(item_id: int, q: Union[str, None] = None):
 return {"item_id": item_id, "q": q}

source ./.venv/bin/activate
uv run fastapi main.py

DB Setup

DB_URL="sqlite+aiosqlite:///temp.sqlite"

source ./.venv/bin/activate
uv add sqlalchemy aiosqlite alembic

from sqlalchemy.orm import DeclarativeBase
from sqlalchemy import Boolean, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
# For alembic to generate migrations it needs to know all our tables
# 
# Since all our models extend the Base model below
# alembic can get all tables using `Base.metadata.tables`
class Base(DeclarativeBase):
 pass
# IMPORTANT:
# In alembic script we need to know about all database models we have
# There are two ways to do this
# 1. import all models in alembic/env.py file
# 2. create a file that contins all our models
# We are using 2nd option
class Todo(Base):
 __tablename__ = "todos"
 id: Mapped[int] = mapped_column(primary_key=True)
 todo: Mapped[str] = mapped_column(String(255))
 done: Mapped[bool]= mapped_column(Boolean(), default=False)

source ./.venv/bin/activate
alembic init --template pyproject_async alembic

from dotenv import load_dotenv
load_dotenv()
DB_URL = os.environ.get('DB_URL')
assert(DB_URL is not None)
print(DB_URL)

def run_migrations_offline() -> None:
 """Run migrations in 'offline' mode.

 This configures the context with just a URL
 and not an Engine, though an Engine is acceptable
 here as well. By skipping the Engine creation
 we don't even need a DBAPI to be available.

 Calls to context.execute() here emit the given string to the
 script output.

 """
 url = DB_URL # this line
 context.configure(
 url=url, # and this line
 target_metadata=target_metadata,
 literal_binds=True,
 dialect_opts={"paramstyle": "named"},
 )
 with context.begin_transaction():
 context.run_migrations()

async def run_async_migrations() -> None:
 """In this scenario we need to create an Engine
 and associate a connection with the context.

 """
 
 section = config.get_section(config.config_ini_section, {}) # this line
 section['sqlalchemy.url'] = str(DB_URL) # this line
 connectable = async_engine_from_config(
 section, # and this line
 prefix="sqlalchemy.",
 poolclass=pool.NullPool,
 )
 # connectable = async_engine_from_config(
 # config.get_section(config.config_ini_section, {}),
 # prefix="sqlalchemy.",
 # poolclass=pool.NullPool,
 # )
 async with connectable.connect() as connection:
 await connection.run_sync(do_run_migrations)
 await connectable.dispose()

source ./.venv/bin/activate
alembic revision --autogenerate -m "added_todos_table"
alembic upgrade head

from fastapi import Depends, FastAPI
load_dotenv()
class DBConfig(BaseSettings):
 model_config = SettingsConfigDict(env_prefix='db_')
 url: str = Field() 
db_config = DBConfig() # pyright: ignore[reportCallIssue]
# create database engine
engine = create_async_engine(db_config.url, echo=True)
# create session factory
session_factory = async_sessionmaker(engine, expire_on_commit=False)
async def get_db():
 db = session_factory()
 try:
 yield db
 finally:
 await db.close()

Add code in main.py to try these things

from typing import Annotated
from dotenv import load_dotenv
from pydantic import BaseModel, ConfigDict, Field
from pydantic_settings import BaseSettings, SettingsConfigDict
from sqlalchemy import select
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.ext.asyncio import async_sessionmaker
from sqlalchemy.ext.asyncio import AsyncSession
from app.models import Todo
from fastapi import Depends, FastAPI
load_dotenv()
app = FastAPI()
class DBConfig(BaseSettings):
 model_config = SettingsConfigDict(env_prefix='db_')
 url: str = Field()
 
db_config = DBConfig() # pyright: ignore[reportCallIssue]
# create database engine
engine = create_async_engine(db_config.url, echo=True)
# create session factory
# expire_on_commit prevents models from expiring after db.commit()
# Without this option, returning SQLAlchemy models 
# from request/route hanndler
# after calling `db.commit()` 
# would throw DetachedInstanceError under some circustances
# todo = Todo(todo='lorem ipsum')
# db.add(todo)
# db.commit()
# return todo
session_factory = async_sessionmaker(engine, expire_on_commit=False)
async def get_db():
 db = session_factory()
 try:
 yield db
 finally:
 await db.close()
class CreateTodoRequest(BaseModel):
 todo: str
# Data transfer object
class TodoBasic(BaseModel):
 id: int
 todo: str
 done: bool
 model_config = ConfigDict(
 from_attributes=True
 )
@app.get('/todos', response_model=list[TodoBasic])
async def get_todos(
 db : Annotated[AsyncSession, Depends(get_db)]
):
 todos = (await db.execute(select(Todo))).scalars().all()
 return todos
@app.post("/todos", response_model=TodoBasic)
async def create_todo(
 request: CreateTodoRequest,
 db: Annotated[AsyncSession, Depends(get_db)]
):
 todo = Todo(todo=request.todo)
 db.add(todo)
 await db.commit()
 return todo

source ./.venv/bin/activate
fastapi dev

Add static type checking

source ./.venv/bin/activate
uv add pyright --dev

source ./.venv/bin/activate
pyright app/main.py