FastAPI Database Integration with SQLAlchemy

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import DeclarativeBase
import os

# Production-grade PostgreSQL async connection string
DATABASE_URL = os.getenv("DATABASE_URL", "postgresql+asyncpg://user:pass@localhost/forge_db")

# create_async_engine prevents the event loop from blocking during I/O
engine = create_async_engine(DATABASE_URL, echo=False, future=True)

# Factory for creating session objects
AsyncSessionLocal = async_sessionmaker(
    bind=engine, 
    class_=AsyncSession, 
    expire_on_commit=False
)

class Base(DeclarativeBase):
    """Base class for all SQLAlchemy models"""
    pass

# Dependency: The 'Session-per-Request' provider
async def get_db():
    async with AsyncSessionLocal() as session:
        try:
            yield session
            # Session is automatically committed/closed if using 'async with' context
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

engine = create_async_engine(
    DATABASE_URL,
    echo=False,
    pool_size=5,          # adjust to (worker_count + 1)
    max_overflow=0,       # no extra connections beyond pool_size
    pool_timeout=15,      # seconds
    pool_recycle=300,     # recycle every 5 minutes
    pool_pre_ping=True    # verify connection before use
)

from sqlalchemy import String, Integer, select
from sqlalchemy.orm import Mapped, mapped_column
from fastapi import FastAPI, Depends, HTTPException, status
from io.thecodeforge.db.session import Base, get_db, AsyncSession
from pydantic import BaseModel, EmailStr

# 1. SQLAlchemy Model (Database Identity)
class User(Base):
    __tablename__ = 'users'
    id: Mapped[int] = mapped_column(primary_key=True, index=True)
    username: Mapped[str] = mapped_column(String(50), unique=True)
    email: Mapped[str] = mapped_column(unique=True)

# 2. Pydantic Schema (API Contract)
class UserCreate(BaseModel):
    username: str
    email: EmailStr

app = FastAPI()

@app.post('/forge/users', status_code=status.HTTP_201_CREATED)
async def create_user(payload: UserCreate, db: AsyncSession = Depends(get_db)):
    new_user = User(username=payload.username, email=payload.email)
    db.add(new_user)
    await db.commit()
    await db.refresh(new_user) # Reloads auto-generated ID from DB
    return new_user

@app.get('/forge/users/{user_id}')
async def read_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()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

from sqlalchemy import select, update as sa_update
from io.thecodeforge.schemas import UserUpdate  # Pydantic schema

@app.put('/forge/users/{user_id}')
async def update_user(user_id: int, payload: UserUpdate, db: AsyncSession = Depends(get_db)):
    # Load the existing user
    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")
    
    # Update fields that were provided
    update_data = payload.model_dump(exclude_unset=True)
    for field, value in update_data.items():
        setattr(user, field, value)
    
    await db.commit()
    await db.refresh(user)  # Get any database-generated values
    return user

@app.delete('/forge/users/{user_id}', status_code=status.HTTP_204_NO_CONTENT)
async def delete_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()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    
    await db.delete(user)
    await db.commit()
    # No refresh needed — we are returning 204 No Content

from io.thecodeforge.db.session import AsyncSession
from io.thecodeforge.db.models import User
from sqlalchemy import select

class UserRepository:
    def __init__(self, db: AsyncSession):
        self.db = db

    async def get_by_id(self, user_id: int) -> User | None:
        result = await self.db.execute(select(User).where(User.id == user_id))
        return result.scalar_one_or_none()

    async def create(self, username: str, email: str) -> User:
        user = User(username=username, email=email)
        self.db.add(user)
        await self.db.commit()
        await self.db.refresh(user)
        return user

# In endpoint:
@app.get('/forge/users/{user_id}')
async def read_user(user_id: int, db: AsyncSession = Depends(get_db)):
    repo = UserRepository(db)
    user = await repo.get_by_id(user_id)
    if not user:
        raise HTTPException(404)
    return user

from logging.config import fileConfig
from sqlalchemy import pool
from sqlalchemy.ext.asyncio import create_async_engine
from alembic import context
from io.thecodeforge.db.session import Base
from io.thecodeforge.db.models import User  # noqa: F401

config = context.config
fileConfig(config.config_file_name)

target_metadata = Base.metadata

def run_migrations_offline():
    url = config.get_main_option("sqlalchemy.url")
    context.configure(url=url, target_metadata=target_metadata, literal_binds=True)
    with context.begin_transaction():
        context.run_migrations()

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

async def run_async_migrations():
    connectable = create_async_engine(config.get_main_option("sqlalchemy.url"), poolclass=pool.NullPool)
    async with connectable.connect() as connection:
        await connection.run_sync(do_run_migrations)
    await connectable.dispose()

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

context.run_migrations = run_migrations_online if context.is_offline_mode() else run_migrations_offline

version: '3.9'

services:
  app:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql+asyncpg://forge_user:forge_pass@db:5432/forge_db
    depends_on:
      db:
        condition: service_healthy

  db:
    image: postgres:15-alpine
    environment:
      - POSTGRES_USER=forge_user
      - POSTGRES_PASSWORD=forge_pass
      - POSTGRES_DB=forge_db
    ports:
      - "5432:5432"
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U forge_user -d forge_db"]
      interval: 5s
      timeout: 5s
      retries: 5
    volumes:
      - postgres_data:/var/lib/postgresql/data

volumes:
  postgres_data:

import pytest
from httpx import AsyncClient, ASGITransport
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from io.thecodeforge.db.session import Base, get_db
from io.thecodeforge.main import app

# Use a separate test database
TEST_DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/test_forge_db"

@pytest.fixture
async def async_engine():
    engine = create_async_engine(TEST_DATABASE_URL, echo=False)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield engine
    await engine.dispose()

@pytest.fixture
async def db_session(async_engine) -> AsyncSession:
    session = async_sessionmaker(bind=async_engine, class_=AsyncSession, expire_on_commit=False)()
    try:
        yield session
    finally:
        await session.close()

@pytest.fixture
async def client(db_session):
    async def override_get_db():
        yield db_session
    app.dependency_overrides[get_db] = override_get_db
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://testserver") as ac:
        yield ac

@pytest.mark.asyncio
async def test_create_user(client: AsyncClient):
    response = await client.post("/forge/users", json={"username": "tester", "email": "tester@test.com"})
    assert response.status_code == 201
    data = response.json()
    assert data["username"] == "tester"

async def create_order(db: AsyncSession, order_data: OrderCreate) -> Order:
    async with db.begin():
        # Automatically commits or rolls back on exception
        user = await db.execute(select(User).where(User.id == order_data.user_id))
        user = user.scalar_one_or_none()
        if not user:
            raise HTTPException(status_code=404, detail="User not found")
        order = Order(user_id=user.id, total=order_data.total)
        db.add(order)
        # Use nested transaction for partial rollback
        async with db.begin_nested():
            for item in order_data.items:
                inventory_item = await db.execute(select(Inventory).where(Inventory.product_id == item.product_id))
                inventory_item = inventory_item.scalar_one_or_none()
                if not inventory_item or inventory_item.quantity < item.quantity:
                    raise HTTPException(status_code=400, detail="Insufficient stock")
                inventory_item.quantity -= item.quantity
    await db.refresh(order)
    return order

// io.thecodeforge — python tutorial

from fastapi import Depends, FastAPI
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker

app = FastAPI()
engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db", pool_size=5, max_overflow=10)
SessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

async def get_db():
    session = SessionLocal()
    try:
        yield session
        await session.commit()  # commit on success
    except Exception:
        await session.rollback()
        raise
    finally:
        await session.close()

@app.post("/users/")
async def create_user(db: AsyncSession = Depends(get_db)):
    # db is an AsyncSession, not a generator
    user = User(name="Alice")
    db.add(user)
    return {"id": user.id}

// io.thecodeforge — python tutorial

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker

# Wrong: expire_on_commit=True (default)
engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")
BadSession = sessionmaker(engine, class_=AsyncSession)

# Correct: expire_on_commit=False
engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")
GoodSession = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

async def create_and_read(db: AsyncSession):
    user = User(name="Bob")
    db.add(user)
    await db.commit()
    # Without expire_on_commit=False, this next line triggers a DB query
    # With expire_on_commit=False, it uses the cached value
    print(user.name)  # No hidden await here

// io.thecodeforge — python tutorial
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from app.schemas import HeroPublic
from app.models import Hero

async def read_heroes(db: AsyncSession, skip: int = 0, limit: int = 10) -> list[HeroPublic]:
    result = await db.execute(select(Hero).offset(skip).limit(limit))
    heroes = result.scalars().all()
    return [HeroPublic.model_validate(h) for h in heroes]

async def read_hero(db: AsyncSession, hero_id: int) -> HeroPublic:
    hero = await db.get(Hero, hero_id)
    if not hero:
        return None
    return HeroPublic.model_validate(hero)

// io.thecodeforge — python tutorial
from sqlalchemy.ext.asyncio import AsyncSession
from app.schemas import HeroUpdate, HeroPublic
from app.models import Hero

async def update_hero(db: AsyncSession, hero_id: int, updates: HeroUpdate) -> HeroPublic:
    hero = await db.get(Hero, hero_id)
    if not hero:
        return None
    update_data = updates.model_dump(exclude_unset=True)
    for field, value in update_data.items():
        setattr(hero, field, value)
    await db.commit()
    await db.refresh(hero)
    return HeroPublic.model_validate(hero)