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.
- Setup database layer - Initialize SQLAlchemy + Pydantic + Alembic from scratch
- Define models - Create SQLAlchemy models with Pydantic schemas
- Create migration - Generate and manage Alembic migrations
- Query patterns - Async CRUD, joins, eager loading, optimization
- Full implementation - Complete database layer for a feature
Auto-detection triggers (use this skill when user mentions):
- database, db, sqlalchemy, postgres, postgresql
- model, migration, alembic
- repository, crud, query
- async session, connection pool
| Reference |
Purpose |
| references/best-practices.md |
Production patterns, security, performance |
| references/patterns.md |
Repository, Unit of Work, common queries |
| references/async-patterns.md |
Async session management, FastAPI integration |
|
|
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"
1---2name: sqlalchemy-postgres3description: Expert guidance for SQLAlchemy 2.0 + Pydantic + PostgreSQL. Use when setting up database layers, defining models, creating migrations, or any database-related work. Automatically activated for DB tasks.4---5
6<essential_principles>
7## SQLAlchemy 2.0 + Pydantic + PostgreSQL Best Practices
8
9This skill provides expert guidance for building production-ready database layers.
10
11### Stack
12- **SQLAlchemy 2.0** with async support (asyncpg driver)
13- **Pydantic v2** for validation and serialization
14- **Alembic** for migrations
15- **PostgreSQL** only
16
17### Core Principles
18
19**1. Separation of Concerns**
20```
21models/ # SQLAlchemy ORM models (database layer)
22schemas/ # Pydantic schemas (API layer)
23repositories/ # Data access patterns
24services/ # Business logic
25```
26
27**2. Type Safety First**
28Always use SQLAlchemy 2.0 style with `Mapped[]` type annotations:
29```python
30from sqlalchemy.orm import Mapped, mapped_column
31
32class User(Base):
33 __tablename__ = "users"
34 id: Mapped[int] = mapped_column(primary_key=True)
35 name: Mapped[str] = mapped_column(String(100))
36```
37
38**3. Async by Default**
39Use async engine and sessions for FastAPI:
40```python
41from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
42engine = create_async_engine("postgresql+asyncpg://...")
43```
44
45**4. Pydantic-SQLAlchemy Bridge**
46Keep models and schemas separate but mappable:
47```python
48# Schema reads from ORM
49class UserRead(BaseModel):
50 model_config = ConfigDict(from_attributes=True)
51```
52
53**5. Repository Pattern**
54Abstract database operations for testability and clean code.
55</essential_principles>
56
57<intake>
58What do you need help with?
59
601. **Setup database layer** - Initialize SQLAlchemy + Pydantic + Alembic from scratch
612. **Define models** - Create SQLAlchemy models with Pydantic schemas
623. **Create migration** - Generate and manage Alembic migrations
634. **Query patterns** - Async CRUD, joins, eager loading, optimization
645. **Full implementation** - Complete database layer for a feature
65</intake>
66
67<routing>
68| Response | Workflow |
69|----------|----------|
70| 1, "setup", "initialize", "start" | workflows/setup-database.md |
71| 2, "model", "define", "create model" | workflows/define-models.md |
72| 3, "migration", "alembic", "schema change" | workflows/create-migration.md |
73| 4, "query", "crud", "repository" | workflows/query-patterns.md |
74| 5, "full", "complete", "feature" | Run setup → define-models → create-migration |
75
76**Auto-detection triggers (use this skill when user mentions):**
77- database, db, sqlalchemy, postgres, postgresql
78- model, migration, alembic
79- repository, crud, query
80- async session, connection pool
81</routing>
82
83<reference_index>
84## Domain Knowledge
85
86| Reference | Purpose |
87|-----------|---------|
88| references/best-practices.md | Production patterns, security, performance |
89| references/patterns.md | Repository, Unit of Work, common queries |
90| references/async-patterns.md | Async session management, FastAPI integration |
91</reference_index>
92
93<workflows_index>
94| Workflow | Purpose |
95|----------|---------|
96| workflows/setup-database.md | Initialize complete database layer |
97| workflows/define-models.md | Create models + schemas + relationships |
98| workflows/create-migration.md | Alembic migration workflow |
99| workflows/query-patterns.md | CRUD operations and optimization |
100</workflows_index>
101
102<quick_reference>
103## File Structure
104```
105src/
106├── db/
107│ ├── __init__.py
108│ ├── base.py # DeclarativeBase
109│ ├── session.py # Engine + async session factory
110│ └── dependencies.py # FastAPI dependency
111├── models/
112│ ├── __init__.py
113│ └── user.py # SQLAlchemy models
114├── schemas/
115│ ├── __init__.py
116│ └── user.py # Pydantic schemas
117├── repositories/
118│ ├── __init__.py
119│ ├── base.py # Generic repository
120│ └── user.py # User repository
121└── alembic/
122 ├── alembic.ini
123 ├── env.py
124 └── versions/
125```
126
127## Essential Imports
128```python
129# Models
130from sqlalchemy import String, Integer, ForeignKey, DateTime
131from sqlalchemy.orm import Mapped, mapped_column, relationship, DeclarativeBase
132
133# Async
134from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
135
136# Pydantic
137from pydantic import BaseModel, ConfigDict, Field
138```
139
140## Connection String
141```python
142# PostgreSQL async
143DATABASE_URL = "postgresql+asyncpg://user:pass@localhost:5432/dbname"
144```
145</quick_reference>
146
147<success_criteria>
148Database layer is complete when:
149- [ ] Async engine and session factory configured
150- [ ] Base model with common fields (id, created_at, updated_at)
151- [ ] Models use Mapped[] type annotations
152- [ ] Pydantic schemas with from_attributes=True
153- [ ] Alembic configured for async
154- [ ] Repository pattern implemented
155- [ ] FastAPI dependency for session injection
156- [ ] Connection pooling configured for production
157</success_criteria>
majiayu000