A social media platform like Instagram lives and dies by its database. Every time someone posts a photo, taps the heart icon, leaves a comment, or follows a creator, multiple tables are queried, updated, and indexed in real time.

To handle complex relational data (users, posts, likes, followers, notifications) while keeping our async Python server blazing fast, we designed a robust persistence stack using:

  • PostgreSQL 16+ (Relational SQL Database)
  • SQLAlchemy 2.0 (Asynchronous Object Relational Mapper)
  • asyncpg (High-Performance Async PostgreSQL Driver)
  • Alembic (Database Schema Migrations Engine)

In this tutorial, we will explore why we chose these tools, how they collaborate, and how to write clean, production-ready async database code.

1. The Architectural Pipeline

Here is how data flows from our FastAPI endpoints down to the disk storage in PostgreSQL:

+-------------------------------------------------------------+ | FastAPI Route Handler | | (e.g., POST /posts/{id}/like) | +------------------------------+------------------------------+ | Uses AsyncSession v +-------------------------------------------------------------+ | SQLAlchemy 2.0 (Async ORM) | | Maps Python Classes <---> SQL Statements (select/insert) | +------------------------------+------------------------------+ | Translates to raw SQL & parameters v +-------------------------------------------------------------+ | asyncpg (Database Driver) | | Non-blocking binary PostgreSQL wire-protocol client | +------------------------------+------------------------------+ | Async TCP Connection Pool v +-------------------------------------------------------------+ | PostgreSQL 16+ Database | | ACID Transactions, Foreign Keys, JSONB, B-Tree Indexes | +-------------------------------------------------------------+

2. Why PostgreSQL 16+ for an Instagram Clone?

Social media apps have strict relational integrity requirements:

  • If a user deletes their account, their likes, comments, and profile data must be cleanly handled (via ON DELETE CASCADE).
  • A user should not be able to follow the same account twice (enforced by UNIQUE compound constraints).
  • Feed queries must filter by follower relationships and order by creation timestamps in milliseconds.

Why PostgreSQL 16+?

  1. Strong ACID Guarantees: Prevents race conditions like duplicate likes or inconsistent follower counts.
  2. Advanced Indexing: B-tree and composite indexes allow lightning-fast timeline queries (WHERE user_id = X ORDER BY created_at DESC).
  3. JSONB Support: Great for flexible metadata, device tokens, or notification payloads.
  4. Performance & Query Optimization: PostgreSQL 16 brings improved query parallelism and memory efficiency for concurrent connections.

3. asyncpg: The Fastest PostgreSQL Driver for Python

In standard synchronous Python, drivers like psycopg2 block the entire execution thread while waiting for the database to return results.

asyncpg is fundamentally different:

  • It is written specifically for Python's asyncio and optimized in Cython.
  • It directly speaks the PostgreSQL binary protocol instead of translating through C libraries.
  • Benchmarks show asyncpg is up to 3x to 5x faster than traditional Python database drivers.

In our project, we configure the database URL with the postgresql+asyncpg:// protocol:

# Connection string using asyncpg driver DATABASE_URL = "postgresql+asyncpg://postgres:secret@localhost:5432/instagram_db"

4. SQLAlchemy 2.0 (Async ORM): Python Objects to SQL

SQLAlchemy is Python's industry-standard Object Relational Mapper (ORM). SQLAlchemy 2.0 introduces first-class async/await support with modern typing.

Step 1: Setting up the Async Engine & Session Factory 

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession from sqlalchemy.orm import DeclarativeBase DATABASE_URL = "postgresql+asyncpg://postgres:secret@localhost:5432/instagram_db" # Create async engine with connection pooling engine = create_async_engine( DATABASE_URL, echo=False, # Set True in debug mode to see raw SQL queries pool_size=10, # Maximum persistent connections max_overflow=20 # Burst connections during traffic spikes ) # Async session factory AsyncSessionLocal = async_sessionmaker( bind=engine, class_=AsyncSession, expire_on_commit=False ) # Base class for all database models class Base(DeclarativeBase): pass

Step 2: Defining Instagram Relational Models

Here is how we represent users, posts, and likes using SQLAlchemy 2.0 typed mappings:

from datetime import datetime from typing import List, Optional from sqlalchemy import String, ForeignKey, UniqueConstraint, DateTime, func 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), unique=True, index=True, nullable=False) email: Mapped[str] = mapped_column(String(255), unique=True, index=True, nullable=False) hashed_password: Mapped[str] = mapped_column(String(255), nullable=False) created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) # Relationships posts: Mapped[List["Post"]] = relationship(back_populates="author", cascade="all, delete-orphan") likes: Mapped[List["Like"]] = relationship(back_populates="user", cascade="all, delete-orphan") class Post(Base): __tablename__ = "posts" id: Mapped[int] = mapped_column(primary_key=True) caption: Mapped[Optional[str]] = mapped_column(String(2200), nullable=True) image_url: Mapped[str] = mapped_column(String(500), nullable=False) user_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"), index=True) created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), index=True) author: Mapped["User"] = relationship(back_populates="posts") likes: Mapped[List["Like"]] = relationship(back_populates="post", cascade="all, delete-orphan") class Like(Base): __tablename__ = "likes" id: Mapped[int] = mapped_column(primary_key=True) user_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"), index=True) post_id: Mapped[int] = mapped_column(ForeignKey("posts.id", ondelete="CASCADE"), index=True) created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) user: Mapped["User"] = relationship(back_populates="likes") post: Mapped["Post"] = relationship(back_populates="likes") # Prevent a user from liking the exact same post twice __table_args__ = ( UniqueConstraint("user_id", "post_id", name="uq_user_post_like"), )

Step 3: Querying the Database Asynchronously in FastAPI 

from fastapi import APIRouter, Depends, HTTPException from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession router = APIRouter() # Dependency providing an isolated session per HTTP request async def get_db(): async with AsyncSessionLocal() as session: try: yield session finally: await session.close() @router.get("/posts/{post_id}") async def get_post_details(post_id: int, db: AsyncSession = Depends(get_db)): query = select(Post).where(Post.id == post_id) result = await db.execute(query) post = result.scalar_one_or_none() if not post: raise HTTPException(status_code=404, detail="Post not found") return { "id": post.id, "caption": post.caption, "image_url": post.image_url, "author_id": post.user_id, }

5. Alembic: Painless Database Migrations

In production, you never delete and recreate your database every time your models change! Instead, you use migrations to evolve the schema version by version.

Alembic is the official database migration tool for SQLAlchemy. It tracks the migration state directly inside a table named alembic_version in your PostgreSQL database.

Our Project Migration Timeline:

In our Instagram backend, our database structure evolved in planned stages:

  1. 0001_initial_schema_baseline: Basic User and Profile tables.
  2. 0002_add_auth_and_sessions: Token tracking and password hashes.
  3. 0003_add_follow_privacy: Follower/following relationships & private accounts.
  4. 0004_feed_indexes: High-performance composite indexes for personalized feeds.
  5. 0005_add_post_lifecycle: Soft-deletes and archived posts.
  6. 0006_add_notifications: User activity notifications.

Essential Alembic Commands:

  1. Create an auto-generated migration (Alembic compares models with the live DB schema): 
 alembic revision --autogenerate -m "add_notifications_table"
  1. Apply migrations (upgrade to the latest schema version): 
 alembic upgrade head
  1. Rollback the last migration: 
 alembic downgrade -1
  1. Inspect current database migration revision: 
 alembic current

Summary & Best Practices

Component Responsibility Why We Use It
PostgreSQL 16+ Relational Database Bulletproof ACID transactions, indexes, and foreign keys.
asyncpg Async DB Driver High throughput non-blocking communication with PostgreSQL.
SQLAlchemy 2.0 Async ORM Pythonic model definitions, type safety, query builder.
Alembic Migration Manager Reproducible, version-controlled schema evolution across environments.

Golden Rules:

  1. Always use connection pools: Never open a raw TCP connection for each request; let create_async_engine handle pooling.
  2. Index foreign keys & timestamps: Queries that filter by user_id or sort by created_at DESC must have database indexes for fast feed loads.
  3. Commit transactions safely: Use async with session.begin(): blocks so that failures automatically rollback all changes.

In the next post, we will explore Authentication & Security: Argon2, JWT Tokens, and Passlib!