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

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