backend/
├── app/
│   ├── db/
│   │   ├── database.py          # Core async database configuration
│   │   └── __init__.py
│   └── routers/
│       ├── auth/
│       │   ├── models.py        # User authentication models
│       │   └── ...
│       ├── payments/
│       │   ├── models.py        # Payment-related models
│       │   └── ...
│       └── your_feature/
│           ├── models.py        # Your custom models
│           └── ...
├── alembic/
│   ├── env.py                   # Alembic configuration
│   ├── versions/                # Migration files
│   └── alembic.ini             # Alembic settings
└── docker-compose.yml          # Database container setup

import os
from dotenv import load_dotenv
from fastapi import Depends
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.orm import declarative_base
from typing import Annotated

# Load environment variables from .env file
load_dotenv()

# Get database URL from environment (must use async driver)
SQLALCHEMY_DATABASE_URL = os.getenv("DATABASE_URL")

# Create async SQLAlchemy engine
engine = create_async_engine(SQLALCHEMY_DATABASE_URL, echo=False)

# Create async session factory
SessionLocal = async_sessionmaker(
    autocommit=False,
    autoflush=False,
    bind=engine,
    class_=AsyncSession,
    expire_on_commit=False  # Prevents lazy-loading issues after commit
)

# Create Base class for declarative models
Base = declarative_base()

# Async database dependency function
async def get_db():
    """
    Async database dependency that provides database sessions.
    Automatically handles session creation and cleanup.
    """
    async with SessionLocal() as session:
        yield session

# Type annotation for FastAPI dependency injection
db_dependency = Annotated[AsyncSession, Depends(get_db)]

from sqlalchemy.orm import relationship
from ...db.database import Base
from sqlalchemy import Column, Integer, String, Boolean

class User(Base):
    """User model for authentication"""
    __tablename__ = "users"

    # Primary key with auto-increment
    id = Column(Integer, primary_key=True, index=True, autoincrement=True)

    # Unique user identifiers
    username = Column(String, unique=True)
    email = Column(String, unique=True)

    # Password storage (always hashed, never plain text)
    hashed_password = Column(String)

    # Email verification status
    is_verified = Column(Boolean, default=False)

    # OAuth integration fields
    is_oauth = Column(Boolean, default=False)
    google_sub = Column(String(100), nullable=True, unique=True, index=True)

    # Email verification and password reset tokens
    verification_token = Column(String, nullable=True)
    reset_token = Column(String, nullable=True)

    # Stripe payment integration
    customer_id = Column(String(255), nullable=True)
    plan_id = Column(Integer, nullable=True)
    subscription_id = Column(String(255), nullable=True)
    subscription_status = Column(String(64), nullable=True)
    subscription_last_renew = Column(String, nullable=True)
    subscription_next_renew = Column(String, nullable=True)

import os
from logging.config import fileConfig
from sqlalchemy import engine_from_config, pool
from alembic import context

# Alembic configuration object
config = context.config

# Setup logging from alembic.ini
if config.config_file_name is not None:
    fileConfig(config.config_file_name)

# Import all your model modules
from app.routers.auth import models as auth_models
from app.routers.payments import models as payments_models

# This tells Alembic about all your models
target_metadata = [auth_models.Base.metadata]

def run_migrations_offline() -> None:
    """Run migrations in 'offline' mode - generates SQL without database connection"""
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )
    with context.begin_transaction():
        context.run_migrations()

def run_migrations_online():
    """Run migrations in 'online' mode - executes against live database"""
    # Get database URL from environment
    db_url = os.getenv('DATABASE_URL')
    if db_url:
        config.set_main_option("sqlalchemy.url", db_url)

    # Create engine (note: migrations use sync engine even with async app)
    connectable = engine_from_config(
        config.get_section(config.config_ini_section),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool
    )

    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata
        )
        with context.begin_transaction():
            context.run_migrations()

# Run appropriate migration mode
if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()

# Rollback to previous migration
alembic downgrade -1

# Rollback to specific revision
alembic downgrade abc123def456

# Rollback all migrations (dangerous!)
alembic downgrade base

from fastapi import HTTPException
from sqlalchemy import select
from app.db.database import db_dependency
from app.routers.blog.models import BlogPost, BlogCategory

@app.get("/blog/posts")
async def get_blog_posts(
    db: db_dependency,
    skip: int = 0,
    limit: int = 10,
    published_only: bool = True
):
    """Get blog posts with pagination and filtering"""
    # Use SQLAlchemy 2.0 select() syntax
    query = select(BlogPost)

    if published_only:
        query = query.where(BlogPost.is_published == True)

    # Execute async query
    result = await db.execute(query.offset(skip).limit(limit))
    posts = result.scalars().all()

    return posts

@app.get("/blog/posts/{post_id}")
async def get_blog_post(post_id: int, db: db_dependency):
    """Get single blog post"""
    result = await db.execute(
        select(BlogPost).where(BlogPost.id == post_id)
    )
    post = result.scalar_one_or_none()

    if not post:
        raise HTTPException(status_code=404, detail="Post not found")

    return post

@app.post("/blog/posts")
async def create_blog_post(
    post_data: BlogPostCreate,
    db: db_dependency,
    current_user: User = Depends(get_current_user)
):
    """Create new blog post"""
    db_post = BlogPost(
        title=post_data.title,
        slug=post_data.slug,
        content=post_data.content,
        excerpt=post_data.excerpt,
        author_id=current_user.id
    )

    db.add(db_post)
    await db.commit()
    await db.refresh(db_post)

    return db_post

@app.put("/blog/posts/{post_id}")
async def update_blog_post(
    post_id: int,
    post_data: BlogPostUpdate,
    db: db_dependency,
    current_user: User = Depends(get_current_user)
):
    """Update existing blog post"""
    result = await db.execute(
        select(BlogPost).where(BlogPost.id == post_id)
    )
    post = result.scalar_one_or_none()

    if not post:
        raise HTTPException(status_code=404, detail="Post not found")

    if post.author_id != current_user.id:
        raise HTTPException(status_code=403, detail="Not authorized")

    # Update fields
    for field, value in post_data.dict(exclude_unset=True).items():
        setattr(post, field, value)

    await db.commit()
    await db.refresh(post)

    return post

@app.delete("/blog/posts/{post_id}")
async def delete_blog_post(
    post_id: int,
    db: db_dependency,
    current_user: User = Depends(get_current_user)
):
    """Delete blog post"""
    result = await db.execute(
        select(BlogPost).where(BlogPost.id == post_id)
    )
    post = result.scalar_one_or_none()

    if not post:
        raise HTTPException(status_code=404, detail="Post not found")

    if post.author_id != current_user.id:
        raise HTTPException(status_code=403, detail="Not authorized")

    await db.delete(post)
    await db.commit()

    return {"message": "Post deleted successfully"}

from sqlalchemy import select, func, and_, or_
from sqlalchemy.orm import selectinload
from datetime import datetime, timedelta

# Count queries
result = await db.execute(select(func.count(BlogPost.id)))
post_count = result.scalar()

# Eager loading relationships
result = await db.execute(
    select(BlogPost).options(selectinload(BlogPost.author))
)
posts_with_authors = result.scalars().all()

# Complex filtering
result = await db.execute(
    select(BlogPost).where(
        and_(
            BlogPost.is_published == True,
            BlogPost.created_at >= datetime.utcnow() - timedelta(days=7)
        )
    )
)
recent_posts = result.scalars().all()

# Ordering and limiting
result = await db.execute(
    select(BlogPost)
    .where(BlogPost.is_published == True)
    .order_by(BlogPost.created_at.desc())
    .limit(5)
)
popular_posts = result.scalars().all()

# Joins and aggregations
result = await db.execute(
    select(User.username, func.count(BlogPost.id).label('post_count'))
    .join(BlogPost)
    .group_by(User.username)
)
posts_by_author = result.all()

# In database.py
engine = create_async_engine(
    SQLALCHEMY_DATABASE_URL,
    echo=True  # Logs all SQL queries
)