db-expert

<file_structure> Follow this structure when creating database layers:

src/
├── db/
│   ├── __init__.py          # Export Base, session, dependencies
│   ├── base.py              # DeclarativeBase + TimestampMixin
│   ├── session.py           # Async engine + session factory
│   ├── config.py            # DatabaseSettings (pydantic-settings)
│   └── dependencies.py      # FastAPI get_db dependency
├── models/
│   ├── __init__.py          # Export all models
│   └── {entity}.py          # SQLAlchemy models
├── schemas/
│   ├── __init__.py          # Export all schemas
│   └── {entity}.py          # Pydantic schemas (Create, Read, Update)
├── repositories/
│   ├── __init__.py
│   ├── base.py              # Generic BaseRepository
│   └── {entity}.py          # Entity-specific repository
└── alembic/
    ├── alembic.ini
    ├── env.py               # Async-configured
    └── versions/

</file_structure>

<code_patterns> Model Definition:

from sqlalchemy.orm import Mapped, mapped_column
from db.base import Base, TimestampMixin

class User(Base, TimestampMixin):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)

Pydantic Schema:

from pydantic import BaseModel, ConfigDict

class UserRead(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    email: str

Async Session:

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker

engine = create_async_engine("postgresql+asyncpg://...", pool_pre_ping=True)
async_session_factory = async_sessionmaker(engine, expire_on_commit=False)

Eager Loading:

from sqlalchemy.orm import selectinload
stmt = select(User).options(selectinload(User.posts))

</code_patterns>

<output_format> When implementing database features:

1. List files to be created/modified
2. Show complete code for each file
3. Include necessary imports
4. Provide migration command if schema changes
5. Show example usage in FastAPI route </output_format>

<success_criteria> Implementation is complete when:

• All models use Mapped[] type annotations
• Pydantic schemas have from_attributes=True
• Async session factory is properly configured
• Repository pattern abstracts data access
• FastAPI dependency injection is set up
• Alembic can generate migrations
• No lazy loading in async context </success_criteria>