Sqlalchemy Postgres

<essential_principles>

SQLAlchemy 2.0 + Pydantic + PostgreSQL Best Practices

This skill provides expert guidance for building production-ready database layers.

Stack

• SQLAlchemy 2.0 with async support (asyncpg driver)
• Pydantic v2 for validation and serialization
• Alembic for migrations
• PostgreSQL only

Core Principles

1. Separation of Concerns

models/       # SQLAlchemy ORM models (database layer)
schemas/      # Pydantic schemas (API layer)
repositories/ # Data access patterns
services/     # Business logic

2. Type Safety First Always use SQLAlchemy 2.0 style with Mapped[] type annotations:

from sqlalchemy.orm import Mapped, mapped_column

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))

3. Async by Default Use async engine and sessions for FastAPI:

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
engine = create_async_engine("postgresql+asyncpg://...")

4. Pydantic-SQLAlchemy Bridge Keep models and schemas separate but mappable:

# Schema reads from ORM
class UserRead(BaseModel):
    model_config = ConfigDict(from_attributes=True)

5. Repository Pattern Abstract database operations for testability and clean code. </essential_principles>

<intake> What do you need help with?

1. Setup database layer - Initialize SQLAlchemy + Pydantic + Alembic from scratch
2. Define models - Create SQLAlchemy models with Pydantic schemas
3. Create migration - Generate and manage Alembic migrations
4. Query patterns - Async CRUD, joins, eager loading, optimization
5. Full implementation - Complete database layer for a feature </intake>

<routing>

Auto-detection triggers (use this skill when user mentions):

• database, db, sqlalchemy, postgres, postgresql
• model, migration, alembic
• repository, crud, query
• async session, connection pool </routing>

<reference_index>

Domain Knowledge

<workflows_index>

<quick_reference>

File Structure

src/
├── db/
│   ├── __init__.py
│   ├── base.py          # DeclarativeBase
│   ├── session.py       # Engine + async session factory
│   └── dependencies.py  # FastAPI dependency
├── models/
│   ├── __init__.py
│   └── user.py          # SQLAlchemy models
├── schemas/
│   ├── __init__.py
│   └── user.py          # Pydantic schemas
├── repositories/
│   ├── __init__.py
│   ├── base.py          # Generic repository
│   └── user.py          # User repository
└── alembic/
    ├── alembic.ini
    ├── env.py
    └── versions/

Essential Imports

# Models
from sqlalchemy import String, Integer, ForeignKey, DateTime
from sqlalchemy.orm import Mapped, mapped_column, relationship, DeclarativeBase

# Async
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker

# Pydantic
from pydantic import BaseModel, ConfigDict, Field

Connection String

# PostgreSQL async
DATABASE_URL = "postgresql+asyncpg://user:pass@localhost:5432/dbname"

</quick_reference>

<success_criteria> Database layer is complete when:

• [ ] Async engine and session factory configured
• [ ] Base model with common fields (id, created_at, updated_at)
• [ ] Models use Mapped[] type annotations
• [ ] Pydantic schemas with from_attributes=True
• [ ] Alembic configured for async
• [ ] Repository pattern implemented
• [ ] FastAPI dependency for session injection
• [ ] Connection pooling configured for production </success_criteria>