database-patterns

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

Database Patterns

数据库设计模式

Forward-only migrations, explicit transactions, measured optimization.

仅向前迁移、显式事务、可衡量的优化。

Migrations

迁移

Forward-only. No rollbacks. Maintain backward compatibility:

sql

-- Add nullable column (backward compatible)
ALTER TABLE users ADD COLUMN phone VARCHAR(20);

-- Later: make required after backfill
ALTER TABLE users ALTER COLUMN phone SET NOT NULL;

Break large changes into smaller steps. Use feature flags during transitions.

仅向前迁移，不回滚，保持向后兼容性：

sql

-- 添加可为空的列（向后兼容）
ALTER TABLE users ADD COLUMN phone VARCHAR(20);

-- 后续：回填后设置为必填
ALTER TABLE users ALTER COLUMN phone SET NOT NULL;

将大型变更拆分为多个小步骤，过渡期间使用功能开关。

Query Optimization

查询优化

Always check execution plans before optimizing:

sql

EXPLAIN ANALYZE SELECT * FROM orders WHERE user_id = 123;

Index based on actual query patterns:

sql

-- Composite for common query
CREATE INDEX idx_orders_user_date ON orders (user_id, created_at DESC);

-- Partial for filtered queries
CREATE INDEX idx_orders_pending ON orders (status) WHERE status = 'pending';

Monitor unused indexes. Remove if
idx_scan < 100
.

优化前务必检查执行计划：

sql

EXPLAIN ANALYZE SELECT * FROM orders WHERE user_id = 123;

基于实际查询模式创建索引：

sql

-- 针对常见查询创建复合索引
CREATE INDEX idx_orders_user_date ON orders (user_id, created_at DESC);

-- 针对过滤查询创建部分索引
CREATE INDEX idx_orders_pending ON orders (status) WHERE status = 'pending';

监控未使用的索引，若
idx_scan < 100
则删除。

N+1 Prevention

N+1 查询问题规避

Always eager load in loops:

python

undefined

循环中务必预加载关联数据：

python

undefined

Good

Bad (N+1)

不推荐写法（存在N+1问题）

users = User.query.all() for user in users: print(user.posts) # N queries!

undefined

users = User.query.all() for user in users: print(user.posts) # 会执行N次查询！

undefined

Transactions

事务处理

Scope to single business operation. Keep short:

python

async with db.transaction():
    order = await create_order(data)
    await update_inventory(order.items)
    # Commit on exit

事务范围限定为单个业务操作，保持事务简短：

python

async with db.transaction():
    order = await create_order(data)
    await update_inventory(order.items)
    # 退出时自动提交

OUTSIDE transaction: send emails, call external APIs

事务外部：发送邮件、调用外部API

await send_confirmation(order)


**Never hold transactions during external calls.**

await send_confirmation(order)


**外部调用期间切勿持有事务。**

Connection Pooling

连接池配置

python

undefined

python

undefined

Size based on measured peak concurrency

根据实测峰值并发量设置大小

create_engine( url, pool_size=15, # Based on load testing max_overflow=5, # Burst capacity pool_timeout=30, # Fail fast pool_recycle=3600, # Prevent stale connections pool_pre_ping=True # Validate before use )


**Monitor utilization. Alert at 80%.**

create_engine( url, pool_size=15, # 基于负载测试结果 max_overflow=5, # 突发容量 pool_timeout=30, # 快速失败 pool_recycle=3600, # 防止连接失效 pool_pre_ping=True # 使用前验证连接有效性 )


**监控连接池使用率，使用率达80%时触发告警。**

Data Validation

数据验证

Validate at boundaries, not just in database:

python

undefined

在边界处进行验证，而非仅依赖数据库：

python

undefined

Validate input before INSERT

插入前验证输入

validated = CreateUserSchema.parse(input) if await email_exists(validated.email): raise ValidationError("Email taken")

validated = CreateUserSchema.parse(input) if await email_exists(validated.email): raise ValidationError("邮箱已被占用")

Validate output after retrieval (detect corruption)

检索后验证输出（检测数据损坏）

return UserOutputSchema.parse(row)

undefined

return UserOutputSchema.parse(row)

undefined

Anti-Patterns

反模式

Rollback migrations (use forward-only)
Indexes without query pattern analysis
N+1 queries in loops
Long-running transactions with external calls
Relying only on DB constraints for validation
Default pool settings without measurement

回滚迁移（应使用仅向前迁移）
未分析查询模式就创建索引
循环中存在N+1查询
外部调用期间持有长事务
仅依赖数据库约束进行验证
使用默认连接池设置而未进行实测

References

参考资料

audit-logging.md - Immutable audit trails

audit-logging.md - 不可变审计日志