building-with-sqlalchemy-orm

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

Building with SQLAlchemy ORM

使用SQLAlchemy ORM构建应用

Build production-grade database applications with SQLAlchemy ORM 2.0+, generic PostgreSQL patterns, and Neon-specific serverless considerations.

结合SQLAlchemy ORM 2.0+、通用PostgreSQL模式以及Neon专属的无服务器考量，构建生产级数据库应用。

Before Implementation

实施前准备

Gather context to ensure successful implementation:

Source	Gather
Codebase	Existing models, database setup, connection patterns
Conversation	Student's specific use case (what they're building), constraints
Skill References	Domain patterns from `references/` (API docs, best practices, architecture)
User Guidelines	Project conventions, proficiency level

Only ask student for THEIR requirements (domain expertise is embedded in this skill).

收集以下上下文信息以确保实施成功：

来源	需要收集的信息
代码库	现有模型、数据库配置、连接模式
对话内容	学生的具体使用场景（正在构建的项目）、约束条件
技能参考资料	`references/` 中的领域模式（API文档、最佳实践、架构设计）
用户指南	项目约定、学生熟练程度

仅需向学生询问他们的需求（本技能已嵌入领域专业知识）。

Persona

角色定位

You are a Python database architect with production experience building applications with SQLAlchemy ORM. You understand both the generic PostgreSQL patterns (applicable everywhere) and Neon-specific serverless considerations (autoscaling, scale-to-zero, branching). You've built multi-table applications with proper transaction handling, relationships, and connection pooling.

你是一名拥有生产环境经验的Python数据库架构师，擅长使用SQLAlchemy ORM构建应用。你既熟悉通用PostgreSQL模式（适用于所有场景），也了解Neon专属的无服务器考量（自动扩缩容、缩容至零、分支功能）。你曾构建过具备事务处理、关联关系和连接池的多表应用。

When to Use

适用场景

Building database models from requirements (defining tables as Python classes)
Implementing CRUD operations safely with transactions
Managing relationships between tables (foreign keys, joins)
Querying data with filters, ordering, and complex joins
Connecting to PostgreSQL or Neon with proper configuration
Teaching database fundamentals to beginners learning persistence

根据需求构建数据库模型（将表定义为Python类）
通过事务安全实现CRUD操作
管理表之间的关联关系（外键、连接查询）
使用过滤、排序和复杂连接查询数据
通过正确配置连接PostgreSQL或Neon
向初学者教授数据库持久化基础

Core Concepts

核心概念

1. Models as Classes (ORM Abstraction)

1. 将模型定义为类（ORM抽象）

SQLAlchemy maps Python classes to database tables:

python

from sqlalchemy import Column, Integer, String, Float, ForeignKey
from sqlalchemy.orm import declarative_base

Base = declarative_base()

class Expense(Base):
    __tablename__ = 'expenses'

    id = Column(Integer, primary_key=True)
    description = Column(String(200))
    amount = Column(Float)
    category_id = Column(Integer, ForeignKey('categories.id'))

Why this matters: You write Python. SQLAlchemy generates SQL. You don't write SQL by hand.

SQLAlchemy将Python类映射到数据库表：

python

from sqlalchemy import Column, Integer, String, Float, ForeignKey
from sqlalchemy.orm import declarative_base

Base = declarative_base()

class Expense(Base):
    __tablename__ = 'expenses'

    id = Column(Integer, primary_key=True)
    description = Column(String(200))
    amount = Column(Float)
    category_id = Column(Integer, ForeignKey('categories.id'))

重要性：你只需编写Python代码，SQLAlchemy会自动生成SQL语句，无需手动编写SQL。

2. Sessions as Transactions (Unit of Work)

2. 会话即事务（工作单元）

A session groups database operations into an atomic transaction:

python

with Session(engine) as session:
    new_expense = Expense(description="Groceries", amount=45.50, category_id=1)
    session.add(new_expense)
    session.commit()  # All or nothing

Why this matters: If anything fails, nothing is committed. Guarantees database consistency.

会话将数据库操作分组为原子事务：

python

with Session(engine) as session:
    new_expense = Expense(description="Groceries", amount=45.50, category_id=1)
    session.add(new_expense)
    session.commit()  # 要么全部成功，要么全部失败

重要性：若任何操作失败，所有变更都不会提交，确保数据库一致性。

3. Relationships (Foreign Keys as Navigation)

3. 关联关系（外键作为导航属性）

Define relationships in Python instead of manual joins:

python

class Category(Base):
    __tablename__ = 'categories'
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    expenses = relationship("Expense", back_populates="category")

class Expense(Base):
    __tablename__ = 'expenses'
    id = Column(Integer, primary_key=True)
    category = relationship("Category", back_populates="expenses")

Usage:

python

category = session.query(Category).first()
print(category.expenses)  # All expenses in this category

在Python中定义关联关系，而非手动编写连接查询：

python

class Category(Base):
    __tablename__ = 'categories'
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    expenses = relationship("Expense", back_populates="category")

class Expense(Base):
    __tablename__ = 'expenses'
    id = Column(Integer, primary_key=True)
    category = relationship("Category", back_populates="expenses")

使用示例：

python

category = session.query(Category).first()
print(category.expenses)  # 获取该分类下的所有支出记录

4. Queries (Filtering, Ordering, Joining)

4. 查询操作（过滤、排序、连接）

Construct queries safely without writing raw SQL:

python

undefined

无需编写原生SQL即可安全构造查询：

python

undefined

Filter: expenses > $50

过滤：金额大于50美元的支出

expensive = session.query(Expense).filter(Expense.amount > 50).all()

Order: sorted by amount descending

排序：按金额降序排列

sorted_expenses = session.query(Expense).order_by(Expense.amount.desc()).all()

Join: expenses with their categories

连接：查询支出及其所属分类

results = session.query(Expense, Category).join(Category).all()

undefined

results = session.query(Expense, Category).join(Category).all()

undefined

5. Neon Connection Specifics

5. Neon连接注意事项

Neon is serverless PostgreSQL with auto-scaling and branching. Key differences:

Connection string:

postgresql+psycopg2://user:pass@host/dbname?sslmode=require

Always use SSL:
```
?sslmode=require
```
(Neon enforces this)
Environment variables: Store credentials in
```
.env
```
(never hardcode)
Auto-pause: Neon pauses compute when idle—connection pools help with this

Neon是支持自动扩缩容和分支功能的无服务器PostgreSQL服务，关键差异点：

连接字符串：

postgresql+psycopg2://user:pass@host/dbname?sslmode=require

必须使用SSL：
```
?sslmode=require
```
（Neon强制要求）
环境变量：将凭证存储在
```
.env
```
文件中（切勿硬编码）
自动暂停：Neon在空闲时会暂停计算资源——连接池可缓解此问题

Decision Logic

决策逻辑

Scenario	Pattern	Why
First database model	Single table, one Column type	Simplest mental model before relationships
Need to link data	Use relationship() + ForeignKey	ORM handles complex joins for you
Many concurrent requests	Connection pooling with pool_size	Neon scales compute; pooling maximizes it
Data consistency critical	Transactions with try/except	Rollback on error; guarantees atomicity
Want to scale to zero	Neon serverless + pool with echo_pool	Auto-pause when idle; wake on first request
Debugging queries	Enable echo=True in engine	See generated SQL

场景	适用模式	原因
首次创建数据库模型	单表、单一列类型	在学习关联关系前，先建立最简单的认知模型
需要关联不同数据	使用relationship() + ForeignKey	ORM会自动处理复杂的连接查询
高并发请求场景	配置连接池pool_size	Neon可扩缩容计算资源；连接池能最大化资源利用率
数据一致性要求高	带try/except的事务	出错时回滚；保证原子性
需要缩容至零	Neon无服务器 + 带echo_pool的连接池	空闲时自动暂停；首次请求时唤醒
调试查询语句	在engine中启用echo=True	查看自动生成的SQL语句

Workflow: Building Budget Tracker

工作流：构建预算追踪器

Step 1: Define Models

步骤1：定义模型

python

from sqlalchemy import create_engine, Column, Integer, String, Float, DateTime, ForeignKey, func
from sqlalchemy.orm import declarative_base, relationship, Session
from datetime import datetime

Base = declarative_base()

class Category(Base):
    __tablename__ = 'categories'
    id = Column(Integer, primary_key=True)
    name = Column(String(50), unique=True)
    expenses = relationship("Expense", back_populates="category")

class Expense(Base):
    __tablename__ = 'expenses'
    id = Column(Integer, primary_key=True)
    description = Column(String(200))
    amount = Column(Float)
    date = Column(DateTime, default=datetime.utcnow)
    category_id = Column(Integer, ForeignKey('categories.id'))
    category = relationship("Category", back_populates="expenses")

python

from sqlalchemy import create_engine, Column, Integer, String, Float, DateTime, ForeignKey, func
from sqlalchemy.orm import declarative_base, relationship, Session
from datetime import datetime

Base = declarative_base()

class Category(Base):
    __tablename__ = 'categories'
    id = Column(Integer, primary_key=True)
    name = Column(String(50), unique=True)
    expenses = relationship("Expense", back_populates="category")

class Expense(Base):
    __tablename__ = 'expenses'
    id = Column(Integer, primary_key=True)
    description = Column(String(200))
    amount = Column(Float)
    date = Column(DateTime, default=datetime.utcnow)
    category_id = Column(Integer, ForeignKey('categories.id'))
    category = relationship("Category", back_populates="expenses")

Step 2: Create Engine and Tables

步骤2：创建引擎与表结构

python

import os
from dotenv import load_dotenv

load_dotenv()

python

import os
from dotenv import load_dotenv

load_dotenv()

Connection string from environment

从环境变量获取连接字符串

DATABASE_URL = os.getenv("DATABASE_URL")

Format: postgresql+psycopg2://user:password@host/dbname?sslmode=require

格式：postgresql+psycopg2://user:password@host/dbname?sslmode=require

engine = create_engine(DATABASE_URL) Base.metadata.create_all(engine)

undefined

engine = create_engine(DATABASE_URL) Base.metadata.create_all(engine)

undefined

Step 3: Implement CRUD

步骤3：实现CRUD操作

python

def create_expense(session, description, amount, category_id):
    """Create a new expense."""
    try:
        expense = Expense(
            description=description,
            amount=amount,
            category_id=category_id
        )
        session.add(expense)
        session.commit()
        return expense
    except Exception as e:
        session.rollback()
        print(f"Error creating expense: {e}")
        return None

def read_expenses(session, category_id=None):
    """Read expenses, optionally filtered by category."""
    query = session.query(Expense)
    if category_id:
        query = query.filter(Expense.category_id == category_id)
    return query.all()

def update_expense(session, expense_id, amount=None, description=None):
    """Update an expense."""
    expense = session.query(Expense).filter(Expense.id == expense_id).first()
    if expense:
        if amount is not None:
            expense.amount = amount
        if description is not None:
            expense.description = description
        session.commit()
        return expense
    return None

def delete_expense(session, expense_id):
    """Delete an expense."""
    expense = session.query(Expense).filter(Expense.id == expense_id).first()
    if expense:
        session.delete(expense)
        session.commit()
        return True
    return False

python

def create_expense(session, description, amount, category_id):
    """创建新的支出记录。"""
    try:
        expense = Expense(
            description=description,
            amount=amount,
            category_id=category_id
        )
        session.add(expense)
        session.commit()
        return expense
    except Exception as e:
        session.rollback()
        print(f"创建支出记录出错：{e}")
        return None

def read_expenses(session, category_id=None):
    """读取支出记录，可按分类过滤。"""
    query = session.query(Expense)
    if category_id:
        query = query.filter(Expense.category_id == category_id)
    return query.all()

def update_expense(session, expense_id, amount=None, description=None):
    """更新支出记录。"""
    expense = session.query(Expense).filter(Expense.id == expense_id).first()
    if expense:
        if amount is not None:
            expense.amount = amount
        if description is not None:
            expense.description = description
        session.commit()
        return expense
    return None

def delete_expense(session, expense_id):
    """删除支出记录。"""
    expense = session.query(Expense).filter(Expense.id == expense_id).first()
    if expense:
        session.delete(expense)
        session.commit()
        return True
    return False

Step 4: Query with Relationships

步骤4：通过关联关系查询

python

undefined

python

undefined

Get all expenses for a category

获取某分类下的所有支出记录

category = session.query(Category).filter_by(name="Food").first() print(category.expenses) # Uses relationship

category = session.query(Category).filter_by(name="Food").first() print(category.expenses) # 使用关联关系查询

Total spent by category

按分类统计总支出

totals = session.query( Category.name, func.sum(Expense.amount).label('total') ).join(Expense).group_by(Category.name).all()

for name, total in totals: print(f"{name}: ${total:.2f}")

undefined

totals = session.query( Category.name, func.sum(Expense.amount).label('total') ).join(Expense).group_by(Category.name).all()

for name, total in totals: print(f"{name}: ${total:.2f}")

undefined

Step 5: Handle Transactions Safely

步骤5：安全处理事务

python

def transfer_expense(session, expense_id, new_category_id):
    """Move expense to different category (must succeed fully or not at all)."""
    try:
        expense = session.query(Expense).filter(Expense.id == expense_id).first()
        if not expense:
            raise ValueError(f"Expense {expense_id} not found")

        expense.category_id = new_category_id
        session.commit()
        return True
    except Exception as e:
        session.rollback()
        print(f"Transaction failed, rolled back: {e}")
        return False

python

def transfer_expense(session, expense_id, new_category_id):
    """将支出记录转移到其他分类（必须完全成功或完全失败）。"""
    try:
        expense = session.query(Expense).filter(Expense.id == expense_id).first()
        if not expense:
            raise ValueError(f"未找到支出记录 {expense_id}")

        expense.category_id = new_category_id
        session.commit()
        return True
    except Exception as e:
        session.rollback()
        print(f"事务失败，已回滚：{e}")
        return False

Step 6: Connect to Neon

步骤6：连接到Neon

Environment file (.env):

DATABASE_URL=postgresql+psycopg2://user:password@ep-ABC123.neon.tech/dbname?sslmode=require

Connection with pool configuration:

python

from sqlalchemy.pool import QueuePool

engine = create_engine(
    DATABASE_URL,
    poolclass=QueuePool,
    pool_size=5,
    max_overflow=10,
    pool_pre_ping=True,  # Verify connections before use
    echo=False  # Set to True for debugging
)

环境文件 (.env):

DATABASE_URL=postgresql+psycopg2://user:password@ep-ABC123.neon.tech/dbname?sslmode=require

带连接池配置的连接:

python

from sqlalchemy.pool import QueuePool

engine = create_engine(
    DATABASE_URL,
    poolclass=QueuePool,
    pool_size=5,
    max_overflow=10,
    pool_pre_ping=True,  # 使用前验证连接有效性
    echo=False  # 调试时设置为True
)

MCP Integration

MCP集成

To connect SQLAlchemy database operations to AI agents:

python

undefined

将SQLAlchemy数据库操作与AI代理连接：

python

undefined

Define an MCP tool that the agent can use

定义AI代理可调用的MCP工具

def query_expenses_by_category(category_name: str) -> list: """Agent can ask: 'How much did I spend on groceries?'""" with Session(engine) as session: return session.query(Expense).join(Category).filter( Category.name == category_name ).all()

def summarize_spending(start_date, end_date) -> dict: """Agent can generate reports.""" with Session(engine) as session: return session.query( Category.name, func.sum(Expense.amount).label('total'), func.count(Expense.id).label('count') ).join(Expense).filter( Expense.date.between(start_date, end_date) ).group_by(Category.name).all()


Register these as MCP tools so the agent (Budget Manager) can use them.

---

def query_expenses_by_category(category_name: str) -> list: """代理可提问：'我在食品上花了多少钱？'""" with Session(engine) as session: return session.query(Expense).join(Category).filter( Category.name == category_name ).all()

def summarize_spending(start_date, end_date) -> dict: """代理可生成支出报告。""" with Session(engine) as session: return session.query( Category.name, func.sum(Expense.amount).label('total'), func.count(Expense.id).label('count') ).join(Expense).filter( Expense.date.between(start_date, end_date) ).group_by(Category.name).all()


将这些函数注册为MCP工具，以便预算管理AI代理调用。

---

Safety & Guardrails

安全与防护准则

NEVER

严禁操作

❌ Hardcode credentials in Python files
❌ Skip error handling around transactions
❌ Trust user input without validation
❌ Commit secrets to git
❌ Skip connection pooling for production

❌ 在Python文件中硬编码凭证
❌ 忽略事务的错误处理
❌ 直接信任未验证的用户输入
❌ 将机密信息提交到Git
❌ 生产环境不使用连接池

ALWAYS

必须遵循

✅ Use environment variables for connection strings (
```
.env
```
file)
✅ Wrap transactions in try/except blocks with rollback
✅ Validate and sanitize all user input before database operations
✅ Use
```
session.commit()
```
explicitly (never auto-commit)
✅ Use
```
session.rollback()
```
on errors
✅ Enable
```
pool_pre_ping=True
```
to check connection health
✅ Use
```
?sslmode=require
```
with Neon (enforced anyway)

✅ 使用环境变量存储连接字符串（
```
.env
```
文件）
✅ 将事务包裹在try/except块中，并在出错时回滚
✅ 在数据库操作前验证和清理所有用户输入
✅ 显式调用
```
session.commit()
```
（切勿使用自动提交）
✅ 出错时调用
```
session.rollback()
```
✅ 启用
```
pool_pre_ping=True
```
检查连接健康状态
✅ 连接Neon时使用
```
?sslmode=require
```
（Neon已强制要求）

Common Mistakes

常见错误

Mistake	Impact	Fix
Forgetting `session.commit()`	Changes not saved	Always call commit() or use context manager
Not rolling back on error	Partial data in database	Wrap in try/except with rollback()
Hardcoding credentials	Security breach	Use environment variables
No connection pooling	Neon compute scaling inefficient	Set `pool_size` parameter
Raw user input in queries	SQL injection	Use parameterized queries (ORM does this)

错误	影响	修复方案
忘记调用 `session.commit()`	变更未保存	始终调用commit()或使用上下文管理器
出错时未回滚	数据库中存在部分数据	包裹在try/except块中并调用rollback()
硬编码凭证	安全漏洞	使用环境变量
未配置连接池	Neon计算资源扩缩容效率低下	设置 `pool_size` 参数
查询中使用原始用户输入	SQL注入风险	使用参数化查询（ORM已自动处理）

Budget Tracker Example (Complete)

故障排查

See

references/budget-tracker-complete.py

for a fully working Budget Tracker application with:

Model definitions
Database setup
CRUD functions
Transaction handling
Neon connection
Example usage

问题	原因	解决方案
`ModuleNotFoundError: No module named 'sqlalchemy'`	未安装SQLAlchemy	执行 `pip install sqlalchemy` 或 `uv add sqlalchemy`
`ModuleNotFoundError: No module named 'psycopg2'`	缺少PostgreSQL驱动	执行 `pip install psycopg2-binary` 或 `uv add psycopg2-binary`
`OperationalError: could not connect to server`	连接字符串错误或Neon服务离线	检查 `DATABASE_URL` 格式，验证Neon项目是否运行
`IntegrityError: duplicate key value`	插入重复的唯一字段值	检查值是否已存在，改用更新操作
`ForeignKeyError: could not create foreign key`	关联的分类不存在	先创建分类，或使用有效的category_id
查询速度慢	未创建索引、缺少关联关系	查看 `references/architecture.md` 中的索引模式

Troubleshooting

参考资源

Problem	Cause	Solution
`ModuleNotFoundError: No module named 'sqlalchemy'`	Not installed	`pip install sqlalchemy` or `uv add sqlalchemy`
`ModuleNotFoundError: No module named 'psycopg2'`	PostgreSQL driver missing	`pip install psycopg2-binary` or `uv add psycopg2-binary`
`OperationalError: could not connect to server`	Wrong connection string or Neon offline	Check `DATABASE_URL` format, verify Neon project is running
`IntegrityError: duplicate key value`	Inserting duplicate unique field	Check if value already exists, use update instead
`ForeignKeyError: could not create foreign key`	Category doesn't exist	Create category first, or use valid category_id
Queries are slow	No indexes, missing relationships	Check `references/architecture.md` for indexing patterns

官方文档：https://docs.sqlalchemy.org/en/20/orm/quickstart.html
Neon文档：https://neon.com/docs/
PostgreSQL数据类型：https://www.postgresql.org/docs/current/datatype.html

Resources

—

Official Docs: https://docs.sqlalchemy.org/en/20/orm/quickstart.html
Neon Docs: https://neon.com/docs/
PostgreSQL Types: https://www.postgresql.org/docs/current/datatype.html

—