Connecting to Relational Databases with SQLAlchemy

In previous chapters, we learned how to receive API requests and validate data using Pydantic. However, if we shut down the server, this data would disappear. To build a real-world business application, we must store data permanently in a database.

While you could directly write raw SQL statements (like INSERT INTO users...) to interact with the database, in modern backend development, we strongly recommend using ORM (Object-Relational Mapping) technology.

In the Python ecosystem, the most established, robust, and industry-standard ORM is SQLAlchemy.

1. What is ORM? Why Use It?

The core concept of ORM is to map database "tables" to Python "classes."

• Instead of writing SELECT * FROM users, you simply write db.query(User).all().
• Instead of writing INSERT INTO, you just create a User(name="Ken") object and db.add() it.

This approach offers three major benefits:

1. Prevents SQL Injection Attacks: ORM automatically filters dangerous characters at the underlying level.
2. Easy Database Switching: Test with SQLite today and switch to PostgreSQL tomorrow—no need to change any logic code, just the connection string.
3. Code Autocompletion: Typing user. in your editor will automatically suggest all fields, a feature not available when writing raw SQL.

2. Environment Setup and Database Connection

First, install SQLAlchemy and the PostgreSQL driver:

pip install sqlalchemy psycopg2-binary

Next, create a database.py file to configure the database connection:

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

# Replace with your own database URL (format: postgresql://username:password@host:port/database)
# Here, we use SQLite for lightweight local testing
SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"

# Create the engine (SQLite requires check_same_thread=False)
engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)

# Create a Session factory
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Create the base class for all Models
Base = declarative_base()

3. Defining SQLAlchemy Models

In models.py, we define the structure of the database tables. Note that this is different from the Pydantic models in the previous chapter. Pydantic validates incoming JSON, while SQLAlchemy defines the structure stored in the database.

from sqlalchemy import Boolean, Column, Integer, String
from database import Base

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean, default=True)

4. Dependency Injection Setup

FastAPI has a powerful feature called "Dependency Injection."
Every time an API is called, we need to open a database connection (Session) and close it when the API finishes. We don’t need to manually write this logic in every API.

Add this to main.py:

from fastapi import Depends, FastAPI
from sqlalchemy.orm import Session
import models
from database import SessionLocal, engine

# Let SQLAlchemy automatically create any missing tables
models.Base.metadata.create_all(bind=engine)

app = FastAPI()

# Define a Dependency to provide the Database Session
def get_db():
    db = SessionLocal()
    try:
        yield db  # Pass the db to the API route
    finally:
        db.close() # Ensure the connection is closed after the API finishes

5. Hands-on: Writing Data to the Database

Now, let’s combine Pydantic, SQLAlchemy, and Dependency Injection!

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session

# Import what we’ve written earlier
from database import get_db
import models
from pydantic import BaseModel

app = FastAPI()

# Define the Pydantic Schema for input
class UserCreate(BaseModel):
    email: str
    password: str

# Define the Pydantic Schema for output (hiding the password)
class UserResponse(BaseModel):
    id: int
    email: str
    is_active: bool

    class Config:
        from_attributes = True # Allows direct conversion from SQLAlchemy objects

@app.post("/users/", response_model=UserResponse)
def create_user(user: UserCreate, db: Session = Depends(get_db)):
    # 1. Check if the email is already registered
    db_user = db.query(models.User).filter(models.User.email == user.email).first()
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    
    # 2. Create the SQLAlchemy ORM object (in practice, passwords should be hashed!)
    fake_hashed_password = user.password + "notreallyhashed"
    new_user = models.User(email=user.email, hashed_password=fake_hashed_password)
    
    # 3. Write to the database
    db.add(new_user)
    db.commit()
    db.refresh(new_user) # Retrieve the auto-generated ID
    
    # 4. Return the response to the frontend
    return new_user

Summary

You’ve now learned the three pillars of backend development:

1. Routing: Handling HTTP requests.
2. Validation (Pydantic): Ensuring correct data formats.
3. Storage (SQLAlchemy): Safely interacting with the database.

However, in the example above, passwords are stored in plain text, and anyone can call this API. In the next chapter, we’ll learn the core of security: Password Hashing and JWT Authentication!

Database Models

SQLAlchemy models define the schema of your database tables as Python classes.

Declarative Base

from sqlalchemy import create_engine, Column, Integer, String, Boolean, DateTime, ForeignKey
from sqlalchemy.orm import declarative_base, relationship
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)
    hashed_password = Column(String(255), nullable=False)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, default=datetime.utcnow)
    
    # Relationship
    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), nullable=False)
    published = Column(Boolean, default=False)
    created_at = Column(DateTime, default=datetime.utcnow)
    author_id = Column(Integer, ForeignKey("users.id"))
    
    author = relationship("User", back_populates="posts")

Database Session Management

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

DATABASE_URL = "postgresql://user:password@localhost:5432/mydb"

engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Dependency for FastAPI routes
def get_db() -> Session:
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

CRUD Operations

Create

@app.post("/users/", response_model=UserResponse)
def create_user(user: UserCreate, db: Session = Depends(get_db)):
    # Check if username exists
    existing = db.query(User).filter(User.username == user.username).first()
    if existing:
        raise HTTPException(status_code=400, detail="Username already exists")
    
    # Create new user
    db_user = User(
        username=user.username,
        email=user.email,
        hashed_password=hash_password(user.password)
    )
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

Read

@app.get("/users/{user_id}", response_model=UserResponse)
def read_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

@app.get("/users/", response_model=list[UserResponse])
def list_users(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    users = db.query(User).offset(skip).limit(limit).all()
    return users

Update

@app.put("/users/{user_id}", response_model=UserResponse)
def update_user(user_id: int, user: UserUpdate, db: Session = Depends(get_db)):
    db_user = db.query(User).filter(User.id == user_id).first()
    if not db_user:
        raise HTTPException(status_code=404, detail="User not found")
    
    update_data = user.model_dump(exclude_unset=True)
    for key, value in update_data.items():
        setattr(db_user, key, value)
    
    db.commit()
    db.refresh(db_user)
    return db_user

Delete

@app.delete("/users/{user_id}", status_code=204)
def delete_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")
    
    db.delete(user)
    db.commit()
    return None

Alembic Migrations

# Install alembic
pip install alembic

# Initialize
cd app
alembic init alembic

# Generate migration
alembic revision --autogenerate -m "create users table"

# Apply migration
alembic upgrade head

# Check history
alembic history

Summary

SQLAlchemy provides a powerful ORM for database operations in FastAPI. Combined with Alembic for migrations, it gives you a complete database layer.

Key takeaways:

• Define models with declarative_base() — classes become tables |
• Relationships with relationship() and ForeignKey link tables |
• Session management: create engine, sessionmaker, dependency injection |
• CRUD: add(), query(), commit(), delete() |
• Pagination: offset() + limit() |
• Always hash passwords, never store plain text |
• Alembic handles schema migrations automatically |
• Dependency injection (Depends(get_db)) provides clean session handling |

What's Next: JWT Authentication

The next chapter covers password hashing and JWT authentication.