13 Sqlalchemy Integration

title: SQLAlchemy Integration in FastAPI REST APIs weight: 23 date: 2026-06-28 lastmod: 2026-06-28 description: Integrate SQLAlchemy with FastAPI including model definitions, relationships, database sessions, CRUD operations, migrations with Alembic, and query optimization. tags: [api-development, fastapi]

SQLAlchemy integration with FastAPI uses declarative models, dependency-injected sessions, and Pydantic schemas for ORM-to-API serialization, with Alembic handling database migrations.

```python
# app/models.py
from sqlalchemy import Column, Integer, String, Float, ForeignKey, DateTime
from sqlalchemy.orm import relationship, declarative_base
from datetime import datetime

Base = declarative_base()

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    username = Column(String(50), unique=True, nullable=False, index=True)
    email = Column(String(100), unique=True, nullable=False)
    password = Column(String(255), nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow)

    posts = relationship("Post", back_populates="author")

class Post(Base):
    __tablename__ = "posts"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String(200), nullable=False)
    content = Column(String(5000))
    author_id = Column(Integer, ForeignKey("users.id"), nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow)

    author = relationship("User", back_populates="posts")

# app/database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "sqlite:///./blog.db"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# app/routes/users.py
from fastapi import Depends, HTTPException
from sqlalchemy.orm import Session
from app import models, schemas

@router.get("/users/{user_id}", response_model=schemas.UserResponse)
def get_user(user_id: int, db: Session = Depends(get_db)):
    user = db.query(models.User).filter(models.User.id == user_id).first()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

What's Next

Now learn about async database in Building REST APIs with FastAPI.