Python与SQL数据库交互全攻略涵盖原生SQLAlchemy及多种ORM框架

Python与SQL数据库交互全攻略：从原生驱动到ORM框架的深度实践

你好，我是源码库的一名技术博主。在多年的Python后端开发中，与数据库打交道是家常便饭。从最初手写原生SQL的“刀耕火种”，到后来拥抱ORM框架的“现代化农业”，我踩过不少坑，也积累了许多实战心得。今天，我就系统地为你梳理一下Python与SQL数据库交互的完整路径，涵盖最常用的几种方式，希望能帮你找到最适合自己项目的“那把钥匙”。

第一章：基石篇——使用原生数据库驱动（DB-API）

无论ORM多么强大，了解底层驱动都是必要的。Python通过PEP 249定义的DB-API 2.0规范，为各种数据库提供了统一的接口。以最常用的MySQL和PostgreSQL为例。

实战踩坑提示：直接使用驱动时，务必注意连接管理和SQL注入防范。我曾经因为一个未关闭的连接导致数据库连接池耗尽，线上服务报警。

1. 连接MySQL（使用PyMySQL或mysql-connector-python）：

pip install pymysql

import pymysql

# 建立连接
connection = pymysql.connect(
    host='localhost',
    user='your_username',
    password='your_password',
    database='test_db',
    charset='utf8mb4',
    cursorclass=pymysql.cursors.DictCursor # 让结果以字典形式返回，强烈推荐
)

try:
    with connection.cursor() as cursor:
        # 执行查询，使用参数化查询防止SQL注入！
        sql = "SELECT `id`, `name` FROM `users` WHERE `email` = %s"
        cursor.execute(sql, ('user@example.com',))
        result = cursor.fetchone()  # 获取单条
        # result = cursor.fetchall() # 获取所有

    # 非查询操作需要提交
    connection.commit()
finally:
    connection.close() # 切记关闭连接！

2. 连接PostgreSQL（使用psycopg2）：

pip install psycopg2-binary

import psycopg2

conn = psycopg2.connect(
    host="localhost",
    database="test_db",
    user="postgres",
    password="your_password"
)
# psycopg2同样支持上下文管理，用法类似

核心心得：原生驱动性能最好，控制力最强，适合执行复杂SQL或对性能有极致要求的场景。但需要自己管理连接、事务和结果集映射，代码量较大。

第二章：进阶篇——使用SQLAlchemy Core（模式层）

当你觉得原生驱动太“裸”，但又不想被ORM的抽象完全束缚时，SQLAlchemy Core是你的绝佳选择。它提供了SQL表达式语言，能帮你以Pythonic的方式安全、高效地构建SQL。

pip install sqlalchemy

from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String, select, text

# 创建引擎（连接池已内置！）
engine = create_engine('mysql+pymysql://user:pass@localhost/test_db')

# 定义表结构（无需ORM类）
metadata = MetaData()
users = Table('users', metadata,
              Column('id', Integer, primary_key=True),
              Column('name', String(50)),
              Column('fullname', String(50)),
)

# 使用表达式语言构建查询（安全，且可读性好）
stmt = select(users.c.name, users.c.fullname).where(users.c.id > 5)
# 等价于 SELECT name, fullname FROM users WHERE id > 5

with engine.connect() as conn:
    result = conn.execute(stmt)
    for row in result:
        print(f"name: {row.name}, fullname: {row.fullname}")

# 也可以直接执行原生SQL，但更安全
with engine.connect() as conn:
    # 使用 text() 构造，并传递参数
    result = conn.execute(text("SELECT * FROM users WHERE name = :name"), {"name": "jack"})
    # SQLAlchemy会帮你处理参数和结果游标

实战优势：我常在数据迁移、复杂报表生成或ETL任务中使用Core。它既避免了字符串拼接SQL的危险，又保持了接近原生SQL的灵活性，性能损耗极小。

第三章：高效开发篇——使用SQLAlchemy ORM

对于大多数业务应用，ORM（对象关系映射）能极大提升开发效率。SQLAlchemy ORM是其最强大的组件。

from sqlalchemy.orm import declarative_base, Session
from sqlalchemy import Column, Integer, String, create_engine

Base = declarative_base()

# 定义模型（Python类）
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    fullname = Column(String)

    def __repr__(self):
        return f""

# 创建所有表（通常用于项目初始化）
engine = create_engine('sqlite:///example.db') # 这里用SQLite演示
Base.metadata.create_all(engine)

# 开始会话（Session是ORM的核心，管理所有对象和事务）
with Session(engine) as session:
    # 增
    new_user = User(name='ed', fullname='Ed Jones')
    session.add(new_user)
    # 可以add_all([user1, user2])

    # 查（使用强大的查询API）
    user = session.query(User).filter_by(name='ed').first()
    # 或使用更现代的2.0风格查询
    from sqlalchemy import select
    stmt = select(User).where(User.name == 'ed')
    user = session.execute(stmt).scalars().first()

    # 改
    if user:
        user.fullname = 'Edward Jones'

    # 删
    # session.delete(user)

    # 所有操作在上下文管理器退出（或显式调用session.commit()）时提交
    session.commit()

踩坑重点：Session的生命周期管理是关键。在Web框架中，通常采用“请求开始时创建，响应返回前提交并关闭”的模式（如Flask-SQLAlchemy的scoped session）。避免在全局或跨请求使用同一个Session。

第四章：轻量敏捷篇——尝试其他ORM框架

SQLAlchemy虽强大但略重。如果你的项目需要更轻量、更符合直觉的ORM，可以考虑以下选择。

1. Peewee：小巧而优雅

Peewee的API设计非常Pythonic，学习曲线平缓，适合中小型项目。

pip install peewee

from peewee import SqliteDatabase, Model, CharField, IntegerField

db = SqliteDatabase('people.db')

class Person(Model):
    name = CharField()
    age = IntegerField()

    class Meta:
        database = db # 模型绑定数据库

db.connect()
db.create_tables([Person])

# 增删改查极其简洁
Person.create(name='Charlie', age=30)
charlie = Person.get(Person.name == 'Charlie')
charlie.age = 31
charlie.save()

for person in Person.select().where(Person.age > 20):
    print(person.name)

2. SQLModel：基于Pydantic和SQLAlchemy的现代选择

如果你喜欢FastAPI和Pydantic的类型提示与数据验证，那么SQLModel会让你感到亲切。它由FastAPI作者开发，融合了SQLAlchemy和Pydantic的优点。

pip install sqlmodel

from sqlmodel import Field, Session, SQLModel, create_engine, select

class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    secret_name: str
    age: int | None = None

engine = create_engine("sqlite:///database.db")
SQLModel.metadata.create_all(engine)

hero_1 = Hero(name="Spider-Man", secret_name="Peter Parker", age=17)

with Session(engine) as session:
    session.add(hero_1)
    session.commit()

    # 查询支持强大的类型提示和自动补全
    statement = select(Hero).where(Hero.age >= 18)
    results = session.exec(statement)
    for hero in results:
        print(hero)

我的选择建议：追求极致控制和性能的大型复杂系统，多用SQLAlchemy Core。快速构建标准业务应用，SQLAlchemy ORM是稳妥的行业标准。开发小型工具或原型，Peewee非常顺手。构建基于FastAPI的现代API服务，强烈推荐尝试SQLModel。

第五章：连接管理最佳实践与性能优化

无论选择哪种方式，连接管理都是重中之重。

1. 使用连接池：SQLAlchemy的create_engine默认已启用连接池。对于原生驱动，可以考虑使用DBUtils或SQLAlchemy的池化功能。

2. 设置合理的连接参数：

engine = create_engine(
    'mysql+pymysql://user:pass@localhost/db',
    pool_size=20,         # 连接池保持的连接数
    max_overflow=10,      # 超出pool_size后最多创建的连接数
    pool_recycle=3600,    # 连接回收时间（秒），避免数据库断开
    echo=False,           # 开发时可设为True查看SQL日志
)

3. 异步支持：现代Python异步生态日益完善。可以考虑：

• asyncpg + sqlalchemy[asyncio] 用于PostgreSQL。
• aiomysql 用于MySQL。
• ORM方面，SQLAlchemy ORM、SQLModel、Tortoise-ORM（专为异步设计）都提供了良好的异步支持。

希望这篇涵盖从底层到上层的攻略，能帮助你构建出更健壮、更高效的Python数据库应用。记住，没有最好的工具，只有最适合场景的工具。理解原理，灵活选用，才是工程师的核心能力。如果在实践中遇到具体问题，欢迎来源码库社区一起探讨。