Database Integration Patterns

# database.py - SQLAlchemy connection pool setup

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session
from sqlalchemy.pool import QueuePool
from fastapi import Depends

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

# Create engine with connection pool
engine = create_engine(
    DATABASE_URL,

    # Connection pool configuration
    poolclass=QueuePool,
    pool_size=20,           # Keep 20 connections open
    max_overflow=10,        # Allow 10 more if needed (total 30 max)
    pool_pre_ping=True,     # Test connection before using (detect dead connections)
    pool_recycle=3600,      # Recycle connections after 1 hour (prevent stale connections)
    echo_pool=False,        # Log pool events (set True for debugging)

    # Performance settings
    pool_timeout=30,        # Wait max 30s for available connection
    pool_reset_on_return='rollback',  # Rollback on connection return
)

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

# Dependency for getting database session
def get_db() -> Session:
    """
    FastAPI dependency that provides database session.

    Lifecycle:
    1. Request arrives → Get connection from pool (~1ms)
    2. Endpoint executes → Use connection
    3. Request ends → Return connection to pool (via finally)

    The connection is NOT closed, just returned to pool for reuse!
    """
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()  # Returns connection to pool


# Usage in endpoint
from fastapi import FastAPI

app = FastAPI()

@app.get("/users/{user_id}")
async def get_user(user_id: int, db: Session = Depends(get_db)):
    """
    Uses pooled connection automatically via dependency injection.
    No manual connection management needed!
    """
    user = db.query(User).filter(User.id == user_id).first()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user


# Connection pool monitoring
@app.get("/health/database")
async def database_health():
    """Check database connection pool status"""
    return {
        "pool_size": engine.pool.size(),
        "checked_out": engine.pool.checkedout(),
        "overflow": engine.pool.overflow(),
        "available": engine.pool.size() - engine.pool.checkedout()
    }

# database_async.py - Async SQLAlchemy 2.0 setup

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.pool import AsyncAdaptedQueuePool

# Async database URL (note: postgresql+asyncpg)
DATABASE_URL = "postgresql+asyncpg://user:password@localhost:5432/mydb"

# Create async engine with connection pool
async_engine = create_async_engine(
    DATABASE_URL,

    # Connection pool settings
    poolclass=AsyncAdaptedQueuePool,
    pool_size=20,
    max_overflow=10,
    pool_pre_ping=True,
    pool_recycle=3600,

    # Async settings
    echo=False,
)

# Async session factory
AsyncSessionLocal = async_sessionmaker(
    async_engine,
    class_=AsyncSession,
    expire_on_commit=False,
    autocommit=False,
    autoflush=False,
)

# Async dependency
async def get_async_db() -> AsyncSession:
    """Async database session dependency"""
    async with AsyncSessionLocal() as session:
        try:
            yield session
        finally:
            await session.close()


# Usage in async endpoint
from fastapi import FastAPI, Depends
from sqlalchemy import select

app = FastAPI()

@app.get("/users/{user_id}")
async def get_user(user_id: int, db: AsyncSession = Depends(get_async_db)):
    """
    Truly async database query - doesn't block event loop.
    Can handle 1000+ concurrent requests.
    """
    result = await db.execute(
        select(User).where(User.id == user_id)
    )
    user = result.scalar_one_or_none()

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

    return user


# Multiple queries in parallel (requires separate sessions)
@app.get("/dashboard")
async def get_dashboard():
    """
    Run multiple queries concurrently using separate sessions.
    Each session has its own connection, enabling true parallelism.
    Note: A single session shares one connection, so asyncio.gather
    on the same session would execute sequentially, not in parallel.
    """
    import asyncio

    async def fetch_users():
        async with AsyncSessionLocal() as session:
            result = await session.execute(select(User).limit(10))
            return result.scalars().all()

    async def fetch_posts():
        async with AsyncSessionLocal() as session:
            result = await session.execute(select(Post).limit(10))
            return result.scalars().all()

    async def fetch_order_count():
        async with AsyncSessionLocal() as session:
            result = await session.execute(select(func.count(Order.id)))
            return result.scalar()

    # Each task uses its own session/connection → true parallel execution
    users, posts, order_count = await asyncio.gather(
        fetch_users(), fetch_posts(), fetch_order_count()
    )

    return {
        "users": users,
        "posts": posts,
        "order_count": order_count
    }

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

app = FastAPI()

@app.post("/orders")
async def create_order(
    order_data: OrderCreate,
    db: Session = Depends(get_db)
):
    """
    Create order with transaction management.
    If ANY step fails, entire transaction rolls back.
    """
    try:
        # Start transaction (implicitly with session)

        # Step 1: Create order
        order = Order(
            user_id=order_data.user_id,
            total_amount=order_data.total_amount,
            status="pending"
        )
        db.add(order)
        db.flush()  # Get order.id without committing

        # Step 2: Deduct inventory
        for item in order_data.items:
            product = db.query(Product).filter(Product.id == item.product_id).first()

            if product.stock < item.quantity:
                # Raise exception → triggers rollback
                raise HTTPException(
                    status_code=400,
                    detail=f"Insufficient stock for {product.name}"
                )

            product.stock -= item.quantity
            db.add(product)

        # Step 3: Create order items
        for item in order_data.items:
            order_item = OrderItem(
                order_id=order.id,
                product_id=item.product_id,
                quantity=item.quantity,
                price=item.price
            )
            db.add(order_item)

        # Step 4: Process payment (external API)
        payment_result = await process_payment(
            amount=order_data.total_amount,
            card_token=order_data.card_token
        )

        if not payment_result.success:
            # Payment failed → rollback everything
            raise HTTPException(
                status_code=402,
                detail="Payment failed: " + payment_result.error
            )

        # All steps succeeded → commit transaction
        db.commit()
        db.refresh(order)

        return {
            "order_id": order.id,
            "status": "completed",
            "total": order.total_amount
        }

    except HTTPException:
        # Explicit rollback (FastAPI dependency handles this automatically)
        db.rollback()
        raise

    except Exception as e:
        # Unexpected error → rollback
        db.rollback()
        raise HTTPException(
            status_code=500,
            detail=f"Order creation failed: {str(e)}"
        )


# Async transaction example
from sqlalchemy.ext.asyncio import AsyncSession

@app.post("/orders-async")
async def create_order_async(
    order_data: OrderCreate,
    db: AsyncSession = Depends(get_async_db)
):
    """Async transaction with explicit BEGIN"""
    async with db.begin():
        # Everything inside this block is a transaction
        # Automatic rollback if exception raised

        order = Order(**order_data.model_dump())
        db.add(order)
        await db.flush()

        # Deduct inventory
        for item in order_data.items:
            result = await db.execute(
                select(Product).where(Product.id == item.product_id)
            )
            product = result.scalar_one()

            if product.stock < item.quantity:
                # Exception → automatic rollback!
                raise HTTPException(400, "Insufficient stock")

            product.stock -= item.quantity

        # If we reach here, transaction commits automatically
        # If exception raised, automatic rollback

    return {"order_id": order.id}

@app.post("/import-users")
async def import_users(users: List[UserImport], db: Session = Depends(get_db)):
    """
    Import users, skip invalid ones but continue processing.
    Uses savepoints for partial rollbacks.
    """
    imported = []
    errors = []

    for user_data in users:
        # Create savepoint before each user
        savepoint = db.begin_nested()

        try:
            user = User(**user_data.model_dump())
            db.add(user)
            db.flush()  # Validate constraints

            # Success → commit savepoint
            savepoint.commit()
            imported.append(user.id)

        except IntegrityError as e:
            # Rollback just this user, continue with others
            savepoint.rollback()
            errors.append({
                "email": user_data.email,
                "error": "Duplicate email or constraint violation"
            })

    # Commit entire transaction (all successful users)
    db.commit()

    return {
        "imported": len(imported),
        "errors": len(errors),
        "failed_users": errors
    }

from sqlalchemy.orm import joinedload, selectinload, subqueryload
from fastapi import FastAPI, Depends

app = FastAPI()

# Strategy 1: joinedload - LEFT OUTER JOIN (1 query)
@app.get("/users/joinedload")
async def get_users_joined(db: Session = Depends(get_db)):
    """
    Uses LEFT OUTER JOIN to fetch users + posts in single query.

    SQL: SELECT * FROM users LEFT JOIN posts ON posts.user_id = users.id

    Pros: Single query, simple
    Cons: Cartesian explosion if user has many posts (duplicate user data)

    Best for: One-to-one or small one-to-many
    """
    users = db.query(User).options(joinedload(User.posts)).all()

    return [
        {
            "id": user.id,
            "username": user.username,
            "posts": [{"id": p.id, "title": p.title} for p in user.posts]
        }
        for user in users
    ]


# Strategy 2: selectinload - Separate IN query (2 queries)
@app.get("/users/selectinload")
async def get_users_selectin(db: Session = Depends(get_db)):
    """
    Fetches users, then posts with WHERE user_id IN (...).

    Query 1: SELECT * FROM users
    Query 2: SELECT * FROM posts WHERE user_id IN (1, 2, 3, ..., 100)

    Pros: No duplicate data, efficient for large collections
    Cons: 2 queries instead of 1

    Best for: One-to-many when users have many posts
    """
    users = db.query(User).options(selectinload(User.posts)).all()

    return [
        {
            "id": user.id,
            "username": user.username,
            "post_count": len(user.posts),
            "posts": [{"id": p.id, "title": p.title} for p in user.posts]
        }
        for user in users
    ]


# Strategy 3: subqueryload - Subquery (2 queries)
@app.get("/users/subquery")
async def get_users_subquery(db: Session = Depends(get_db)):
    """
    Similar to selectinload but uses subquery instead of IN.

    Query 1: SELECT * FROM users
    Query 2: SELECT * FROM posts WHERE user_id IN (SELECT id FROM users)

    Slightly different execution plan than selectinload.
    """
    users = db.query(User).options(subqueryload(User.posts)).all()
    return users


# Nested eager loading
@app.get("/users/nested")
async def get_users_nested(db: Session = Depends(get_db)):
    """
    Load users → posts → comments in one go.
    Prevents N+1 at multiple levels.
    """
    users = db.query(User)\
        .options(
            selectinload(User.posts).selectinload(Post.comments)
        )\
        .all()

    return [
        {
            "id": user.id,
            "username": user.username,
            "posts": [
                {
                    "id": post.id,
                    "title": post.title,
                    "comments": [
                        {"id": c.id, "text": c.text}
                        for c in post.comments
                    ]
                }
                for post in user.posts
            ]
        }
        for user in users
    ]


# Conditional eager loading
@app.get("/users")
async def get_users(
    include_posts: bool = False,
    db: Session = Depends(get_db)
):
    """
    Only eager load if client requests it (via query param).
    """
    query = db.query(User)

    if include_posts:
        query = query.options(selectinload(User.posts))

    users = query.all()
    return users

from sqlalchemy import Column, Integer, String, DateTime, Index, ForeignKey, Boolean
from sqlalchemy.orm import DeclarativeBase
from datetime import datetime, timezone

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)  # Primary key = automatic index
    username = Column(String(50), unique=True, index=True)  # Index for lookups
    email = Column(String(255), unique=True, index=True)  # Index for WHERE email = ?
    created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc), index=True)  # Index for sorting
    is_active = Column(Boolean, default=True, index=True)  # Index for filtering

    # Composite index for common query patterns
    __table_args__ = (
        # Index for: WHERE email = ? AND is_active = true
        Index('idx_user_email_active', 'email', 'is_active'),

        # Index for: WHERE is_active = ? ORDER BY created_at DESC
        Index('idx_user_active_created', 'is_active', 'created_at'),
    )


class Post(Base):
    __tablename__ = "posts"

    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey('users.id'), index=True)  # ALWAYS index FKs
    title = Column(String(255))
    content = Column(String)
    status = Column(String(20), index=True)  # Index for WHERE status = 'published'
    created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc), index=True)
    view_count = Column(Integer, default=0)

    __table_args__ = (
        # Index for: WHERE user_id = ? AND status = 'published'
        Index('idx_post_user_status', 'user_id', 'status'),

        # Index for: WHERE status = 'published' ORDER BY created_at DESC
        Index('idx_post_status_created', 'status', 'created_at'),

        # Partial index (PostgreSQL): only index published posts
        Index(
            'idx_post_published_created',
            'created_at',
            postgresql_where=(Column('status') == 'published')
        ),
    )


# Query performance with indexes
@app.get("/users/by-email/{email}")
async def get_user_by_email(email: str, db: Session = Depends(get_db)):
    """
    Without index on email: 500ms (full table scan)
    With index on email: 2ms (index lookup)
    """
    user = db.query(User).filter(User.email == email).first()
    return user


@app.get("/posts/published")
async def get_published_posts(db: Session = Depends(get_db)):
    """
    Uses idx_post_status_created composite index.
    WHERE status = 'published' ORDER BY created_at DESC

    Without index: 800ms
    With composite index: 5ms
    """
    posts = db.query(Post)\
        .filter(Post.status == 'published')\
        .order_by(Post.created_at.desc())\
        .limit(20)\
        .all()

    return posts

# Check if query uses index
from sqlalchemy import text

@app.get("/debug/query-plan")
async def debug_query(db: Session = Depends(get_db)):
    """
    Shows query execution plan - verify index usage.
    """
    query = """
        EXPLAIN ANALYZE
        SELECT * FROM users
        WHERE email = 'john@example.com'
    """

    result = db.execute(text(query)).fetchall()

    # Look for "Index Scan" (good) vs "Seq Scan" (bad)
    return {"explain": [row[0] for row in result]}

# Example output with index:
# Index Scan using idx_users_email on users (cost=0.42..8.44 rows=1)
# Time: 2.134 ms

# Example output WITHOUT index:
# Seq Scan on users (cost=0.00..35000.00 rows=1000000)
# Time: 524.891 ms

# database.py - Multiple database connections

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from fastapi import Depends

# Primary database (handles all writes + reads)
PRIMARY_DB_URL = "postgresql://user:pass@primary.db.internal:5432/mydb"

# Read replica (handles only reads)
REPLICA_DB_URL = "postgresql://user:pass@replica.db.internal:5432/mydb"

# Create engines
primary_engine = create_engine(
    PRIMARY_DB_URL,
    pool_size=20,
    max_overflow=10,
    pool_pre_ping=True,
)

replica_engine = create_engine(
    REPLICA_DB_URL,
    pool_size=50,  # More connections for read replica
    max_overflow=20,
    pool_pre_ping=True,
)

# Session factories
PrimarySession = sessionmaker(bind=primary_engine)
ReplicaSession = sessionmaker(bind=replica_engine)

# Dependencies
def get_primary_db():
    """Use for writes (POST, PUT, DELETE) and reads requiring fresh data"""
    db = PrimarySession()
    try:
        yield db
    finally:
        db.close()

def get_replica_db():
    """Use for reads (GET) that can tolerate slight replication lag"""
    db = ReplicaSession()
    try:
        yield db
    finally:
        db.close()


# Usage in endpoints
from fastapi import FastAPI

app = FastAPI()

# Read from replica
@app.get("/users")
async def list_users(db: Session = Depends(get_replica_db)):
    """
    Read-only query → uses replica.
    Can handle 10x more traffic than primary.
    """
    users = db.query(User).limit(100).all()
    return users

# Write to primary
@app.post("/users")
async def create_user(user: UserCreate, db: Session = Depends(get_primary_db)):
    """
    Write operation → uses primary.
    Replica will eventually get this data (replication lag ~100ms).
    """
    new_user = User(**user.model_dump())
    db.add(new_user)
    db.commit()
    db.refresh(new_user)
    return new_user

# Read after write → use primary!
@app.get("/users/{user_id}")
async def get_user(user_id: int, db: Session = Depends(get_primary_db)):
    """
    User might have just been created.
    Read from primary to avoid replication lag.
    """
    user = db.query(User).filter(User.id == user_id).first()
    if not user:
        raise HTTPException(404, "User not found")
    return user


# Automatic routing based on method
from fastapi import Request

async def get_db(request: Request):
    """
    Automatically route to replica for GET, primary for writes.
    """
    if request.method == "GET":
        db = ReplicaSession()
    else:
        db = PrimarySession()

    try:
        yield db
    finally:
        db.close()

# Now all endpoints can use single dependency
@app.get("/posts")
async def list_posts(db: Session = Depends(get_db)):
    """Automatically uses replica"""
    return db.query(Post).all()

@app.post("/posts")
async def create_post(post: PostCreate, db: Session = Depends(get_db)):
    """Automatically uses primary"""
    new_post = Post(**post.model_dump())
    db.add(new_post)
    db.commit()
    return new_post

# repositories/user_repository.py

from sqlalchemy.orm import Session
from typing import Optional, List
from models import User
from schemas import UserCreate, UserUpdate

class UserRepository:
    """
    Encapsulates all database operations for User model.
    """

    def __init__(self, db: Session):
        self.db = db

    def get_by_id(self, user_id: int) -> Optional[User]:
        """Get user by ID"""
        return self.db.query(User).filter(User.id == user_id).first()

    def get_by_email(self, email: str) -> Optional[User]:
        """Get user by email"""
        return self.db.query(User)\
            .filter(User.email == email)\
            .first()

    def get_by_username(self, username: str) -> Optional[User]:
        """Get user by username"""
        return self.db.query(User)\
            .filter(User.username == username)\
            .first()

    def list(self, skip: int = 0, limit: int = 100) -> List[User]:
        """List users with pagination"""
        return self.db.query(User)\
            .offset(skip)\
            .limit(limit)\
            .all()

    def create(self, user: UserCreate) -> User:
        """Create new user"""
        db_user = User(**user.model_dump())
        self.db.add(db_user)
        self.db.commit()
        self.db.refresh(db_user)
        return db_user

    def update(self, user_id: int, user_update: UserUpdate) -> Optional[User]:
        """Update user"""
        db_user = self.get_by_id(user_id)
        if not db_user:
            return None

        for key, value in user_update.model_dump(exclude_unset=True).items():
            setattr(db_user, key, value)

        self.db.commit()
        self.db.refresh(db_user)
        return db_user

    def delete(self, user_id: int) -> bool:
        """Delete user"""
        db_user = self.get_by_id(user_id)
        if not db_user:
            return False

        self.db.delete(db_user)
        self.db.commit()
        return True

    def search(self, query: str) -> List[User]:
        """Search users by username or email"""
        search_pattern = f"%{query}%"
        return self.db.query(User)\
            .filter(
                (User.username.ilike(search_pattern)) |
                (User.email.ilike(search_pattern))
            )\
            .all()


# Dependency to inject repository
def get_user_repository(db: Session = Depends(get_db)) -> UserRepository:
    return UserRepository(db)


# Using repository in endpoints
from fastapi import FastAPI, HTTPException, Depends

app = FastAPI()

@app.get("/users/{user_id}")
async def get_user(
    user_id: int,
    user_repo: UserRepository = Depends(get_user_repository)
):
    """
    Clean endpoint - all database logic in repository.
    Easy to test by mocking user_repo.
    """
    user = user_repo.get_by_id(user_id)
    if not user:
        raise HTTPException(404, "User not found")
    return user

@app.post("/users")
async def create_user(
    user: UserCreate,
    user_repo: UserRepository = Depends(get_user_repository)
):
    """
    Business logic in endpoint, data access in repository.
    """
    # Check if email already exists
    existing = user_repo.get_by_email(user.email)
    if existing:
        raise HTTPException(400, "Email already registered")

    # Create user via repository
    new_user = user_repo.create(user)
    return new_user

@app.get("/users/search")
async def search_users(
    q: str,
    user_repo: UserRepository = Depends(get_user_repository)
):
    """Search users"""
    results = user_repo.search(q)
    return {"results": results}

# tests/test_users.py

from unittest.mock import Mock
import pytest
from repositories.user_repository import UserRepository
from fastapi.testclient import TestClient

def test_get_user_success():
    """Test getting user by ID"""
    # Mock repository
    mock_repo = Mock(spec=UserRepository)
    mock_repo.get_by_id.return_value = User(
        id=1,
        username="john",
        email="john@example.com"
    )

    # Override dependency
    app.dependency_overrides[get_user_repository] = lambda: mock_repo

    # Test endpoint
    client = TestClient(app)
    response = client.get("/users/1")

    assert response.status_code == 200
    assert response.json()["username"] == "john"
    mock_repo.get_by_id.assert_called_once_with(1)

def test_get_user_not_found():
    """Test user not found"""
    mock_repo = Mock(spec=UserRepository)
    mock_repo.get_by_id.return_value = None

    app.dependency_overrides[get_user_repository] = lambda: mock_repo

    client = TestClient(app)
    response = client.get("/users/999")

    assert response.status_code == 404

# Install Alembic
pip install alembic

# Initialize Alembic in your project
alembic init alembic

# This creates:
# alembic/
#   versions/  # Migration files go here
#   env.py     # Configuration
# alembic.ini  # Settings

# Configure alembic.ini
# Edit alembic.ini and set database URL:
sqlalchemy.url = postgresql://user:password@localhost:5432/mydb

# Or use environment variable (better for production):
# sqlalchemy.url = ${DATABASE_URL}

# Configure alembic/env.py to use your models
# Edit alembic/env.py:
from models import Base  # Import your Base
target_metadata = Base.metadata


# Create first migration (auto-detect from models)
alembic revision --autogenerate -m "Create users and posts tables"

# This generates alembic/versions/abc123_create_users_and_posts_tables.py

# Apply migration
alembic upgrade head

# Rollback migration
alembic downgrade -1

# Example migration file
# alembic/versions/abc123_add_user_bio.py

"""Add bio column to users

Revision ID: abc123
Revises: xyz789
Create Date: 2024-01-15 10:30:00
"""

from alembic import op
import sqlalchemy as sa

# Migration metadata
revision = 'abc123'
down_revision = 'xyz789'
branch_labels = None
depends_on = None

def upgrade():
    """Apply migration (forward)"""
    # Add column
    op.add_column('users', sa.Column('bio', sa.String(500), nullable=True))

    # Add index
    op.create_index('idx_users_created_at', 'users', ['created_at'])

    # Create table
    op.create_table(
        'user_sessions',
        sa.Column('id', sa.Integer(), primary_key=True),
        sa.Column('user_id', sa.Integer(), nullable=False),
        sa.Column('token', sa.String(255), nullable=False),
        sa.Column('expires_at', sa.DateTime(), nullable=False),
        sa.ForeignKeyConstraint(['user_id'], ['users.id'])
    )

def downgrade():
    """Rollback migration (backward)"""
    # Drop table
    op.drop_table('user_sessions')

    # Drop index
    op.drop_index('idx_users_created_at', 'users')

    # Remove column
    op.drop_column('users', 'bio')


# Data migration example
def upgrade():
    """Add status column with default, then populate"""
    # Add column with default
    op.add_column('posts', sa.Column('status', sa.String(20), server_default='draft'))

    # Populate existing rows
    op.execute("UPDATE posts SET status = 'published' WHERE published_at IS NOT NULL")

    # Remove default (optional)
    op.alter_column('posts', 'status', server_default=None)