FastAPI + SQLAlchemy

Connect FastAPI to a database with SQLAlchemy ORM, sessions, and async support

Knowledge0%

Learn & Drill

Fluency0%

Drill & Speed

Retention0%

Mastery & Review

Confidence0%

All modes

Practice

Free tier: read the explanation here. Upgrade to Pro for Drills, Speed challenges & Mastery badges.

Upgrade

Knowledge

Fluency

Retention

Knowledge Debt detected

You can study this freely — but your score may plateau if these foundations have gaps. The Mastery badge requires them to be solid.

FastAPI Fundamentals0%

Explanation

SQLAlchemy is the standard Python ORM for working with relational databases. FastAPI's Depends() makes it clean to manage database sessions.

Setup:

python from sqlalchemy import create_engine, Column, Integer, String from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker, Session DATABASE_URL = "postgresql://user:pass@localhost/mydb" engine = create_engine(DATABASE_URL) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) Base = declarative_base()

Model definition:

python class User(Base): __tablename__ = "users" id = Column(Integer, primary_key=True, index=True) email = Column(String, unique=True, index=True, nullable=False) name = Column(String, nullable=False) is_active = Column(Boolean, default=True)

Session dependency — the core pattern:

python def get_db(): db = SessionLocal() try: yield db # route runs here with the db session finally: db.close() # always closes, even on error @app.get("/users/{user_id}") def get_user(user_id: int, db: Session = Depends(get_db)): user = db.query(User).filter(User.id == user_id).first() if not user: raise HTTPException(status_code=404, detail="User not found") return user

CRUD operations:

python # Create new_user = User(email="alice@example.com", name="Alice") db.add(new_user) db.commit() db.refresh(new_user) # load auto-generated id # Read users = db.query(User).filter(User.is_active == True).all() user = db.query(User).filter(User.email == email).first() # Update user.name = "Alice Smith" db.commit() # Delete db.delete(user) db.commit()

Pydantic schemas for request/response (separate from ORM models):

python from pydantic import BaseModel class UserCreate(BaseModel): email: str name: str class UserResponse(BaseModel): id: int email: str name: str is_active: bool class Config: from_attributes = True # allows reading from ORM model attributes @app.post("/users", response_model=UserResponse) def create_user(user_in: UserCreate, db: Session = Depends(get_db)): user = User(**user_in.model_dump()) db.add(user) db.commit() db.refresh(user) return user `` Always use separate Pydantic schemas for API input/output — never expose your ORM model directly (it can leak fields like passwords or internal IDs).

Examples

One-to-many relationship: User → Posts

relationships() lets you navigate between models in Python; joinedload() avoids the N+1 query problem

from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship

class User(Base):
    __tablename__ = "users"
    id    = Column(Integer, primary_key=True)
    email = Column(String, unique=True, nullable=False)
    posts = relationship("Post", back_populates="author")

class Post(Base):
    __tablename__ = "posts"
    id        = Column(Integer, primary_key=True)
    title     = Column(String, nullable=False)
    author_id = Column(Integer, ForeignKey("users.id"), nullable=False)
    author    = relationship("User", back_populates="posts")

# Query with join (avoids N+1)
@app.get("/users/{user_id}/posts")
def get_user_posts(user_id: int, db: Session = Depends(get_db)):
    user = db.query(User).options(
        joinedload(User.posts)  # single SQL JOIN, not N queries
    ).filter(User.id == user_id).first()
    return user.posts

Async SQLAlchemy with FastAPI

Async SQLAlchemy uses asyncpg driver and select() syntax instead of query(); needed when your routes are async

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/mydb"
engine = create_async_engine(DATABASE_URL)
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession,
                                  expire_on_commit=False)

async def get_db():
    async with AsyncSessionLocal() as db:
        yield db

@app.get("/users/{user_id}")
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(User).where(User.id == user_id))
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(404, "Not found")
    return user

How well did you understand this?