FastAPI with SQLModel

fastapi==0.104.1
uvicorn[standard]==0.24.0
sqlmodel==0.0.14
sqlalchemy==2.0.23
psycopg2-binary==2.9.9

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session
from sqlmodel import SQLModel

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

# Create engine
engine = create_engine(
    DATABASE_URL,
    echo=True,  # Log SQL queries (disable in production)
    future=True
)

# Session factory
SessionLocal = sessionmaker(
    bind=engine,
    class_=Session,
    expire_on_commit=False
)

def create_db_and_tables():
    """Create all tables"""
    SQLModel.metadata.create_all(engine)

def get_session():
    """Get database session for dependency injection"""
    with SessionLocal() as session:
        yield session

from sqlmodel import SQLModel, Field
from typing import Optional
from datetime import datetime

class User(SQLModel, table=True):
    """User model for both database and API"""
    id: Optional[int] = Field(default=None, primary_key=True)
    email: str = Field(unique=True, index=True)  # Index for fast queries
    username: str = Field(index=True)
    full_name: str
    hashed_password: str
    is_active: bool = Field(default=True)
    created_at: datetime = Field(default_factory=datetime.utcnow)
    updated_at: datetime = Field(default_factory=datetime.utcnow)

# Separate schemas for API input/output
class UserCreate(SQLModel):
    """Schema for user creation (no id, password)"""
    email: str
    username: str
    full_name: str
    password: str  # Plain text, never stored

class UserRead(SQLModel):
    """Schema for API response"""
    id: int
    email: str
    username: str
    full_name: str
    is_active: bool
    created_at: datetime

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from models import User, UserCreate, UserRead
from database import engine, get_session, create_db_and_tables

app = FastAPI()

@app.on_event("startup")
def on_startup():
    create_db_and_tables()

# CREATE
@app.post("/users", response_model=UserRead)
def create_user(user: UserCreate, session: Session = Depends(get_session)):
    """Create a new user"""
    # Hash password (use bcrypt in production)
    db_user = User(
        email=user.email,
        username=user.username,
        full_name=user.full_name,
        hashed_password=f"hashed_{user.password}"  # TODO: use bcrypt
    )
    session.add(db_user)
    session.commit()
    session.refresh(db_user)  # Refresh to get ID
    return db_user

# READ one
@app.get("/users/{user_id}", response_model=UserRead)
def get_user(user_id: int, session: Session = Depends(get_session)):
    """Get user by ID"""
    user = session.query(User).filter(User.id == user_id).first()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

# READ all
@app.get("/users", response_model=list[UserRead])
def list_users(
    skip: int = 0,
    limit: int = 10,
    session: Session = Depends(get_session)
):
    """List all users with pagination"""
    users = session.query(User).offset(skip).limit(limit).all()
    return users

# UPDATE
@app.put("/users/{user_id}", response_model=UserRead)
def update_user(
    user_id: int,
    user_update: UserCreate,
    session: Session = Depends(get_session)
):
    """Update user"""
    db_user = session.query(User).filter(User.id == user_id).first()
    if not db_user:
        raise HTTPException(status_code=404, detail="User not found")
    
    # Update fields
    db_user.email = user_update.email
    db_user.username = user_update.username
    db_user.full_name = user_update.full_name
    db_user.updated_at = datetime.utcnow()
    
    session.add(db_user)
    session.commit()
    session.refresh(db_user)
    return db_user

# DELETE
@app.delete("/users/{user_id}")
def delete_user(user_id: int, session: Session = Depends(get_session)):
    """Delete user"""
    db_user = session.query(User).filter(User.id == user_id).first()
    if not db_user:
        raise HTTPException(status_code=404, detail="User not found")
    
    session.delete(db_user)
    session.commit()
    return {"message": "User deleted"}

from typing import Optional, List

class Team(SQLModel, table=True):
    """Team model"""
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    description: str

class Player(SQLModel, table=True):
    """Player belongs to a team"""
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    team_id: int = Field(foreign_key="team.id")  # Foreign key
    team: Optional[Team] = None  # Relationship

# One-to-many with back population
from sqlalchemy.orm import relationship

class TeamWithPlayers(SQLModel, table=False):
    """Team with all players"""
    id: int
    name: str
    description: str
    players: List[Player] = []

# Create team with players
@app.post("/teams")
def create_team(name: str, session: Session = Depends(get_session)):
    team = Team(name=name, description="")
    session.add(team)
    session.commit()
    session.refresh(team)
    return team

# Add player to team
@app.post("/players")
def create_player(
    name: str,
    team_id: int,
    session: Session = Depends(get_session)
):
    player = Player(name=name, team_id=team_id)
    session.add(player)
    session.commit()
    session.refresh(player)
    return player

# Get team with all players
@app.get("/teams/{team_id}")
def get_team_with_players(
    team_id: int,
    session: Session = Depends(get_session)
):
    team = session.query(Team).filter(Team.id == team_id).first()
    if not team:
        raise HTTPException(status_code=404, detail="Team not found")
    
    players = session.query(Player).filter(Player.team_id == team_id).all()
    return {
        "id": team.id,
        "name": team.name,
        "description": team.description,
        "players": players
    }

from sqlalchemy import or_, and_

# Filter by single condition
@app.get("/users/by-email/{email}")
def get_user_by_email(email: str, session: Session = Depends(get_session)):
    user = session.query(User).filter(User.email == email).first()
    return user

# Multiple conditions (AND)
@app.get("/users/active")
def get_active_users(session: Session = Depends(get_session)):
    users = session.query(User).filter(
        and_(
            User.is_active == True,
            User.created_at.like("2024%")
        )
    ).all()
    return users

# OR conditions
@app.get("/search")
def search_users(query: str, session: Session = Depends(get_session)):
    users = session.query(User).filter(
        or_(
            User.email.ilike(f"%{query}%"),
            User.username.ilike(f"%{query}%"),
            User.full_name.ilike(f"%{query}%")
        )
    ).all()
    return users

# Sorting and pagination
@app.get("/users")
def list_users(
    skip: int = 0,
    limit: int = 10,
    sort_by: str = "created_at",
    order: str = "desc",
    session: Session = Depends(get_session)
):
    query = session.query(User)
    
    # Apply sorting
    if sort_by == "created_at":
        if order == "desc":
            query = query.order_by(User.created_at.desc())
        else:
            query = query.order_by(User.created_at.asc())
    
    # Apply pagination
    users = query.offset(skip).limit(limit).all()
    return users

# Count queries
@app.get("/users/count")
def count_users(session: Session = Depends(get_session)):
    count = session.query(User).count()
    return {"total_users": count}

from pydantic import EmailStr, Field
from enum import Enum

class UserRole(str, Enum):
    """Role enumeration"""
    ADMIN = "admin"
    USER = "user"
    GUEST = "guest"

class User(SQLModel, table=True):
    """User with validation"""
    id: Optional[int] = Field(default=None, primary_key=True)
    email: EmailStr = Field(unique=True, index=True)  # Email validation
    username: str = Field(
        min_length=3,  # At least 3 characters
        max_length=50,  # Max 50 characters
        regex="^[a-zA-Z0-9_]+$"  # Alphanumeric + underscore
    )
    age: int = Field(ge=0, le=150)  # Between 0 and 150
    role: UserRole = Field(default=UserRole.USER)
    bio: Optional[str] = Field(default=None, max_length=500)

# Validation triggers on API input and SQLModel creation
@app.post("/users", response_model=UserRead)
def create_user(user: UserCreate, session: Session = Depends(get_session)):
    # If email invalid, FastAPI returns 422 automatically
    # If username doesn't match regex, 422 error
    # If age > 150, 422 error
    db_user = User(**user.model_dump())
    session.add(db_user)
    session.commit()
    return db_user

from sqlalchemy import Index

class Post(SQLModel, table=True):
    """Blog post with indexes"""
    id: Optional[int] = Field(default=None, primary_key=True)
    title: str = Field(index=True)  # Simple index
    content: str
    author_id: int = Field(foreign_key="user.id", index=True)
    published: bool = Field(default=False, index=True)
    created_at: datetime = Field(default_factory=datetime.utcnow, index=True)
    
    # Composite index for common queries
    __table_args__ = (
        Index("idx_author_published", "author_id", "published"),
    )

class Comment(SQLModel, table=True):
    """Comment with composite index"""
    id: Optional[int] = Field(default=None, primary_key=True)
    post_id: int = Field(foreign_key="post.id", index=True)
    user_id: int = Field(foreign_key="user.id", index=True)
    content: str
    created_at: datetime = Field(default_factory=datetime.utcnow)
    
    # Index for finding comments by post + user
    __table_args__ = (
        Index("idx_post_user", "post_id", "user_id"),
    )

from sqlalchemy.exc import SQLAlchemyError

@app.post("/transfer")
def transfer_funds(
    from_user_id: int,
    to_user_id: int,
    amount: float,
    session: Session = Depends(get_session)
):
    """Transfer funds with transaction"""
    try:
        # Deduct from sender
        from_user = session.query(User).filter(User.id == from_user_id).first()
        if not from_user or from_user.balance < amount:
            raise ValueError("Insufficient funds")
        from_user.balance -= amount
        
        # Add to receiver
        to_user = session.query(User).filter(User.id == to_user_id).first()
        if not to_user:
            raise ValueError("Receiver not found")
        to_user.balance += amount
        
        # Both succeed or both fail
        session.add(from_user)
        session.add(to_user)
        session.commit()
        
        return {"status": "success", "from": from_user_id, "to": to_user_id, "amount": amount}
    
    except SQLAlchemyError as e:
        session.rollback()
        raise HTTPException(status_code=500, detail="Transaction failed")
    except ValueError as e:
        session.rollback()
        raise HTTPException(status_code=400, detail=str(e))

# Bulk insert
@app.post("/bulk-users")
def bulk_create_users(
    users_data: list[UserCreate],
    session: Session = Depends(get_session)
):
    """Create many users at once"""
    users = [User(**user.model_dump()) for user in users_data]
    session.add_all(users)
    session.commit()
    return {"created": len(users)}

# Bulk update
@app.patch("/deactivate-inactive")
def deactivate_inactive_users(
    days_inactive: int,
    session: Session = Depends(get_session)
):
    """Deactivate users inactive for N days"""
    from datetime import datetime, timedelta
    
    cutoff_date = datetime.utcnow() - timedelta(days=days_inactive)
    
    # Bulk update query
    session.query(User).filter(
        User.is_active == True,
        User.updated_at < cutoff_date
    ).update({User.is_active: False})
    
    session.commit()
    return {"message": "Deactivated inactive users"}

# Bulk delete
@app.delete("/delete-old-users")
def delete_old_users(
    days_old: int,
    session: Session = Depends(get_session)
):
    """Delete users created more than N days ago"""
    from datetime import datetime, timedelta
    
    cutoff_date = datetime.utcnow() - timedelta(days=days_old)
    
    count = session.query(User).filter(User.created_at < cutoff_date).delete()
    session.commit()
    return {"deleted": count}

# Initialize Alembic
alembic init alembic

# Configure database URL in alembic.ini
# sqlalchemy.url = postgresql://user:password@localhost/dbname

# Create migration for current models
alembic revision --autogenerate -m "Initial schema"

# Apply migration
alembic upgrade head

# See migration history
alembic history

# Rollback to previous migration
alembic downgrade -1

from sqlmodel import SQLModel
from models import User, Team, Player, Post, Comment

# This makes Alembic aware of all models
target_metadata = SQLModel.metadata

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlmodel import Session, SQLModel

from main import app, get_session
from models import User

# Use in-memory SQLite for tests
SQLALCHEMY_TEST_DATABASE_URL = "sqlite:///:memory:"

@pytest.fixture(name="session")
def session_fixture():
    engine = create_engine(
        SQLALCHEMY_TEST_DATABASE_URL,
        connect_args={"check_same_thread": False}
    )
    SQLModel.metadata.create_all(engine)
    with Session(engine) as session:
        yield session

@pytest.fixture(name="client")
def client_fixture(session: Session):
    def get_session_override():
        return session
    
    app.dependency_overrides[get_session] = get_session_override
    client = TestClient(app)
    yield client
    app.dependency_overrides.clear()

def test_create_user(client: TestClient):
    """Test user creation"""
    response = client.post(
        "/users",
        json={
            "email": "[email protected]",
            "username": "testuser",
            "full_name": "Test User",
            "password": "secret123"
        }
    )
    assert response.status_code == 200
    data = response.json()
    assert data["email"] == "[email protected]"
    assert data["username"] == "testuser"

def test_get_user(client: TestClient):
    """Test reading user"""
    # Create first
    create_response = client.post(
        "/users",
        json={
            "email": "[email protected]",
            "username": "testuser",
            "full_name": "Test User",
            "password": "secret123"
        }
    )
    user_id = create_response.json()["id"]
    
    # Read
    response = client.get(f"/users/{user_id}")
    assert response.status_code == 200
    assert response.json()["email"] == "[email protected]"