Chapter 01 — Overview

🎯 What is SQLAlchemy?

A Python library for working with databases. It lets you interact with databases using Python code instead of writing SQL.

Think of it as

Pydantic

Define what API data looks like — validate the shape of requests and responses.

SQLAlchemy

Define what database tables look like — and interact with them from Python.

🔄 The Big Picture

Your FastAPI Application │ ├─ Pydantic (Request/Response Data) │ └─ Validates: {"name": "John", "email": "john@example.com"} │ ├─ SQLAlchemy (Database) │ └─ Stores/Retrieves: Row in users table │ └─ Database (PostgreSQL, MySQL, SQLite) └─ Actual data storage

📊 Pydantic vs SQLAlchemy

Aspect	Pydantic	SQLAlchemy
Purpose	Validate API data	Interact with databases
What it does	Checks incoming requests	Saves/retrieves data from DB
Data location	In memory (request/response)	Persistent storage (database)
Class base	`BaseModel` — API schema	`Base` — database model
Used for	API validation	CRUD operations

Key Distinction Pydantic guards the door (what comes in and goes out over HTTP). SQLAlchemy guards the vault (what is written to and read from the database).

Chapter 02 — Fundamentals

🛠️ How SQLAlchemy Works

The same job, done twice: once with raw SQL, once with the ORM.

Without SQLAlchemy (Raw SQL) ❌

python

import sqlite3

# Manual SQL queries 🤮
connection = sqlite3.connect('database.db')
cursor = connection.cursor()

# Create table
cursor.execute('''
    CREATE TABLE users (
        id INTEGER PRIMARY KEY,
        name TEXT NOT NULL,
        email TEXT NOT NULL
    )
''')

# Insert data
cursor.execute(
    'INSERT INTO users (name, email) VALUES (?, ?)',
    ('John', 'john@example.com')
)
connection.commit()

# Query data
cursor.execute('SELECT * FROM users WHERE id = 1')
user = cursor.fetchone()
print(user)  # (1, 'John', 'john@example.com') - raw tuple

connection.close()

With SQLAlchemy (ORM) ✨

python

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import declarative_base, Session

Base = declarative_base()

# Define table as Python class
class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String)
    email = Column(String)

# Create engine (database connection)
engine = create_engine('sqlite:///database.db')
Base.metadata.create_all(engine)

# Insert data
session = Session(engine)
new_user = User(name='John', email='john@example.com')
session.add(new_user)
session.commit()

# Query data
user = session.query(User).filter(User.id == 1).first()
print(user.name)  # 'John' - actual object property

💡 The difference that matters Raw SQL hands you a tuple you have to index by position. The ORM hands you a Python object with named attributes — readable, refactorable, and type-friendly.

Chapter 03 — Fundamentals

🏗️ SQLAlchemy Key Concepts

Three pieces you need before writing a single query: the model, the engine, and the session.

1. Define a Table as a Python Class

python

from sqlalchemy import create_engine, Column, Integer, String, DateTime
from sqlalchemy.orm import declarative_base
from datetime import datetime

Base = declarative_base()

class User(Base):  # ← Maps to 'users' table
    __tablename__ = 'users'  # ← Table name in database

    # Define columns
    id = Column(Integer, primary_key=True)  # Auto-increment ID
    name = Column(String)  # VARCHAR
    email = Column(String)  # VARCHAR
    created_at = Column(DateTime, default=datetime.utcnow)  # TIMESTAMP

# In database, this creates:
# CREATE TABLE users (
#   id INTEGER PRIMARY KEY AUTOINCREMENT,
#   name VARCHAR,
#   email VARCHAR,
#   created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
# )

2. Connect to the Database

python

from sqlalchemy import create_engine

# Create engine (connection pool)
engine = create_engine('sqlite:///database.db')  # SQLite
# engine = create_engine('postgresql://user:password@localhost/dbname')  # PostgreSQL
# engine = create_engine('mysql://user:password@localhost/dbname')  # MySQL

# Create all tables
Base.metadata.create_all(engine)

3. Create a Session (Active Connection)

python

from sqlalchemy.orm import Session

# Create session
session = Session(engine)

# Use it
# ... CRUD operations ...

# Close
session.close()

Engine vs Session The engine is the connection pool — created once for the whole app. The session is one unit of work — created per request, and closed when that request is done.

Chapter 04 — Working With Data

🔧 CRUD Operations

Create, Read, Update, Delete — the four operations every database app is built from.

CREATE (Insert)

python

from sqlalchemy.orm import Session

session = Session(engine)

# Create new user
new_user = User(
    name='John Doe',
    email='john@example.com'
)

# Add to database
session.add(new_user)
session.commit()  # Save changes
session.refresh(new_user)  # Get auto-generated ID

print(new_user.id)  # 1 (auto-generated)
session.close()

READ (Query)

python

session = Session(engine)

# Get all users
all_users = session.query(User).all()
for user in all_users:
    print(user.name, user.email)

# Get one user
user = session.query(User).filter(User.id == 1).first()
print(user.name)  # 'John Doe'

# Filter with conditions
users = session.query(User).filter(User.email.like('%@gmail.com')).all()

session.close()

UPDATE (Modify)

python

session = Session(engine)

# Get user
user = session.query(User).filter(User.id == 1).first()

# Modify
user.name = 'Jane Doe'
user.email = 'jane@example.com'

# Save
session.commit()

session.close()

DELETE (Remove)

python

session = Session(engine)

# Get user
user = session.query(User).filter(User.id == 1).first()

# Delete
session.delete(user)

# Save
session.commit()

session.close()

⚠️ Nothing is saved until commit() add(), attribute changes, and delete() only stage the work in the session. Without session.commit() the database never sees it.

Chapter 05 — Working With Data

💾 Column Types

SQLAlchemy column types map directly onto the native types of whichever database you connect to.

python

from sqlalchemy import (
    Column, Integer, String, Float, Boolean,
    DateTime, Date, Text, JSON, LargeBinary
)

class Product(Base):
    __tablename__ = 'products'

    id = Column(Integer, primary_key=True)
    name = Column(String(100))  # VARCHAR(100)
    price = Column(Float)  # DECIMAL/FLOAT
    description = Column(Text)  # LONGTEXT
    is_active = Column(Boolean, default=True)  # BOOLEAN
    created_at = Column(DateTime)  # TIMESTAMP
    attributes = Column(JSON)  # JSON
    image = Column(LargeBinary)  # BLOB

⚠️ Don't name a column metadata metadata is reserved by the declarative base — SQLAlchemy raises InvalidRequestError on that attribute name. Use a different name (like attributes above) or map it explicitly with Column('metadata', JSON).

SQLAlchemy Type	SQL Type	Use For
`Integer`	INTEGER	IDs, counts, foreign keys
`String(n)`	VARCHAR(n)	Short bounded text — names, emails
`Text`	TEXT / LONGTEXT	Long unbounded text — bodies, messages
`Float`	FLOAT / DECIMAL	Prices, scores, measurements
`Boolean`	BOOLEAN	Flags — is_active, is_deleted
`DateTime`	TIMESTAMP	created_at, updated_at
`JSON`	JSON / JSONB	Flexible nested payloads
`LargeBinary`	BLOB	Files, images, embeddings blobs

Chapter 06 — Working With Data

🔗 Relationships

The real power of SQLAlchemy — navigate between tables as if they were plain Python objects.

One-to-Many (User has Many Blogs)

python

from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String)

    # Relationship: User can have many blogs
    blogs = relationship('Blog', back_populates='author')

class Blog(Base):
    __tablename__ = 'blogs'

    id = Column(Integer, primary_key=True)
    title = Column(String)
    user_id = Column(Integer, ForeignKey('users.id'))  # Foreign key

    # Relationship: Blog belongs to one User
    author = relationship('User', back_populates='blogs')

# Usage
session = Session(engine)

# Get user with all their blogs
user = session.query(User).filter(User.id == 1).first()
print(user.blogs)  # [Blog(...), Blog(...), ...]

# Get blog with author info
blog = session.query(Blog).filter(Blog.id == 1).first()
print(blog.author.name)  # 'John Doe'

session.close()

Many-to-Many (Students have Many Courses)

python

from sqlalchemy import Table

# Association table
student_course = Table(
    'student_course',
    Base.metadata,
    Column('student_id', Integer, ForeignKey('students.id')),
    Column('course_id', Integer, ForeignKey('courses.id'))
)

class Student(Base):
    __tablename__ = 'students'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    courses = relationship('Course', secondary=student_course)

class Course(Base):
    __tablename__ = 'courses'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    students = relationship('Student', secondary=student_course)

💡 ForeignKey vs relationship ForeignKey is the database constraint — it enforces integrity in SQL. relationship() is the Python convenience — it gives you user.blogs and blog.author. You almost always want both.

Chapter 07 — Applied

🎯 FastAPI + SQLAlchemy

How the two libraries sit side by side in a real blog API — Pydantic at the edge, SQLAlchemy at the database.

python

from fastapi import FastAPI, Depends
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.orm import Session, declarative_base, relationship, sessionmaker
from pydantic import BaseModel

# SQLAlchemy setup
SQLALCHEMY_DATABASE_URL = "sqlite:///./blog.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(bind=engine)
Base = declarative_base()

# ========== DATABASE MODELS (SQLAlchemy) ==========
class User(Base):  # ← SQLAlchemy model
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    name = Column(String)
    email = Column(String, unique=True)
    password = Column(String)  # hashed

    blogs = relationship("Blog", back_populates="creator")

class Blog(Base):  # ← SQLAlchemy model
    __tablename__ = "blogs"

    id = Column(Integer, primary_key=True)
    title = Column(String)
    body = Column(String)
    user_id = Column(Integer, ForeignKey("users.id"))

    creator = relationship("User", back_populates="blogs")

# Create tables
Base.metadata.create_all(engine)

# ========== API MODELS (Pydantic) ==========
class BlogCreate(BaseModel):  # ← Pydantic model (REQUEST)
    title: str
    body: str

class UserResponse(BaseModel):  # ← Pydantic model (RESPONSE)
    id: int
    name: str
    email: str

    class Config:
        from_attributes = True

class BlogResponse(BaseModel):  # ← Pydantic model (RESPONSE)
    id: int
    title: str
    body: str
    creator: UserResponse

    class Config:
        from_attributes = True  # Convert SQLAlchemy to Pydantic

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

# ========== FASTAPI ENDPOINTS ==========
app = FastAPI()

@app.post("/blogs")
def create_blog(
    blog: BlogCreate,  # ← Pydantic validation
    db: Session = Depends(get_db)  # ← SQLAlchemy session
):
    # Create SQLAlchemy object
    db_blog = Blog(
        title=blog.title,
        body=blog.body,
        user_id=1
    )

    # Save to database
    db.add(db_blog)
    db.commit()
    db.refresh(db_blog)

    # Convert to Pydantic response
    return db_blog

@app.get("/blogs/{blog_id}")
def get_blog(
    blog_id: int,
    db: Session = Depends(get_db)  # ← SQLAlchemy session
):
    # Query database
    blog = db.query(Blog).filter(Blog.id == blog_id).first()

    if not blog:
        return {"error": "not found"}

    # FastAPI automatically converts to BlogResponse (Pydantic)
    return blog

Data Flow

1. Client sends JSON request ↓ 2. Pydantic validates (BlogCreate) ↓ 3. FastAPI calls endpoint ↓ 4. SQLAlchemy saves to database (Blog model) ↓ 5. Query returns SQLAlchemy object ↓ 6. Pydantic converts to JSON response (BlogResponse) ↓ 7. Client receives JSON response

from_attributes = True This is what lets a Pydantic response model read a SQLAlchemy object directly instead of a dict. In Pydantic v1 the same setting was called orm_mode = True.

Chapter 08 — Applied

🔍 Real-World Example: Contact Submission

One contact form, traced from the browser through Pydantic and SQLAlchemy into the database.

Database Model (SQLAlchemy)

python

from sqlalchemy import Column, Integer, String, DateTime, create_engine
from sqlalchemy.orm import declarative_base
from datetime import datetime

Base = declarative_base()

class ContactSubmission(Base):  # ← SQLAlchemy
    __tablename__ = "contact_submissions"

    id = Column(Integer, primary_key=True)
    name = Column(String(100), nullable=False)
    email = Column(String(100), nullable=False)
    message = Column(String(5000), nullable=False)
    phone = Column(String(20), nullable=True)
    subject = Column(String(50), nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow)
    ip_address = Column(String(50), nullable=False)

# Creates this in database:
# CREATE TABLE contact_submissions (
#   id INTEGER PRIMARY KEY AUTOINCREMENT,
#   name VARCHAR(100) NOT NULL,
#   email VARCHAR(100) NOT NULL,
#   message VARCHAR(5000) NOT NULL,
#   phone VARCHAR(20),
#   subject VARCHAR(50) NOT NULL,
#   created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
#   ip_address VARCHAR(50) NOT NULL
# )

API Model (Pydantic)

python

from typing import Optional
from datetime import datetime
from pydantic import BaseModel, EmailStr

class ContactFormRequest(BaseModel):  # ← Pydantic (REQUEST)
    name: str
    email: EmailStr
    message: str
    phone: Optional[str] = None
    subject: str = "general"

class ContactFormResponse(BaseModel):  # ← Pydantic (RESPONSE)
    id: int
    name: str
    email: str
    created_at: datetime

FastAPI Endpoint

python

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

app = FastAPI()

@app.post("/api/contact")
def submit_contact(
    form: ContactFormRequest,  # ← Pydantic validates
    request: Request,
    db: Session = Depends(get_db)  # ← SQLAlchemy session
):
    # 1. Create SQLAlchemy object (database row)
    db_submission = ContactSubmission(
        name=form.name,
        email=form.email,
        message=form.message,
        phone=form.phone,
        subject=form.subject,
        ip_address=request.client.host
    )

    # 2. Save to database
    db.add(db_submission)
    db.commit()
    db.refresh(db_submission)

    # 3. Send email async (don't wait)
    send_email_async(form.email, form.message)

    # 4. Return Pydantic response
    return {
        "success": True,
        "submission_id": db_submission.id
    }

What Happens

1. Client: POST /api/contact { "name": "John", "email": "john@example.com", "message": "..." } 2. Pydantic: Validates with ContactFormRequest ✅ name is string, email is valid, message is long enough 3. SQLAlchemy: Creates database row INSERT INTO contact_submissions (name, email, message, ...) VALUES ('John', 'john@example.com', '...', ...) 4. Response: Returns ContactFormResponse { "success": true, "submission_id": 1 }

Chapter 09 — Applied

🚀 Summary & Key Takeaways

Everything SQLAlchemy gives you, and exactly where it sits next to Pydantic and FastAPI.

📊 SQLAlchemy Features

Feature	Purpose	Example
Declarative Models	Define tables as Python classes	`class User(Base)`
ORM	Query with Python instead of SQL	`session.query(User).filter(...)`
Relationships	Link tables together	`relationship('Blog')`
Foreign Keys	Maintain data integrity	`ForeignKey('users.id')`
Constraints	Validate data in database	`unique=True`, `nullable=False`
Migrations	Change schema over time	Use Alembic with SQLAlchemy
Connection Pooling	Reuse connections	Built-in
Multi-DB Support	PostgreSQL, MySQL, SQLite, etc.	Just change connection string

🎓 Pydantic vs SQLAlchemy

Pydantic

class User(BaseModel) — API validation. Data is in transit (request/response). Used for validating API calls.

SQLAlchemy

class User(Base) — database table. Data is in persistent storage. Used for saving and retrieving data.

python

# PYDANTIC
class User(BaseModel):  # ← API validation
    name: str
    email: EmailStr

# Data: In transit (request/response)
# Used for: Validating API calls


# SQLALCHEMY
class User(Base):  # ← Database
    __tablename__ = 'users'
    name = Column(String)
    email = Column(String)

# Data: Persistent storage (database)
# Used for: Saving/retrieving data

🔄 Full Data Flow

┌──────────────────────────────────────────────────────────────┐ │ CLIENT (Browser) │ POST /api/contact │ { "name": "John", "email": "john@...", "message": "..." } └───────────────────────────┬──────────────────────────────────┘ ↓ ┌──────────────────────────────────────────────────────────────┐ │ PYDANTIC (Validation) │ class ContactFormRequest(BaseModel): │ name: str ... │ @app.post("/api/contact") │ def submit_contact(form: ContactFormRequest, db: Session): └───────────────────────────┬──────────────────────────────────┘ ↓ ┌──────────────────────────────────────────────────────────────┐ │ SQLALCHEMY (Database Save) │ class ContactSubmission(Base): │ __tablename__ = "contact_submissions" │ name = Column(String) │ email = Column(String) │ message = Column(String) └───────────────────────────┬──────────────────────────────────┘ ↓ ┌──────────────────────────────────────────────────────────────┐ │ DATABASE (SQLite/PostgreSQL/MySQL) │ INSERT INTO contact_submissions │ (name, email, message, created_at) │ VALUES ('John', 'john@...', '...', NOW()) └───────────────────────────┬──────────────────────────────────┘ ↓ ┌──────────────────────────────────────────────────────────────┐ │ PYDANTIC (Response) │ class ContactFormResponse(BaseModel): ... │ { "success": true, "submission_id": 1 } └──────────────────────────────────────────────────────────────┘

🚀 Key Takeaway

Tool	Layer	What It Does
Pydantic	API / Validation	Defines data shape for requests and responses
SQLAlchemy	Database / Persistence	Defines database tables and queries
FastAPI	Web Framework	Connects them together

Browser ↔ FastAPI (Web) ↔ Pydantic (Validation) ↔ SQLAlchemy (Database)

Where This Fits in the Video Course

Chapter 10: Pydantic schemas — define API data
Chapter 12: SQLAlchemy models — define database tables
Chapters 13–16: CRUD operations — save and retrieve data

🎓 You've Completed the Guide SQLAlchemy is what lets you persist data — keep it permanently in a database — instead of only validating it as it passes through your API.