python

Python 数据库实战：SQLite、PostgreSQL、SQLAlchemy 与 ORM 完全指南

Python 数据库开发完全指南：SQLite 快速上手、PostgreSQL/MySQL 连接、SQLAlchemy ORM 模型定义与查询、asyncpg 异步数据库访问、Alembic 迁移工具。从简单脚本到生产级数据库应用的全栈实践。

spcent 2026-08-13 8 分钟阅读 3934 字

几乎所有 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

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章返回首页