SQLAlchemy 2.0 & Alembic Migrations: A Complete Beginner's Guide🗄️

🗄️ SQLAlchemy 2.0 & Alembic Migrations: Let's Learn from ZERO!

Once you understand SQLAlchemy 2.0 + Alembic, you understand how a real Python backend talks to PostgreSQL, stores data, retrieves records, updates them, and safely changes database structure as the app grows. 🚀

No assumptions here. You don't need to know ORMs, sessions, migrations, or database internals. We'll use examples from an Instagram-like SocialFeed app throughout.

---

1. 🖼️ The Big Picture

SocialFeed stores users, posts, comments, likes, follows, notifications and sessions. All of it must live permanently in PostgreSQL. But your backend is written in Python, while PostgreSQL understands SQL. How do they talk?

🐍 Python / FastAPI     → your application code
        ▼
🧱 SQLAlchemy 2.0       → converts Python database operations into SQL
        ▼
🐘 PostgreSQL           → stores and retrieves your data

🔄 Alembic              → manages changes to your database structure over time

• SQLAlchemy helps Python communicate with the database
• Alembic helps you modify the database structure safely over time

---

🧱 PART 1: SQLAlchemy Fundamentals

2. 🤔 What exactly is SQLAlchemy?

📌 SQLAlchemy is a Python library for interacting with relational databases.

It lets you create tables, insert, retrieve, update and delete records, manage connections and transactions, and work with relationships.

Raw SQL to fetch user 10:

SELECT * FROM users WHERE id = 10;

With the SQLAlchemy ORM:

user = session.get(User, 10)

• Raw SQL: you write SQL statements yourself
• SQLAlchemy ORM: you work with Python classes and objects

⚠️ SQLAlchemy doesn't eliminate SQL. It helps you work with SQL through Python.

3. 🪄 What is an ORM?

ORM = Object Relational Mapping.

• Object: Python objects
• Relational: relational databases with tables and relationships
• Mapping: connecting Python objects to database tables

The users table:

id	username	email
1	deepika	[e-mail]
2	rahul	[e-mail]
3	priya	[e-mail]

A Python class representing it:

class User:
    def __init__(self, id, username, email):
        self.id = id
        self.username = username
        self.email = email


user = User(id=1, username="deepika", email="[e-mail]")

Python class  User   (id, username, email)
        ⇅  ORM mapping
Database table  users   (id, username, email)

One User object = one row in the users table

SQLAlchemy ORM makes this mapping automatic. Instead of writing SQL for every operation you write Python:

user.username = "new_username"

SQLAlchemy detects the change and generates an SQL UPDATE when the transaction is flushed.

4. ⚔️ SQLAlchemy Core vs SQLAlchemy ORM

🎯 A very important interview and practical concept.

SQLAlchemy offers two ways of working with a database. Both talk to the same database. The difference is how you express operations.

🅰️ SQLAlchemy Core

A SQL expression toolkit. You work with tables, columns, SQL expressions and statements, with no Python object per row.

from sqlalchemy import select

statement = select(users_table)
result = session.execute(statement)
users = result.all()

• select() creates a SELECT statement
• users_table represents the database table
• execute() runs it
• all() retrieves all matching rows

Inserting with Core:

from sqlalchemy import insert

statement = insert(users_table).values(
    username="deepika",
    email="[e-mail]",
)

session.execute(statement)
session.commit()

🅱️ SQLAlchemy ORM

You work with Python classes representing tables:

statement = select(User)
result = session.scalars(statement)
users = result.all()

for user in users:
    print(user.username)
    print(user.email)

User is an ORM model, and SQLAlchemy returns User objects instead of raw column-oriented rows.

📊 Core vs ORM

Feature	Core	ORM
Main approach	Tables and SQL expressions	Python classes and objects
Uses	Table definitions	ORM mapped classes
Returns	Rows / column results	ORM objects
Abstraction	Lower-level	Higher-level
Relationships	Explicit SQL joins	ORM relationships supported
Best for	SQL-heavy operations	Application development

💡 Neither is universally better, and you can combine them. A FastAPI app might use the ORM for normal user and post operations, and Core for bulk inserts or complex reports. Learn the ORM thoroughly first, while understanding Core enough to recognize its SQL expression API.

---

🏗️ PART 2: Declarative Models

5. 📐 What are Declarative Models?

In SQL you'd write:

CREATE TABLE users (
    id INTEGER PRIMARY KEY,
    username VARCHAR(50) NOT NULL,
    email VARCHAR(100) NOT NULL
);

SQLAlchemy lets you define the same table with a Python class. That's a declarative model.

📌 A declarative model is a Python class that describes a database table, including its columns, data types, constraints and relationships.

Step 1: install

pip install sqlalchemy

Step 2: the imports

from sqlalchemy import String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

Import	Purpose
String	Defines a string database column
DeclarativeBase	Base class for declarative models
Mapped	Type annotation for mapped attributes
mapped_column	Defines a mapped database column

Step 3: create a Base class

class Base(DeclarativeBase):
    pass

Base is the common parent class for your models. Every model inherits from it:

class User(Base):
    ...


class Post(Base):
    ...


class Comment(Base):
    ...

Because they inherit from Base, SQLAlchemy knows they're ORM models and tracks their table definitions in shared metadata.

Step 4: create the User model

from sqlalchemy import String
from sqlalchemy.orm import Mapped, mapped_column


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)

    username: Mapped[str] = mapped_column(
        String(50),
        nullable=False,
    )

    email: Mapped[str] = mapped_column(
        String(100),
        nullable=False,
        unique=True,
    )

Line by line:

1. class User(Base): a Python class named User that inherits from Base
2. __tablename__ = "users": the actual database table name (class User ↔ table users)
3. id: Mapped[int]: id is a mapped Python attribute of type int
4. mapped_column(primary_key=True): defines the column and makes it the primary key, which uniquely identifies each row. No two rows can share one
5. nullable=False: the database doesn't allow NULL here
6. unique=True: duplicate values aren't allowed (two users can't share an email)

🎨 Common column types

SQLAlchemy type	Python type	Database meaning
Integer	int	Whole number
String(100)	str	Limited-length text
Text	str	Longer text
Boolean	bool	True / False
DateTime	datetime	Date and time
Float	float	Floating-point number
Numeric	Decimal	Precise numeric value
JSON	dict / list etc.	JSON data

📝 Python annotations and database types are related, but not exactly the same thing.

6. 🔍 Mapped and mapped_column in Depth

You'll see both everywhere in SQLAlchemy 2.0. Why do both exist?

username: Mapped[str] = mapped_column(
    String(50),
    nullable=False,
)

Mapped[str] is the Python-side typing information. It tells developers what type to expect, tells IDEs what to suggest, lets static type checkers check, and tells SQLAlchemy the attribute is mapped.

mapped_column() describes the actual database column: its type, primary key, foreign key, nullability, unique constraints and defaults.

🧠 Mapped[str] = Python-side declaration. mapped_column() = database column configuration.

❓ Optional columns

A user's bio isn't compulsory:

from typing import Optional

bio: Mapped[Optional[str]] = mapped_column(
    String(300),
    nullable=True,
)

In modern Python you can write:

bio: Mapped[str | None] = mapped_column(
    String(300),
    nullable=True,
)

The attribute can hold a string or None, and the database column allows NULL.

🎚️ Default values

is_active: Mapped[bool] = mapped_column(
    default=True,
    nullable=False,
)

You can also configure a database-side default:

server_default="true"

Option	Who applies it?
default	SQLAlchemy, when constructing an INSERT without an explicit value
server_default	The database itself, when an INSERT omits that column

---

🔗 PART 3: Relationships Between Models

7. 🗝️ What are Foreign Keys?

A user creates multiple posts.

users:

id	username
1	deepika
2	rahul

posts:

id	content	user_id
101	Hello world	1
102	Beautiful day	1
103	Learning SQLAlchemy	2

user_id tells us who created the post. It's a foreign key referencing the id column of users:

from sqlalchemy import ForeignKey

user_id: Mapped[int] = mapped_column(
    ForeignKey("users.id"),
    nullable=False,
)

This creates a database-level reference and helps maintain referential integrity. For example, you normally can't insert a post with user_id=999 if no such user exists (assuming foreign key enforcement is on).

8. 🔄 What is relationship()?

A foreign key connects tables. But what if you want convenient access to related Python objects? That's relationship():

from sqlalchemy import ForeignKey, String
from sqlalchemy.orm import Mapped, mapped_column, relationship


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(String(50))

    posts: Mapped[list["Post"]] = relationship(
        back_populates="author",
    )


class Post(Base):
    __tablename__ = "posts"

    id: Mapped[int] = mapped_column(primary_key=True)
    content: Mapped[str] = mapped_column(String(500))

    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))

    author: Mapped["User"] = relationship(
        back_populates="posts",
    )

• User.posts → a user can access their posts
• Post.author → a post can access its author

user = session.get(User, 1)

print(user.username)

for post in user.posts:
    print(post.content)

post = session.get(Post, 101)
print(post.author.username)

One-to-many relationship

User (id = 1)  ──▶  Post 101
                ──▶  Post 102
                ──▶  Post 103

One User → many Posts. Each Post → one User.

🧠 ForeignKey defines the database-level relationship. relationship() defines the convenient Python-level object relationship. You usually need both.

9. 🧩 Other Relationship Types

One-to-one: one user has one profile.

User 1 ───── Profile 1

One-to-many: one user has many posts.

User 1 ───── Post 1
        ├─── Post 2
        └─── Post 3

Many-to-many: many users like many posts.

User 1 ───┬── Post 1
          └── Post 2

User 2 ───┬── Post 1
          └── Post 3

Many-to-many relationships generally use an association table. For SocialFeed, a likes table is a natural example:

user_id	post_id
1	101
2	101
1	102

---

🔌 PART 4: Engine and Database Connections

10. ⚙️ What is a SQLAlchemy Engine?

Before Python can execute queries, it needs a way to connect to PostgreSQL. SQLAlchemy uses an Engine, the central database connection manager. It knows which database to connect to, which driver to use, how to create connections, and how to manage the connection pool.

from sqlalchemy import create_engine

engine = create_engine(
    "postgresql+psycopg://user:password@localhost/mydb"
)

• postgresql → database
• psycopg → driver
• user / password → credentials
• localhost → host
• mydb → database name

📝 Creating the engine doesn't necessarily open a connection right away. Connections are generally obtained when needed.

🏊 What is a connection pool?

100 API requests each need the database. Would you create a brand-new connection every time? That's inefficient. SQLAlchemy uses connection pooling: a collection of reusable connections.

FastAPI requests:  A · B · C
        ▼
SQLAlchemy Engine  → manages reusable connections
        ▼
Connection 1 · Connection 2 · Connection 3
        ▼
🐘 PostgreSQL

A connection is checked out when needed and returned to the pool when its transaction or usage ends. Pool settings include pool_size, max_overflow, pool_timeout and pool_recycle. You don't need to tune them yet. First understand why the pool exists.

---

🧑‍💻 PART 5: Session and AsyncSession

🎯 One of the MOST IMPORTANT sections!

11. 🗂️ What is a Session?

A user calls POST /posts. Your backend must connect to the database, insert the post, commit the transaction, and release resources.

📌 A Session is a workspace for interacting with the database through the ORM. It manages ORM objects, database operations, pending changes, transactions and the identity map.

🅰️ Creating a Session

from sqlalchemy.orm import Session, sessionmaker

SessionLocal = sessionmaker(
    bind=engine,
    class_=Session,
)

session = SessionLocal()

sessionmaker is a factory for creating sessions.

🅱️ Adding a record

user = User(
    username="deepika",
    email="[e-mail]",
)

session.add(user)
session.commit()

1. Create a Python object. It exists only in Python memory and isn't necessarily in PostgreSQL yet
2. session.add(user). SQLAlchemy tracks the object as pending
3. session.commit(). SQLAlchemy flushes pending changes and commits the transaction. Now the database stores the record

🅲 What is flush()?

🚨 A VERY important distinction!

session.add(user)
session.flush()

SQLAlchemy sends the INSERT to the database within the current transaction, but does NOT commit:

• The SQL has been executed
• The transaction remains uncommitted
• Other transactions generally can't see the change
• You can still roll back

session.add(user)
session.flush()

print(user.id)     # the generated ID is available!

session.rollback()

Why is flush() useful? Sometimes you need the generated ID before committing. For example, insert a user, get their ID, then create another record that references it, all in one transaction.

🅳 commit()

Commits the current transaction and makes changes permanent (subject to database durability).

🅴 rollback()

Something went wrong mid-way and you don't want a half-finished transaction:

session.rollback()

🅵 close()

session.close()

Closes the Session and releases its transactional resources. It doesn't destroy your connection pool.

📋 Remember these five methods

Method	Meaning
add()	Track an object for persistence
flush()	Send pending SQL changes without committing
commit()	Commit the transaction
rollback()	Undo uncommitted transaction changes
close()	Close the Session and release resources

12. 🪪 What is the Identity Map?

A Session maintains an identity map: within a Session, the same database identity is represented by the same Python object.

user1 = session.get(User, 1)
user2 = session.get(User, 1)

print(user1 is user2)   # typically True

If the object is still in the identity map and hasn't expired, SQLAlchemy can return the same Python object. This keeps things consistent: you don't get two separate in-memory objects for the same row in the same Session.

📝 An identity map is not a general-purpose query cache. A new query with different filters may still execute SQL.

13. ⚡ What is AsyncSession?

FastAPI supports async endpoints:

@app.get("/users")
async def get_users():
    ...

But using a synchronous database driver inside an async endpoint can block the event loop. SQLAlchemy therefore provides AsyncEngine, AsyncSession and asynchronous execution.

Session	AsyncSession
Synchronous	Asynchronous
Session	AsyncSession
create_engine()	create_async_engine()
session.execute()	await session.execute()
Blocking DB I/O	Non-blocking DB I/O (with an async driver)

🛠️ Creating an AsyncEngine

pip install sqlalchemy asyncpg

asyncpg is an asynchronous PostgreSQL driver.

from sqlalchemy.ext.asyncio import create_async_engine

DATABASE_URL = "postgresql+asyncpg://user:password@localhost/mydb"

engine = create_async_engine(
    DATABASE_URL,
    echo=True,
)

• postgresql → database type
• asyncpg → async driver
• echo=True → prints generated SQL to the console (great for development; don't leave sensitive SQL parameters exposed in production logs)

🏭 Creating an AsyncSession factory

from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

AsyncSessionLocal = async_sessionmaker(
    bind=engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

expire_on_commit=False means ORM objects keep their already-loaded attribute values after a commit instead of automatically expiring. This matters in async code because accessing an expired attribute could require an implicit database load. It does not automatically load relationships or guarantee every attribute is available.

▶️ Using it

async with AsyncSessionLocal() as session:
    result = await session.execute(select(User))
    users = result.scalars().all()

Why await? Executing a query requires I/O, which takes time. await lets the coroutine suspend while waiting so other tasks can run on the event loop.

14. 🚪 What is async with?

async with AsyncSessionLocal() as session:
    ...

This is an asynchronous context manager. It manages the Session's lifecycle, and the Session is closed when the block finishes. It's like:

session = AsyncSessionLocal()

try:
    # Perform database operations
    pass
finally:
    await session.close()

⚠️ Closing a session does not automatically commit your changes. Commit explicitly when appropriate.

🧪 A complete AsyncSession example

async def create_user():
    async with AsyncSessionLocal() as session:
        user = User(
            username="deepika",
            email="[e-mail]",
        )

        session.add(user)

        await session.commit()
        await session.refresh(user)

        return user

refresh() reloads an ORM object's attributes from the database. The database may have generated values such as auto-generated IDs, server-side timestamps or database defaults, and refreshing retrieves them.

15. ⏳ What is expire_on_commit=False?

await session.commit()

print(user.username)

By default, SQLAlchemy expires ORM object attributes after a commit. Expiration means SQLAlchemy may need to reload them from the database when you access them.

• In sync code this reload happens transparently
• In async code implicit database I/O is a problem, because it needs to be awaited

So many async applications configure expire_on_commit=False. But remember: it doesn't make unloaded relationships or deferred columns magically available.

16. 🚨 Don't Share a Session Across Concurrent Tasks

A common beginner mistake is sharing one Session across multiple requests or tasks. Don't!

A SQLAlchemy Session is mutable and stateful. It isn't safe to share concurrently between threads. Likewise, an AsyncSession isn't safe to share across concurrent asyncio tasks:

# ❌ Avoid: both functions use the SAME AsyncSession concurrently
await asyncio.gather(
    do_query(session),
    do_another_query(session),
)

Use separate AsyncSession instances for independent concurrent tasks. This is especially important in real FastAPI apps.

---

🏗️ PART 6: A Real Async Database Setup

17. 🗂️ The project structure

backend/
│
├── app/
│   ├── api/
│   ├── models/
│   │   ├── user.py
│   │   └── post.py
│   │
│   ├── repositories/
│   │   ├── user_repository.py
│   │   └── post_repository.py
│   │
│   ├── services/
│   ├── schemas/
│   ├── database.py
│   └── main.py
│
├── alembic/
├── alembic.ini
└── tests/

database.py

from sqlalchemy.ext.asyncio import (
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)

DATABASE_URL = "postgresql+asyncpg://user:password@localhost/mydb"

engine = create_async_engine(
    DATABASE_URL,
    echo=True,
)

AsyncSessionLocal = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

Now a dependency:

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

yield provides the Session to FastAPI and keeps it open while the request uses it. When the request's dependency scope finishes, the context manager closes the Session.

⚡ Using it in FastAPI

from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession


@app.get("/users")
async def get_users(db: AsyncSession = Depends(get_db)):
    ...

• db → the variable holding the session
• AsyncSession → the expected type
• Depends(get_db) → FastAPI calls the dependency and injects the resulting session

That's dependency injection. You don't manually create a session in every endpoint.

🏛️ Architecture point: one request generally gets its own Session. The Session should not be a global shared object.

---

🔄 PART 7: Alembic Migrations

18. 🤔 What is Alembic?

📌 Alembic is a database migration tool designed to work with SQLAlchemy. Its purpose is to manage changes to your database schema over time.

A schema describes the structure of your database: tables, columns, data types, primary keys, foreign keys, constraints and indexes.

Version 1: users(id, username)
        ▼  new feature added
Version 2: users(id, username, email)

You could run this by hand:

ALTER TABLE users
ADD COLUMN email VARCHAR(100);

But after hundreds of changes over several months, how do you track:

• Which changes have been made?
• Which have been applied to production?
• Which are pending?
• How do you reverse one?
• How do your teammates apply the same changes?

Alembic solves this by keeping a history of schema changes through migration files.

19. 📜 What is a migration?

📌 A migration is a Python file containing instructions to change your database schema.

For example, to add email to users, Alembic creates a migration with:

1. Upgrade: add the email column
2. Downgrade: remove the email column

🌿 Migrations are like Git commits

Git	Alembic
Commit	Migration revision
Commit history	Migration history
Checkout / revision navigation	Migration version tracking
Apply code changes	Apply schema changes
Revert a commit	Write/run a downgrade where appropriate

⚠️ Migrations can modify real production data and structures. Downgrades aren't always perfectly reversible.

20. 📥 Installing Alembic

pip install alembic
alembic init alembic

backend/
│
├── app/
│   ├── models/
│   ├── database.py
│   └── main.py
│
├── alembic/
│   ├── versions/
│   ├── env.py
│   └── script.py.mako
│
└── alembic.ini

File / folder	Purpose
alembic.ini	Alembic configuration
alembic/env.py	Environment and migration configuration
alembic/versions/	Stores migration files
script.py.mako	Template used to generate migration files

env.py is one of the most important Alembic files. It tells Alembic which database to connect to, where the SQLAlchemy model metadata is, whether migrations run synchronously or asynchronously, and how to compare the existing schema with your model definitions.

21. 📄 What is alembic.ini?

Alembic's main configuration file, containing the migration script location, database connection configuration and logging settings:

[alembic]

script_location = %(here)s/alembic

prepend_sys_path = .

sqlalchemy.url = postgresql://user:password@localhost/mydb

But our app uses postgresql+asyncpg://... and is asynchronous. We can configure the migration environment to use an async SQLAlchemy engine.

🔐 In a real app, don't hardcode credentials in a committed config file. Read them from environment variables or your settings module:

from app.config import settings

database_url = settings.DATABASE_URL

22. 🎯 What is target_metadata?

🚨 A VERY important concept!

Remember our Base class?

class Base(DeclarativeBase):
    pass

SQLAlchemy stores information about every model that inherits from it in Base.metadata, which holds the table definitions SQLAlchemy knows about:

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)

Alembic needs to compare two things:

1. Your current SQLAlchemy model definitions
2. Your actual database schema

The model metadata provides the first side of that comparison. Inside alembic/env.py:

target_metadata = Base.metadata

⚠️ Alembic must also be able to import your model classes so they're registered in Base.metadata. If you forget to import a model, its table may be missing from the metadata, and autogenerate might not detect it.

---

🧭 PART 8: The Alembic Migration Workflow

23. 🤖 What is Autogenerate?

You have a User model, and the table already exists in your database:

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(String(50))

Now you add an email column to the model:

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(String(50))

    email: Mapped[str] = mapped_column(
        String(100),
        nullable=False,
    )

🚨 The key point: PostgreSQL doesn't change just because you edited Python code! SQLAlchemy models describe the desired structure. Alembic migrations change the actual database structure.

Now run:

alembic revision --autogenerate -m "add email to users"

• alembic → the command-line tool
• revision → create a new migration revision
• --autogenerate → compare database schema with model metadata and propose changes
• -m → a descriptive message

Alembic generates a file in alembic/versions/:

"""add email to users"""

import sqlalchemy as sa
from alembic import op

revision = "abc123"
down_revision = "previous_revision"


def upgrade():
    op.add_column(
        "users",
        sa.Column(
            "email",
            sa.String(length=100),
            nullable=False,
        ),
    )


def downgrade():
    op.drop_column("users", "email")

• revision is the migration's unique ID
• down_revision identifies the migration that came before it, which establishes the order:

Migration A
    ▼
Migration B
    ▼
Migration C      (C's down_revision points to B)

• upgrade() contains the instructions to move the database forward (add the column)
• downgrade() contains instructions to reverse it (drop the column)

Old schema: users(id, username)
      │  upgrade()  ──▶  move forward
      │  downgrade() ◀──  move backward
New schema: users(id, username, email)

⚠️ Adding a nullable=False column to a table that already has rows needs a default (or a multi-step migration), or PostgreSQL will reject it. Always review the generated file.

24. ⬆️ alembic upgrade head

Creating a migration file doesn't apply it. You must run:

alembic upgrade head

• upgrade → apply migrations in the forward direction
• head → move to the latest available migration revision

If you have migrations A, B and C and your database is at A, then alembic upgrade head applies B and C.

🧠 revision --autogenerate creates a migration file. upgrade head applies it to the database. They're two separate steps.

25. ⬇️ alembic downgrade

Applied a migration by accident?

alembic downgrade -1

This moves back one revision using the current migration's downgrade() function.

Before:  A → B → C
                 ^ current DB

After:   A → B → C
             ^ current DB

alembic downgrade abc123    # to a specific revision
alembic downgrade base      # back to the original empty base

🚨 A downgrade may delete columns, tables or data. Never blindly downgrade a production database.

26. 📋 Alembic Command Cheat Sheet

Command	Purpose
alembic init alembic	Initialize Alembic
alembic revision -m "message"	Create an empty migration
alembic revision --autogenerate -m "message"	Generate a migration from detected schema differences
alembic upgrade head	Apply all pending migrations
alembic upgrade +1	Apply the next migration
alembic downgrade -1	Reverse one migration
alembic downgrade base	Reverse all migrations
alembic current	Show the current database revision
alembic history	Show migration history
alembic heads	Show migration head revisions
alembic stamp head	Mark the database at head without executing migrations

🏷️ What is stamp?

Say you already have a database with tables, but Alembic isn't tracking its migration history. You can tell Alembic the database should be considered to be at a particular revision:

alembic stamp head

This updates Alembic's version tracking without running the migration scripts. It does NOT create missing tables or modify columns. Only use it when you've verified the actual schema matches the revision you're stamping.

27. 📒 What is alembic_version?

How does Alembic know which migrations have already been applied? It uses a special table called alembic_version. For a simple single-head history it holds the current revision ID:

version_num
abc123

When you run alembic upgrade head, Alembic:

1. Checks the current revision
2. Determines which migrations are pending
3. Executes the required upgrade functions
4. Updates the version tracking table

That prevents applied migrations from re-running. (With multiple heads, the tracking can involve multiple revision IDs.)

28. 🔬 How does Autogenerate actually work?

Your database has users(id, username), but your Python model defines users(id, username, email). Alembic performs a schema comparison:

SQLAlchemy metadata:  id, username, email
Actual database:      id, username
        ▼
Schema comparison
        ▼
Detected difference: new column `email VARCHAR(100)`
        ▼
Alembic proposes a migration operation

⚠️ Autogenerate is not perfect. It detects many common differences but can't understand every database change or your business intention:

• Some table renames appear as a drop + create
• Some column renames appear as drop + add
• Data transformations need manual logic
• Some constraint and database-specific changes need manual handling

Always inspect the generated migration before applying it.

---

⚡ PART 9: Async Alembic Configuration

29. 🤝 How does Alembic work with asyncpg?

Your FastAPI app uses postgresql+asyncpg://..., but Alembic's migration environment must be configured appropriately. A common setup uses async_engine_from_config and runs Alembic's synchronous migration operations through connection.run_sync().

Inside alembic/env.py:

import asyncio

from alembic import context
from sqlalchemy import pool
from sqlalchemy.ext.asyncio import async_engine_from_config

from app.database import Base
from app.models import post, user  # noqa: F401  (registers models)

config = context.config

target_metadata = Base.metadata

Why import user and post? So SQLAlchemy registers these models in Base.metadata.

📝 This example imports Base from app.database. Import it from wherever you defined your Base class.

The synchronous migration operation:

def do_run_migrations(connection):
    context.configure(
        connection=connection,
        target_metadata=target_metadata,
    )

    with context.begin_transaction():
        context.run_migrations()

Getting the asynchronous connection:

async def run_async_migrations():
    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()

Finally:

def run_migrations_online():
    asyncio.run(run_async_migrations())


run_migrations_online()

Why run_sync() if we're using asyncpg? Alembic's migration context and operations are designed around synchronous-style operations. run_sync() is a bridge: it lets Alembic run its synchronous migration functions on the underlying async connection through SQLAlchemy's async integration. It doesn't mean your app switched to a synchronous PostgreSQL driver.

📝 This is a simplified online-migration setup. Production Alembic environments may add offline migrations, logging, URL loading and transaction handling.

🚫 What is pool.NullPool?

NullPool doesn't keep a pool of reusable connections. For a migration command that runs and exits, that's sensible: it opens a connection when needed and closes it when finished. Your FastAPI app can still use its normal connection pool.

---

🏛️ PART 10: The Repository Pattern

30. 🤔 What is the Repository Pattern?

A user clicks Login in SocialFeed and your backend must find them in PostgreSQL. Where should that query live? You could write it directly in the route:

@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()

    return user

That works. But with 50 endpoints, many needing to retrieve users, you'd repeat it, which causes:

• 🔁 Duplicate code
• 😩 Difficult maintenance
• 🧶 Database logic mixed with API logic
• 🧪 Difficult testing
• ♻️ Harder code reuse

📌 A repository is a layer responsible for communicating with the database and performing data-access operations. Think of it as the database specialist in your backend.

❌ Without the Repository Pattern
FastAPI Route: SQL queries + business logic + response handling, all mixed together
        ▼
🐘 PostgreSQL

✅ With the Repository Pattern
🛣️ FastAPI Route     → receives the HTTP request
        ▼
🧠 Service Layer     → business rules and application logic
        ▼
📚 Repository Layer  → database queries and data access
        ▼
🐘 PostgreSQL        → stores data

🛣️ Layer 1: API / Router

Receives HTTP requests, validates path/query parameters, obtains dependencies, calls service functions and returns HTTP responses. It shouldn't contain complicated database logic.

🧠 Layer 2: Service

Contains business logic: check whether a user exists, check permissions, enforce application rules, coordinate multiple repositories, decide what happens next.

async def get_user_profile(user_id: int, db):
    user = await user_repository.get_by_id(db, user_id)

    if user is None:
        raise UserNotFoundError()

    return user

Deciding how the app behaves when a user doesn't exist is a business/application-level responsibility.

📚 Layer 3: Repository

Handles database access:

async def get_by_id(db: AsyncSession, user_id: int):
    result = await db.execute(
        select(User).where(User.id == user_id)
    )

    return result.scalar_one_or_none()

It only retrieves a user. It doesn't decide the HTTP status code, how to authenticate someone, or what message to show.

31. 🧵 A Complete Repository Example

app/
│
├── api/
│   └── user_routes.py
│
├── services/
│   └── user_service.py
│
├── repositories/
│   └── user_repository.py
│
├── models/
│   └── user.py
│
└── database.py

Step 1: repositories/user_repository.py

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.models.user import User


async def get_by_id(db: AsyncSession, user_id: int):
    result = await db.execute(
        select(User).where(User.id == user_id)
    )

    return result.scalar_one_or_none()


async def get_by_email(db: AsyncSession, email: str):
    result = await db.execute(
        select(User).where(User.email == email)
    )

    return result.scalar_one_or_none()


async def create(db: AsyncSession, username: str, email: str):
    user = User(
        username=username,
        email=email,
    )

    db.add(user)

    await db.flush()

    return user

• get_by_id() retrieves a user by ID
• get_by_email() retrieves a user by email
• create() creates a new User object and adds it to the Session

Notice await db.flush(). Why not commit here? Because the repository doesn't necessarily own the transaction. The service (or a higher-level unit of work) may need several operations together: create a user, create their profile, create their default settings. They might need to succeed or fail together.

Step 2: services/user_service.py

from app.repositories import user_repository


async def get_user_profile(db, user_id: int):
    user = await user_repository.get_by_id(db, user_id)

    if user is None:
        return None

    return user


async def register_user(db, username: str, email: str):
    existing_user = await user_repository.get_by_email(db, email)

    if existing_user is not None:
        raise ValueError("Email already registered")

    user = await user_repository.create(db, username, email)

    await db.commit()

    return user

The service checks whether the email exists, decides whether registration proceeds, calls the repository, and commits.

⚠️ This is a teaching example. In production, the database unique constraint must still enforce email uniqueness, because two simultaneous registrations can race. You'd catch the resulting IntegrityError, roll back, and translate it into an appropriate application error.

Step 3: api/user_routes.py

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession

from app.database import get_db
from app.services import user_service

router = APIRouter()


@router.get("/users/{user_id}")
async def get_user(
    user_id: int,
    db: AsyncSession = Depends(get_db),
):
    user = await user_service.get_user_profile(db, user_id)

    if user is None:
        raise HTTPException(
            status_code=404,
            detail="User not found",
        )

    return user

The route doesn't know how the SQL query works. It asks the service for a user. That's clean separation of responsibilities. ✨

32. 🤔 Why Shouldn't Repositories Usually Commit?

Your app performs three operations: create_user(), create_profile(), create_settings(). If each repository function commits on its own and:

• ✅ User creation succeeds
• ✅ Profile creation succeeds
• ❌ Settings creation fails

...you've got a partially completed registration. Use one transaction instead:

async with db.begin():
    user = await user_repository.create(
        db,
        "deepika",
        "[e-mail]",
    )

    await profile_repository.create(db, user.id)
    await settings_repository.create(db, user.id)

async with db.begin() manages a transaction:

• If the block completes successfully, the transaction commits
• If an exception occurs, the transaction is rolled back

That makes several related operations atomic: all succeed together, or the transaction is rolled back.

📝 db.begin() expects the Session not to have already begun a transaction (for example, from an earlier query on the same Session). The exact transaction boundary is an architectural decision. The key is clear ownership, rather than committing randomly inside every repository method.

---

🔎 PART 11: Important SQLAlchemy Query Methods

33. 🆚 execute() vs scalars()

You'll see these constantly in SQLAlchemy 2.0.

statement = select(User)
result = await db.execute(statement)

execute() returns a SQLAlchemy Result object containing rows from the query.

🅰️ result.scalars() extracts the first selected element of each row. Since we selected User, it returns User ORM objects:

users = result.scalars().all()

🅱️ result.scalar_one_or_none() is for when you expect zero or one result:

• No matching result → None
• Exactly one result → that value
• More than one → raises MultipleResultsFound

🅲 result.scalar_one() expects exactly one:

• No result → raises NoResultFound
• One result → returns it
• Multiple → raises MultipleResultsFound

🅳 result.scalars().first() returns the first result or None, and doesn't enforce uniqueness.

📋 Quick comparison

Method	Expected result
all()	All rows
first()	First row or None
scalar_one()	Exactly one scalar
scalar_one_or_none()	Zero or one scalar
scalars().all()	All first-column values
scalars().first()	First scalar value

🔑 What is session.get()?

user = await db.get(User, 1)

Retrieves a User by primary key. It's especially convenient for primary-key lookups, and it can use the Session's identity map, possibly avoiding another SQL query if the object is already present and valid in that Session.

34. 🎛️ Filtering, Ordering and Limiting

Filtering

statement = select(User).where(User.username == "deepika")

SELECT * FROM users WHERE username = 'deepika';

Multiple conditions

from sqlalchemy import and_

statement = select(User).where(
    and_(
        User.is_active == True,
        User.username == "deepika",
    )
)

or simply:

statement = select(User).where(
    User.is_active.is_(True),
    User.username == "deepika",
)

Ordering

statement = select(User).order_by(User.username)
statement = select(User).order_by(User.id.desc())   # descending

Limiting

statement = select(User).limit(10)

SELECT * FROM users LIMIT 10;

Offset

statement = select(User).offset(10).limit(10)

That's the basic idea behind offset pagination. For a SocialFeed project, also learn cursor pagination, which is often preferable for large, changing feeds.

---

😴 PART 12: Relationships and Async Loading

35. 🐢 What is Lazy Loading?

user = await db.get(User, 1)

The User is retrieved. But what about user.posts? Are the posts automatically fetched? Not necessarily. The default relationship strategy is generally lazy="select", meaning related records load when the relationship is accessed. That's lazy loading.

⚠️ Traditional implicit lazy loading can cause problems in async SQLAlchemy, because accessing a relationship might require database I/O, and normal attribute access can't use await. So async apps should usually plan relationship loading explicitly.

🅰️ selectinload()

from sqlalchemy import select
from sqlalchemy.orm import selectinload

statement = select(User).options(selectinload(User.posts))

result = await db.execute(statement)
users = result.scalars().all()

1. SQLAlchemy retrieves the users
2. It retrieves their related posts with an additional query, generally using an IN condition
3. It populates the posts relationships

Now you can access the already-loaded relationships without triggering implicit lazy loading.

🅱️ joinedload()

from sqlalchemy.orm import joinedload

statement = select(User).options(joinedload(User.posts))

Loads related objects using a SQL JOIN. When loading collections this way, you generally need:

result = await db.execute(statement)
users = result.unique().scalars().all()

Why unique()? Joining a user with multiple posts can produce repeated rows for the same user in the SQL result, and unique() deduplicates ORM identities.

⚖️ selectinload() vs joinedload()

Feature	selectinload	joinedload
Query strategy	Separate SELECT	SQL JOIN
Collection loading	Often efficient	Can multiply result rows
Main advantage	Avoids large joined result sets	Related data in one SQL statement
Typical use	One-to-many collections	Many-to-one or small related loads

💡 Neither is universally better. Choose based on the query and the data size.

---

🧠 PART 13: Important Database Concepts

36. 🔁 What is a Transaction?

📌 A transaction is a group of database operations treated as one logical unit of work.

A user transfers ₹500 from Account A to Account B:

1. Deduct ₹500 from A
2. Add ₹500 to B

What if the first succeeds but the second fails? Money would disappear! Transactions prevent partial updates: both commit, or both roll back. SQLAlchemy manages transactions through Session methods and transaction context managers.

37. 🧪 What is ACID?

Property	Meaning	Example
Atomicity	All or nothing	Both transfer operations succeed or neither does
Consistency	Valid database rules remain satisfied	Foreign keys and constraints are respected
Isolation	Concurrent transactions are controlled	One transaction shouldn't see another's uncommitted changes under normal isolation
Durability	Committed changes persist	A successful transfer remains stored

38. 🚧 What is a Constraint?

📌 A constraint is a database rule that controls what data can be stored.

# Primary key
id: Mapped[int] = mapped_column(primary_key=True)

# Unique constraint
email: Mapped[str] = mapped_column(String(100), unique=True)

# Not null
username: Mapped[str] = mapped_column(String(50), nullable=False)

# Foreign key
user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))

Why do constraints matter? Application-level validation alone can't guarantee database consistency. Two concurrent requests could both check whether an email exists, both find it doesn't, and both try to insert it. A unique database constraint ensures only one succeeds.

39. ⚡ What is an Index?

Your users table has 10 million records and you run:

SELECT * FROM users WHERE email = '[e-mail]';

Without an appropriate index, PostgreSQL may need to scan many rows.

📌 An index is a data structure that helps the database find matching records efficiently.

email: Mapped[str] = mapped_column(
    String(100),
    unique=True,
    index=True,
)

Or at table level:

from sqlalchemy import Index


class User(Base):
    __tablename__ = "users"

    # columns ...

    __table_args__ = (
        Index("ix_users_username", "username"),
    )

• ✅ Indexes can speed up certain reads
• 💾 They consume storage
• 🐌 They add overhead to inserts, updates and deletes
• 🎯 Not every column needs one
• A unique constraint commonly creates a supporting unique index in PostgreSQL

---

⚠️ PART 14: Common Mistakes and Best Practices

40. ❌ Common SQLAlchemy Mistakes

Mistake	Correct approach
Sharing one AsyncSession between concurrent tasks	Use separate sessions
Forgetting await on async database calls	Await asynchronous operations
Forgetting to commit	Commit at the intended transaction boundary
Assuming flush() commits	Flush only sends SQL; it doesn't commit
Forgetting to roll back after a transaction failure	Roll back or use transaction context managers
Putting all queries in routers	Use repositories
Assuming model changes update PostgreSQL	Generate and apply migrations
Blindly applying autogenerated migrations	Review them first
Accessing unloaded relationships in async code	Use explicit loading strategies
Storing DB passwords in Git	Use environment configuration

41. 💥 What Happens When a Database Operation Fails?

You insert two users with the same email. PostgreSQL rejects the second insert because of the unique constraint, and SQLAlchemy raises an exception, commonly an IntegrityError. After a failed flush or commit, the Session must be rolled back before it can be used again.

from sqlalchemy.exc import IntegrityError

try:
    db.add(user)
    await db.commit()
except IntegrityError:
    await db.rollback()
    raise

A failed transaction leaves the Session in a failed transaction state until you roll back. Also, don't catch every exception and silently ignore it. Translate expected database errors into meaningful application errors and let unexpected ones be handled appropriately.

42. ⚖️ SQLAlchemy vs Alembic

SQLAlchemy	Alembic
Database toolkit and ORM	Database migration tool
Defines ORM models	Tracks schema changes
Executes queries	Applies migration operations
Creates, reads, updates and deletes records	Adds/removes tables and columns
Manages Sessions and transactions	Tracks schema revision history
Used by application code	Used during database deployment and maintenance

🧠 SQLAlchemy manages data operations. Alembic manages schema evolution. They work together but have different responsibilities.

---

🧵 PART 15: Putting Everything Together

43. 🛠️ A Complete Real-World Workflow

Your mentor asks you to add a Post table. Here's the full workflow:

1️⃣ Define the SQLAlchemy model in models/post.py

class Post(Base):
    __tablename__ = "posts"

    id: Mapped[int] = mapped_column(primary_key=True)

2️⃣ Register the model. Import it so its table is included in Base.metadata

3️⃣ Generate the migration

alembic revision --autogenerate -m "create posts table"

4️⃣ Review the migration. Inspect the generated upgrade() and downgrade() functions

5️⃣ Apply the migration

alembic upgrade head

PostgreSQL now has the new table. ✅

6️⃣ Create repository functions: create_post(), get_post_by_id(), delete_post()

7️⃣ Create service functions with business rules such as checking ownership and validating permissions

8️⃣ Create FastAPI endpoints

@router.post("/posts")
async def create_post(...):
    ...

9️⃣ Test the functionality. Verify the endpoint works and the data is correctly stored in PostgreSQL

44. 🗺️ How the Whole Architecture Connects

Component	Responsibility
⚡ FastAPI	Handles HTTP requests
✔️ Pydantic	Validates request and response data
🧱 SQLAlchemy ORM	Maps Python objects to database tables
🔧 SQLAlchemy Core	Provides SQL expression tools
🔌 Engine	Manages database connectivity
🗂️ AsyncSession	Manages asynchronous ORM operations and transactions
🚀 asyncpg	Communicates asynchronously with PostgreSQL
📚 Repository	Performs database access
🧠 Service	Implements business rules
🔄 Alembic	Changes the database schema over time
🐘 PostgreSQL	Permanently stores relational data

---

🎤 PART 16: Interview Questions

1. What is SQLAlchemy?
A Python SQL toolkit and ORM.

2. What's the main difference between Core and ORM?
Core works with tables and SQL expressions. The ORM works with mapped Python classes and objects.

3. What does flush() do?
It sends pending SQL changes to the database within the current transaction, without committing.

4. What is AsyncSession?
A session for asynchronous ORM database operations, used with an async engine and driver such as asyncpg.

5. What is Alembic used for?
Managing database schema migrations over time.

6. What does --autogenerate do?
It compares your model metadata with the actual database and proposes a migration. Always review the result.

7. What does alembic upgrade head do?
It applies all pending migrations up to the latest revision.

8. What is the Repository Pattern?
A layer that encapsulates database access, keeping queries out of routes and business logic.

9. What does target_metadata = Base.metadata do?
It tells Alembic which SQLAlchemy model metadata to compare against the database.

10. Can one AsyncSession be safely shared by concurrent asyncio tasks?
No. Each concurrent task should use its own AsyncSession.

---

🧪 PART 17: Test Yourself!

Try these before peeking at the answers! ❤️

1. What is SQLAlchemy?

   • A) A Python SQL toolkit and ORM
   • B) A database server
   • C) A Python web framework

2. What's the main difference between Core and ORM?

   • A) Core uses tables and SQL expressions; ORM uses mapped Python objects
   • B) Core is asynchronous; ORM is synchronous
   • C) They're completely unrelated tools

3. What does flush() do?

   • A) Permanently commits changes
   • B) Sends pending SQL changes within the transaction without committing
   • C) Closes the database connection

4. What is AsyncSession?

   • A) A session for asynchronous database operations
   • B) A PostgreSQL table
   • C) An Alembic migration

5. What is Alembic used for?

   • A) Validating HTTP request bodies
   • B) Managing database schema migrations
   • C) Creating FastAPI routes

6. What does --autogenerate do?

   • A) Automatically deploys the database
   • B) Compares model metadata with the database and proposes a migration
   • C) Deletes old migrations

7. What does alembic upgrade head do?

   • A) Applies pending migrations to the latest revision
   • B) Deletes all tables
   • C) Creates a new SQLAlchemy model

8. What is the Repository Pattern?

   • A) A way to organize Git repositories
   • B) A layer that encapsulates database access
   • C) A database indexing technique

9. What does target_metadata = Base.metadata do?

   • A) Tells Alembic which SQLAlchemy model metadata to compare
   • B) Creates a PostgreSQL database
   • C) Commits a transaction

10. Can one AsyncSession be safely shared by concurrent asyncio tasks?

    • A) Yes, always
    • B) No, each concurrent task should use its own AsyncSession
    • C) Only when using PostgreSQL

👀 Click to reveal answers

1. A: SQLAlchemy is a Python SQL toolkit and ORM
2. A: Core is SQL-expression oriented; the ORM maps Python objects
3. B: flush() sends SQL in the current transaction but doesn't commit
4. A: AsyncSession handles asynchronous ORM operations
5. B: Alembic manages schema migrations
6. B: Autogenerate compares metadata with the database and proposes changes
7. A: It applies pending migrations up to the latest revision
8. B: A repository encapsulates database access
9. A: It gives Alembic the model metadata to compare against the database
10. B: Sessions are stateful, so use one per concurrent task

---

🗺️ PART 18: What to Learn Next

Work through this checklist in order:

• [ ] Understand SQLAlchemy Core vs ORM
• [ ] Create DeclarativeBase and ORM models
• [ ] Understand Mapped, mapped_column and column types
• [ ] Create primary keys, foreign keys and relationships
• [ ] Understand Engine and connection pooling
• [ ] Master Session, flush, commit and rollback
• [ ] Learn AsyncEngine, AsyncSession and asyncpg
• [ ] Learn select(), execute(), scalars() and filtering
• [ ] Understand relationship loading strategies
• [ ] Create repository and service layers
• [ ] Initialize Alembic and configure target_metadata
• [ ] Generate and review autogenerated migrations
• [ ] Apply upgrades and perform downgrades
• [ ] Handle database transactions and IntegrityError

🏁 The 10 Most Important Things to Remember

1. 🧱 SQLAlchemy is a Python SQL toolkit and ORM
2. 🪄 The ORM maps Python classes to database tables
3. 📐 Declarative models define your tables using Python classes
4. 🔧 Core works with SQL expressions and table definitions
5. 🗂️ Session / AsyncSession manage ORM database interactions and transactions
6. 🚀 asyncpg is an asynchronous PostgreSQL driver
7. 🚿 flush() executes pending SQL without committing
8. 🔄 Alembic manages database schema changes through migrations
9. 🤖 Autogenerate proposes migrations by comparing metadata with the actual database
10. 📚 The Repository Pattern separates database access from business logic and API routes

🎯 One final distinction: if you add a bio column to your SQLAlchemy User model, PostgreSQL doesn't automatically get that column. You need to generate and apply an Alembic migration. And when you want to retrieve a user from PostgreSQL, SQLAlchemy's Session or AsyncSession executes the query. These are two different responsibilities.

---

🚀 What's Next?

Add a Post model to a small project, generate the migration with --autogenerate, read it before applying it, run alembic upgrade head, and then build the repository, service and route on top. Hands-on practice is where it all clicks! 💡

If this post helped you, drop a ❤️ or 🦄 and share it with a friend learning backend development. Happy coding! 👩‍💻✨