几乎所有 Python 应用都需要持久化数据。本文从 SQLite 起步,到 SQLAlchemy ORM,再到异步数据库访问,给你一个完整的数据库技术栈。
目录
- 数据库选型指南
- SQLite:零配置数据库
- PostgreSQL 与 psycopg2
- SQLAlchemy ORM 核心
- 关系与查询
- 异步数据库:asyncpg
- 数据库迁移:Alembic
- SQLAlchemy 2.0 新范式
- 连接池深度配置
- ORM 性能优化实战
- 数据库事务隔离级别与死锁处理
- 测试数据库策略
- 生产环境最佳实践
1. 数据库选型指南
| 数据库 | 适用场景 | Python 驱动 |
|---|---|---|
| SQLite | 单机应用、测试、嵌入式 | sqlite3(内置) |
| PostgreSQL | 生产应用、复杂查询 | psycopg2, asyncpg |
| MySQL | Web 应用、LAMP 栈 | PyMySQL, mysqlclient |
| MongoDB | 文档型数据 | pymongo |
| Redis | 缓存、消息队列 | redis-py |
2. SQLite:零配置数据库
2.1 基础 CRUD
import sqlite3
from pathlib import Path
DB_PATH = Path("app.db")
def init_db():
with sqlite3.connect(DB_PATH) as conn:
conn.execute("""
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
email TEXT UNIQUE,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
def add_user(name: str, email: str):
with sqlite3.connect(DB_PATH) as conn:
conn.execute("INSERT INTO users (name, email) VALUES (?, ?)", (name, email))
conn.commit()
def get_users():
with sqlite3.connect(DB_PATH) as conn:
conn.row_factory = sqlite3.Row # 返回字典式结果
return conn.execute("SELECT * FROM users").fetchall()
# 使用
init_db()
add_user("Alice", "alice@example.com")
for user in get_users():
print(dict(user))
2.2 SQLite 高级特性
# 事务控制
with sqlite3.connect(DB_PATH) as conn:
try:
conn.execute("INSERT INTO users (name) VALUES (?)", ("Bob",))
conn.execute("INSERT INTO invalid_table ...") # 会失败
conn.commit()
except sqlite3.Error:
conn.rollback()
# 返回字典结果
conn.row_factory = lambda c, r: {col[0]: r[idx] for idx, col in enumerate(c.description)}
# FTS5 全文搜索(需要编译时启用)
conn.execute("CREATE VIRTUAL TABLE docs USING fts5(title, content)")
3. PostgreSQL 与 psycopg2
# pip install psycopg2-binary
import psycopg2
from psycopg2.extras import RealDictCursor
conn = psycopg2.connect(
host="localhost",
database="mydb",
user="myuser",
password="mypass",
)
with conn.cursor(cursor_factory=RealDictCursor) as cur:
cur.execute("SELECT * FROM users WHERE age > %s", (18,))
for row in cur.fetchall():
print(dict(row))
conn.close()
4. SQLAlchemy ORM 核心
4.1 模型定义
# pip install sqlalchemy
from sqlalchemy import create_engine, Column, Integer, String, DateTime, ForeignKey
from sqlalchemy.orm import declarative_base, relationship, sessionmaker
from datetime import datetime
Base = declarative_base()
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True)
name = Column(String(100), nullable=False)
email = Column(String(255), unique=True)
created_at = Column(DateTime, default=datetime.utcnow)
posts = relationship("Post", back_populates="author")
class Post(Base):
__tablename__ = "posts"
id = Column(Integer, primary_key=True)
title = Column(String(200), nullable=False)
content = Column(String)
author_id = Column(Integer, ForeignKey("users.id"))
author = relationship("User", back_populates="posts")
# 创建表
engine = create_engine("sqlite:///app.db", echo=True)
Base.metadata.create_all(engine)
# Session
Session = sessionmaker(bind=engine)
4.2 CRUD 操作
# 创建
with Session() as session:
user = User(name="Alice", email="alice@example.com")
session.add(user)
session.commit()
print(user.id) # 自动获取自增 ID
# 查询
with Session() as session:
# 主键查询
user = session.get(User, 1)
# 条件查询
users = session.query(User).filter(User.name.like("A%")).all()
# 分页
page = session.query(User).offset(10).limit(10).all()
# 更新
with Session() as session:
user = session.get(User, 1)
user.name = "New Name"
session.commit()
# 删除
with Session() as session:
user = session.get(User, 1)
session.delete(user)
session.commit()
5. 关系与查询
5.1 关联查询
# 一对多:用户 - 文章
with Session() as session:
user = session.get(User, 1)
for post in user.posts:
print(post.title)
# 预加载(避免 N+1 查询)
from sqlalchemy.orm import joinedload
users = session.query(User).options(joinedload(User.posts)).all()
# 聚合查询
from sqlalchemy import func
result = session.query(
User.name,
func.count(Post.id).label("post_count")
).join(Post).group_by(User.id).all()
6. 异步数据库:asyncpg
# pip install asyncpg
import asyncpg
import asyncio
async def main():
conn = await asyncpg.connect("postgresql://user:pass@localhost/db")
# 查询
rows = await conn.fetch("SELECT * FROM users WHERE age > $1", 18)
for row in rows:
print(row["name"])
# 事务
async with conn.transaction():
await conn.execute("INSERT INTO users (name) VALUES ($1)", "Bob")
await conn.close()
asyncio.run(main())
SQLAlchemy 异步版本
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
async_engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")
AsyncSessionLocal = sessionmaker(async_engine, class_=AsyncSession)
async def get_users():
async with AsyncSessionLocal() as session:
result = await session.execute(select(User).where(User.age > 18))
return result.scalars().all()
7. 数据库迁移:Alembic
# 安装
pip install alembic
# 初始化
alembic init migrations
# 修改 alembic.ini → sqlalchemy.url = postgresql://user:pass@localhost/db
# 修改 migrations/env.py → 导入你的 Base
# 生成迁移脚本
alembic revision --autogenerate -m "create users table"
# 升级
alembic upgrade head
# 降级
alembic downgrade -1
8. SQLAlchemy 2.0 新范式
SQLAlchemy 2.0 是一次重大升级,彻底革新了声明式定义和查询风格。新版本要求使用 mapped_column() 替代 Column(),天然集成 Python 类型提示,让 IDE 自动补全和静态类型检查器(如 mypy)可以正确推断模型字段类型。这种方式解决了旧版 ORM 中类型系统与运行时行为分离的痛点,使得大型项目中的模型定义更加清晰、可维护。此外,查询语法全面转向函数式风格,select()、insert()、update() 等显式构造取代了隐式的 session.query() 链式调用,语义更加明确。
from sqlalchemy.orm import DeclarativeBase, mapped_column, relationship
from sqlalchemy import String, ForeignKey, select
from typing import List, Optional
class Base(DeclarativeBase):
pass
class User(Base):
__tablename__ = "users"
id: int = mapped_column(primary_key=True)
name: str = mapped_column(String(100), nullable=False)
email: Optional[str] = mapped_column(String(255), unique=True)
posts: List["Post"] = relationship(back_populates="author")
class Post(Base):
__tablename__ = "posts"
id: int = mapped_column(primary_key=True)
title: str = mapped_column(String(200))
author_id: int = mapped_column(ForeignKey("users.id"))
author: "User" = relationship(back_populates="posts")
# 2.0 风格查询:显式 select()
from sqlalchemy import select
with Session() as session:
stmt = select(User).where(User.name == "Alice")
result = session.execute(stmt)
alice = result.scalar_one_or_none()
# 批量插入:insert() 构造器
from sqlalchemy import insert
with Session() as session:
session.execute(
insert(User),
[
{"name": "Bob", "email": "bob@test.com"},
{"name": "Carol", "email": "carol@test.com"},
],
)
session.commit()
SQLAlchemy 2.0 还改进了异步支持,AsyncSession 与同步 Session 的 API 更加一致。新范式虽然增加了一定的学习成本,但从类型安全和代码清晰度来看,迁移到 2.0 是长线投入。如果你正在用 FastAPI 或任何注重类型提示的框架开发,强烈推荐直接采用 2.0 语法起步。更多 Web 框架相关内容可参考 https://plumephp.com/python-web-frameworks/。
9. 连接池深度配置
生产环境中,数据库连接池是最常被忽视却最关键的组件之一。SQLAlchemy 默认使用 QueuePool,但如果连接参数未经调优,应用很容易在高并发下出现「连接池耗尽」错误,或者因网络中断导致持有僵尸连接。正确配置连接池的核心是理解 pool_size、max_overflow、pool_recycle 和 pool_pre_ping 这四个参数。pool_size 设置常驻连接数,max_overflow 控制高峰时额外创建的连接上限,pool_recycle 防止数据库端超时断开而客户端仍持有的过期连接,pool_pre_ping 则在借用连接前发送一次轻量检测,自动剔除失效连接。一个经过调优的连接池可以显著降低连接建立与销毁的开销,提升整体吞吐量。
from sqlalchemy import create_engine
from sqlalchemy.pool import QueuePool
# 深度调优的连接池配置
engine = create_engine(
"postgresql://user:pass@localhost/mydb",
poolclass=QueuePool,
pool_size=10, # 常驻连接数
max_overflow=20, # 最大额外连接数
pool_recycle=3600, # 1小时后回收连接
pool_pre_ping=True, # 借用前检测连接是否存活
pool_timeout=30, # 获取连接的最大等待秒数
echo=False,
)
# 连接池事件监听:用于调试连接泄漏
from sqlalchemy import event
@event.listens_for(engine, "checkout")
def on_checkout(dbapi_conn, connection_record, connection_proxy):
print(f"Connection checked out: {connection_record}")
@event.listens_for(engine, "checkin")
def on_checkin(dbapi_conn, connection_record):
print(f"Connection returned to pool: {connection_record}")
当你的应用部署到容器或 Kubernetes 环境时,连接池的行为还会受到 pod 数量和负载均衡策略的影响。横向扩展多个实例时,每个实例的 pool_size 乘以实例数量不应超过数据库的最大连接限制。为了更深入了解部署层面的数据库配置,推荐阅读 https://plumephp.com/python-deployment/。
10. ORM 性能优化实战
ORM 的便利性往往以性能为代价,其中最臭名昭著的就是「N+1 查询问题」:遍历一个对象列表并访问关联属性时,ORM 会对每个对象再发一次查询。SQLAlchemy 提供了多种急切加载策略来解决这个问题。joinedload 使用 JOIN 一次性拉取主表和关联表,适合一对多关系;selectinload 则使用 IN 子句分两次查询,在多对多或一对多数据量大的场景中性能更优。除了加载策略,批量插入 bulk_insert_mappings 能绕过 ORM 的对象实例化过程,直接让数据库驱动处理字典数据,速度提升可达一个数量级。对于高频只读查询,还可以在应用层引入查询结果缓存。
from sqlalchemy.orm import joinedload, selectinload, load_only
from sqlalchemy import select
with Session() as session:
# joinedload:一条 JOIN 语句预加载 posts
stmt = select(User).options(joinedload(User.posts)).where(User.id == 1)
user = session.execute(stmt).unique().scalar_one()
# selectinload:分两次查询,避免大数据量 JOIN 的笛卡尔积膨胀
stmt2 = select(User).options(selectinload(User.posts))
all_users = session.execute(stmt2).scalars().all()
# load_only:仅加载指定列,减少网络传输
stmt3 = select(User).options(load_only(User.name, User.email))
users_summary = session.execute(stmt3).scalars().all()
# 批量插入:跳过 ORM 对象生命周期,直接写库
with Session() as session:
session.bulk_insert_mappings(
User,
[
{"name": f"user_{i}", "email": f"user_{i}@test.com"}
for i in range(10000)
]
)
session.commit()
性能优化没有银弹,关键在于为每个场景选择正确的加载策略并测量。在生产监控中加入查询耗时统计,配合 SQL 慢日志分析,才能持续发现性能瓶颈。
11. 数据库事务隔离级别与死锁处理
事务隔离级别决定了并发事务之间的可见性边界,直接影响数据一致性与并发性能。SQL 标准定义了四种隔离级别:READ UNCOMMITTED、READ COMMITTED、REPEATABLE READ 和 SERIALIZABLE。PostgreSQL 默认使用 READ COMMITTED,意味着事务只能看到已提交的更改,但无法避免「不可重复读」。在需要严格一致性的批量处理或报表生成场景中,可以显式提升隔离级别到 REPEATABLE READ。然而,隔离级别越强,并发冲突概率越大,最容易出现的就是「死锁」:两个事务各自持有资源并等待对方释放,形成循环依赖。数据库会自动检测死锁并强制其中一个事务回滚,但应用程序必须捕获异常并设计重试机制,否则用户请求会直接失败。
import psycopg2
from psycopg2.extensions import ISOLATION_LEVEL_REPEATABLE_READ
conn = psycopg2.connect("postgresql://user:pass@localhost/db")
conn.set_isolation_level(ISOLATION_LEVEL_REPEATABLE_READ)
with conn.cursor() as cur:
cur.execute("BEGIN")
cur.execute("SELECT balance FROM accounts WHERE id = 1")
balance = cur.fetchone()[0]
cur.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 1")
cur.execute("UPDATE accounts SET balance = balance + 100 WHERE id = 2")
conn.commit()
# 死锁自动重试装饰器
import time
import functools
from sqlalchemy.exc import OperationalError
def retry_on_deadlock(max_retries=3, backoff=0.5):
def decorator(fn):
@functools.wraps(fn)
def wrapper(*args, **kwargs):
for attempt in range(max_retries):
try:
return fn(*args, **kwargs)
except OperationalError as e:
if "deadlock" in str(e).lower() and attempt < max_retries - 1:
time.sleep(backoff * (2 ** attempt))
continue
raise
return fn(*args, **kwargs)
return wrapper
return decorator
@retry_on_deadlock(max_retries=3)
def transfer_funds(session, from_id: int, to_id: int, amount: int):
# 始终按固定顺序获取锁,避免死锁
first, second = sorted([from_id, to_id])
acct1 = session.get(Account, first, with_for_update=True)
acct2 = session.get(Account, second, with_for_update=True)
acct1.balance -= amount
acct2.balance += amount
session.commit()
理解隔离级别和死锁机制,是写出高并发安全数据库代码的必修课。除了锁策略,合理的索引设计也能减少锁竞争范围,这一点在性能优化和部署调优中同样重要。
12. 测试数据库策略
单元测试和集成测试中,数据库的处理方式直接影响测试速度与可靠性。最实用的技术之一是「事务回滚测试」:在每个测试用例开始时开启一个数据库事务,测试结束后执行回滚。这样所有对数据库的修改都会被撤销,测试之间互不干扰,无需频繁创建和删除表。pytest 的 fixture 机制非常适合实现这种模式。然而,事务回滚测试无法完全替代真实数据库测试,因为它模拟的是单事务内的行为,无法发现锁竞争或事务隔离级别相关的问题。引入 Testcontainers 可以在测试时自动启动一个真实的 PostgreSQL 或 MySQL 容器,测试结束后再销毁,兼顾了真实性与隔离性。
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
# 事务回滚 pytest fixture
@pytest.fixture(scope="function")
def db_session():
engine = create_engine("postgresql://user:pass@localhost/testdb")
connection = engine.connect()
transaction = connection.begin()
Session = sessionmaker(bind=connection)
session = Session()
yield session
session.close()
transaction.rollback()
connection.close()
def test_create_user(db_session):
user = User(name="Test", email="test@test.com")
db_session.add(user)
db_session.flush()
assert user.id is not None
# 测试结束后事务回滚,数据库保持干净
# Testcontainers:真实数据库集成测试
import pytest
from testcontainers.postgres import PostgresContainer
from sqlalchemy import create_engine
@pytest.fixture(scope="session")
def postgres_engine():
with PostgresContainer("postgres:15") as postgres:
url = postgres.get_connection_url()
engine = create_engine(url)
Base.metadata.create_all(engine)
yield engine
Base.metadata.drop_all(engine)
def test_integration_user_crud(postgres_engine):
Session = sessionmaker(bind=postgres_engine)
with Session() as session:
user = User(name="Real", email="real@db.com")
session.add(user)
session.commit()
fetched = session.get(User, user.id)
assert fetched.name == "Real"
测试策略的选择取决于项目阶段。早期开发使用事务回滚即可快速迭代;在关键业务逻辑或发布前,建议加入 Testcontainers 集成测试,确保代码在真实数据库上表现一致。测试是保证持续交付质量的最后防线,投资自动化测试的长期回报极其可观。更多测试相关内容可以参考 Python 测试与质量工程:pytest、mock 与覆盖率实战 。
13. 生产环境最佳实践
- 使用连接池(SQLAlchemy 内置 QueuePool)
- 敏感信息存环境变量
- 索引外键列
- 定期备份(pg_dump、mysqldump)
- 查询慢日志分析
- 监控连接池利用率,及时调整 pool_size
- 对复杂报表场景,适当使用 Raw SQL 绕过 ORM overhead
延伸阅读
- Python Web 框架:FastAPI 实战 —— FastAPI + SQLAlchemy 项目结构
- Python 类型系统与 Pydantic V2 —— 数据校验与数据库模型
- Python 并发与性能深度指南 —— 数据库连接池与并发
- Python asyncio 极速异步编程 —— asyncpg 异步查询
- Python 部署与运维实战 —— 容器化部署中的数据库连接管理
- Python 类型安全与数据建模 —— SQLAlchemy 2.0 类型提示最佳搭档
- Python 专题总览 —— 更多 Python 后端开发文章
数据库是应用的根基。从 SQLite 的快速原型到 PostgreSQL 的生产部署,SQLAlchemy 让你用同一套 API 驾驭不同数据库。掌握 ORM 的查询优化和迁移管理,是 Python 后端工程师的核心能力。SQLAlchemy 2.0 带来的类型安全和函数式查询范式进一步巩固了 Python 在数据密集型应用中的地位。
继续阅读
探索更多技术文章
浏览归档,发现更多关于系统设计、工具链和工程实践的内容。