Back to Details

sqlalchemy-postgres

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese
<essential_principles>
<essential_principles>

SQLAlchemy 2.0 + Pydantic + PostgreSQL Best Practices

SQLAlchemy 2.0 + Pydantic + PostgreSQL 最佳实践

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
  • SQLAlchemy 2.0（支持异步，使用asyncpg驱动）
  • Pydantic v2（用于验证与序列化）
  • Alembic（用于数据库迁移）
  • 仅支持PostgreSQL

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:
python
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:
python
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:
python
undefined
1. 关注点分离
models/       # SQLAlchemy ORM models (database layer)
schemas/      # Pydantic schemas (API layer)
repositories/ # Data access patterns
services/     # Business logic
2. 类型安全优先 始终使用带
Mapped[]
类型注解的SQLAlchemy 2.0风格：
python
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. 默认使用异步 为FastAPI使用异步引擎和会话：
python
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
engine = create_async_engine("postgresql+asyncpg://...")
4. Pydantic与SQLAlchemy的桥接 保持模型与模式分离但可映射：
python
undefined

Schema reads from ORM

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>
| Response | Workflow |
|----------|----------|
| 1, "setup", "initialize", "start" | workflows/setup-database.md |
| 2, "model", "define", "create model" | workflows/define-models.md |
| 3, "migration", "alembic", "schema change" | workflows/create-migration.md |
| 4, "query", "crud", "repository" | workflows/query-patterns.md |
| 5, "full", "complete", "feature" | Run setup → define-models → create-migration |

**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>
class UserRead(BaseModel): model_config = ConfigDict(from_attributes=True)

**5. 仓库模式**
抽象数据库操作以提升可测试性并保持代码整洁。
</essential_principles>

<intake>
您需要哪方面的帮助？

1. **搭建数据库层** - 从零开始初始化SQLAlchemy + Pydantic + Alembic
2. **定义模型** - 创建SQLAlchemy模型及对应的Pydantic模式
3. **创建迁移** - 生成并管理Alembic迁移
4. **查询模式** - 异步CRUD、关联查询、预加载、性能优化
5. **完整实现** - 为某功能构建完整的数据库层
</intake>

<routing>
| 响应关键词 | 工作流 |
|----------|----------|
| 1, "setup", "initialize", "start" | workflows/setup-database.md |
| 2, "model", "define", "create model" | workflows/define-models.md |
| 3, "migration", "alembic", "schema change" | workflows/create-migration.md |
| 4, "query", "crud", "repository" | workflows/query-patterns.md |
| 5, "full", "complete", "feature" | 执行 setup → define-models → create-migration |

**自动触发条件（当用户提及以下内容时启用本技能）：**
- database, db, sqlalchemy, postgres, postgresql
- model, migration, alembic
- repository, crud, query
- async session, connection pool
</routing>

<reference_index>

Domain Knowledge

领域知识

ReferencePurpose
references/best-practices.mdProduction patterns, security, performance
references/patterns.mdRepository, Unit of Work, common queries
references/async-patterns.mdAsync session management, FastAPI integration
</reference_index>
<workflows_index>
WorkflowPurpose
workflows/setup-database.mdInitialize complete database layer
workflows/define-models.mdCreate models + schemas + relationships
workflows/create-migration.mdAlembic migration workflow
workflows/query-patterns.mdCRUD operations and optimization
</workflows_index>
<quick_reference>
参考文档用途
references/best-practices.md生产模式、安全、性能相关内容
references/patterns.md仓库模式、工作单元、通用查询
references/async-patterns.md异步会话管理、FastAPI集成
</reference_index>
<workflows_index>
工作流用途
workflows/setup-database.md初始化完整的数据库层
workflows/define-models.md创建模型 + 模式 + 关联关系
workflows/create-migration.mdAlembic迁移工作流
workflows/query-patterns.mdCRUD操作与性能优化
</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/
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

必备导入

python
undefined
python
undefined

Models

Models

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

Async

Async

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

Pydantic

Pydantic

from pydantic import BaseModel, ConfigDict, Field
undefined
from pydantic import BaseModel, ConfigDict, Field
undefined

Connection String

连接字符串

python
undefined
python
undefined

PostgreSQL async

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>
DATABASE_URL = "postgresql+asyncpg://user:pass@localhost:5432/dbname"
</quick_reference>

<success_criteria>
数据库层搭建完成的标准：
- [ ] 已配置异步引擎与会话工厂
- [ ] 包含通用字段（id、created_at、updated_at）的基础模型
- [ ] 模型使用Mapped[]类型注解
- [ ] Pydantic模式设置from_attributes=True
- [ ] 已为异步环境配置Alembic
- [ ] 已实现仓库模式
- [ ] 已为FastAPI配置会话注入依赖
- [ ] 已为生产环境配置连接池
</success_criteria>