SQLAlchemy 2.0+ 实战：彻底替代原生 SQL，实现用户表完整 CRUD

在 Python 数据库开发中，原生 SQL 操作不仅繁琐且易出错，还存在 SQL 注入风险。而 SQLAlchemy 作为最主流的 ORM 框架，完美解决了这些问题 —— 它无需编写原生 SQL，通过 Python 对象即可完成数据库操作，同时支持跨数据库兼容、内置安全防护等强大特性，适用于 Flask、Django 及纯 Python 项目。本文将从环境搭建到实战演示，手把手教你用 SQLAlchemy 2.0 + 实现用户表的完整 CRUD 操作，代码简洁易维护，新手也能快速上手！

一、SQLAlchemy 核心优势

相比原生 pymysql，SQLAlchemy 的优势堪称碾压级，这也是它成为行业标准的原因：

二、环境准备

1. 安装依赖

SQLAlchemy 需要配合 MySQL 驱动使用，这里选择 pymysql 作为底层驱动（兼容性更好）：

pip install sqlalchemy pymysql

2. 前置条件

• 已启动 MySQL 服务，拥有可操作的数据库账号；
• 提前创建测试数据库（示例中为db_user），并创建用户表：

CREATE TABLE IF NOT EXISTS `user` (
`id` INT PRIMARY KEY AUTO_INCREMENT COMMENT '用户唯一ID（自增）',
`username` VARCHAR(50) NOT NULL UNIQUE COMMENT '用户名（唯一）',
`password` VARCHAR(100) NOT NULL COMMENT '用户密码',
`age` TINYINT COMMENT '用户年龄',
`create_time` DATETIME DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间（自动生成）'
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='用户信息表';

3. 项目结构设计

为保证代码的可读性和可维护性，采用分层架构设计，推荐项目结构如下：

your_project/ # 项目根目录
├── config/ # 配置层：数据库连接等全局配置
│ └── database.py # SQLAlchemy引擎、会话工厂初始化
├── models/ # 模型层：ORM表模型定义
│ └── user.py # User表模型类（核心）
├── user_crud.py # 业务逻辑层：CRUD核心操作
├── main.py # 入口层：测试CRUD操作
└── requirements.txt # 依赖清单：管理第三方包版本

三、核心实现步骤

1. 数据库连接与 ORM 初始化（config/database.py）

这一步是 SQLAlchemy 的基础配置，负责创建数据库连接引擎、声明模型基类、创建会话工厂：

from sqlalchemy import create_engine
from sqlalchemy.orm import declarative_base, sessionmaker

# MySQL连接地址格式：mysql+pymysql://用户名:密码@主机:端口/数据库名?编码参数
DB_URL = "mysql+pymysql://root:root@localhost:3306/db_user?charset=utf8mb4"

# 1. 创建引擎（Engine）：管理数据库连接池，底层调用pymysql
engine = create_engine(
DB_URL,
echo=False, # 调试时设为True，会打印底层执行的SQL；生产环境关闭
pool_recycle=3600, # 连接池回收时间，避免无效连接
pool_size=10 # 连接池大小，按需调整
)

# 2. 声明基类（Base）：所有表模型类的父类，自动完成类与表的映射
Base = declarative_base()

# 3. 创建会话工厂（SessionLocal）：替代原生的conn/cursor
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

2. 定义 User 表模型（models/user.py）

Python 类与数据库表的映射核心，类属性直接对应表字段，无需手动编写 CREATE 语句：

from sqlalchemy import Column, Integer, String, DateTime, func
from config.database import Base # 导入声明基类

class User(Base):
__tablename__ = "user" # 对应数据库表名（必须指定）

# 表字段定义：类属性 = Column(字段类型, 约束条件, 备注)
id = Column(Integer, primary_key=True, autoincrement=True, comment="用户唯一ID（自增）")
username = Column(String(50), unique=True, nullable=False, comment="用户名（唯一，非空）")
password = Column(String(100), nullable=False, comment="用户密码（非空，生产环境需加密）")
age = Column(Integer, nullable=True, comment="用户年龄（可选）")
# func.now()适配不同数据库的当前时间函数，实现跨库兼容
create_time = Column(DateTime, default=func.now(), comment="创建时间（自动生成）")

def __repr__(self):
"""自定义对象打印格式，方便调试"""
create_time_str = self.create_time.strftime('%Y-%m-%d %H:%M:%S') if self.create_time else None
return f"User(id={self.id}, username='{self.username}', age={self.age}, create_time='{create_time_str}')"

3. 实现 CRUD 核心操作（user_crud.py）

封装通用的数据库会话获取方法和 CRUD 函数，所有操作均通过 Python 对象完成，无原生 SQL：

from typing import Optional, List
from config.database import SessionLocal
from models.user import User

def get_db():
"""
获取数据库会话，使用yield实现上下文管理，自动关闭会话
调用方式：with get_db() as db: 或 for db in get_db():
"""
db = SessionLocal()
try:
yield db # 提供会话对象给业务代码
finally:
db.close() # 无论是否异常，自动释放连接

# ————————– 新增（Create）————————–
def create_user(db, username: str, password: str, age: Optional[int] = None) -> User:
"""新增用户：创建User对象并添加到会话"""
new_user = User(username=username, password=password, age=age)
db.add(new_user) # 预执行INSERT
db.commit() # 事务提交，真正写入数据库
db.refresh(new_user) # 同步自动生成的id/create_time字段
print(f"✅ 新增用户成功：{new_user}")
return new_user

# ————————– 查询（Read）————————–
def get_user_by_id(db, user_id: int) -> Optional[User]:
"""根据ID查询单条用户（主键查询，效率最高）"""
user = db.get(User, user_id) # ORM内置主键查询方法
if not user:
print(f"🔍 用户ID{user_id}不存在")
return user

def get_user_by_username(db, username: str) -> Optional[User]:
"""根据用户名查询单条用户（条件过滤）"""
user = db.query(User).filter(User.username == username).first()
if not user:
print(f"🔍 用户名{username}不存在")
return user

def get_all_users(db, age_ge: Optional[int] = None) -> List[User]:
"""批量查询用户，支持年龄≥age_ge的条件过滤和排序"""
query = db.query(User).order_by(User.id.asc()) # 基础查询+排序
if age_ge is not None:
query = query.filter(User.age >= age_ge) # 条件过滤
user_list = query.all() # 获取所有结果，返回User对象列表
print(f"🔍 批量查询完成，共{len(user_list)}条用户数据")
return user_list

# ————————– 更新（Update）————————–
def update_user(db, user_id: int, **kwargs) -> Optional[User]:
"""动态更新用户信息，支持密码、年龄等字段"""
user = get_user_by_id(db, user_id)
if not user:
return None
if not kwargs:
print("❌ 更新失败：未传入任何字段")
return user
# 动态修改对象属性，ORM自动跟踪变化
for key, value in kwargs.items():
if hasattr(user, key):
setattr(user, key, value)
db.commit()
db.refresh(user)
print(f"✅ 用户ID{user_id}更新成功：{user}")
return user

# ————————– 删除（Delete）————————–
def delete_user(db, user_id: int) -> bool:
"""根据ID删除用户，返回删除结果"""
user = get_user_by_id(db, user_id)
if not user:
return False
db.delete(user) # 标记删除
db.commit() # 提交事务，执行DELETE
print(f"✅ 用户ID{user_id}删除成功")
return True

4. 测试 CRUD 操作（main.py）

编写测试入口，验证所有 CRUD 功能是否正常工作：

from user_crud import *

if __name__ == "__main__":
# 获取数据库会话（自动管理连接，无需手动关闭）
for db in get_db():
# 1. 新增3个用户
user1 = create_user(db, "zhangsan", "123456", 20)
user2 = create_user(db, "lisi", "654321", 25)
user3 = create_user(db, "wangwu", "888888", 30)
print("\\n" + "-"*60 + "\\n")

# 2. 单条查询（按ID/用户名）
user_by_id = get_user_by_id(db, user1.id)
user_by_name = get_user_by_username(db, "lisi")
print(f"🔍 按ID查询：{user_by_id}")
print(f"🔍 按用户名查询：{user_by_name}")
print("\\n" + "-"*60 + "\\n")

# 3. 批量查询（所有用户/年龄≥18的用户）
all_users = get_all_users(db)
adult_users = get_all_users(db, age_ge=18)
print(f"🔍 所有用户：{all_users}")
print(f"🔍 年龄≥18的用户：{adult_users}")
print("\\n" + "-"*60 + "\\n")

# 4. 更新用户信息（修改密码和年龄）
updated_user = update_user(db, user1.id, password="zhangsan_new_pwd", age=22)
print("\\n" + "-"*60 + "\\n")

# 5. 删除用户并验证结果
delete_result = delete_user(db, user3.id)
after_delete_users = get_all_users(db)
print(f"🔍 删除后剩余用户：{after_delete_users}")

print("\\n🎉 所有CRUD操作执行完成（SQLAlchemy ORM实现）")

四、SQLAlchemy 核心概念解析

想要灵活运用 SQLAlchemy，必须理解以下 5 个核心概念：

1. 引擎（Engine）

• 作用：管理数据库连接池，是 SQLAlchemy 与数据库的底层交互入口，底层调用 pymysql；
• 关键参数：echo（是否打印 SQL）、pool_size（连接池大小）、pool_recycle（连接回收时间）。

2. 基类（Base）

• 通过declarative_base()创建，是所有表模型类的父类；
• 核心作用：自动将子类映射为数据库表，无需手动编写 CREATE TABLE 语句。

3. 表模型类（如 User）

• 核心映射规则：Python类 = 数据库表，类属性 = 表字段；
• 关键配置：
  • __tablename__：指定对应数据库表名（必填）；
  • Column：定义字段，包含字段类型（Integer/String 等）、约束（primary_key/unique 等）、默认值；
  • func.now()：跨数据库兼容的当前时间函数，适配 MySQL 的 NOW () 和 PostgreSQL 的 CURRENT_TIMESTAMP。

4. 会话（Session）

• 由sessionmaker创建会话工厂，再通过get_db()生成会话对象（db）；
• 作用：替代原生的 conn 和 cursor，是 ORM 操作数据库的核心入口，支持事务（commit 提交 /rollback 回滚）；
• 特性：自动管理连接，通过 finally 块确保关闭，避免资源泄漏。

5. 常用查询方法映射表

五、实战总结与进阶建议

1. 核心优势回顾

• 代码更简洁：用 Python 对象操作替代原生 SQL，减少重复代码；
• 维护成本低：表结构变更仅需修改模型类，无需同步修改所有 SQL；
• 安全性更高：内置防 SQL 注入，无需手动处理参数；
• 扩展性更强：支持跨数据库迁移、复杂关联查询、事务管理等高级功能。

2. 进阶使用建议

• 密码加密：生产环境中，切勿明文存储密码，建议使用 bcrypt 等算法加密；
• 数据校验：可结合 pydantic 对输入数据进行校验，避免无效数据入库；
• 分页查询：对于大数据量场景，使用limit()和offset()实现分页，避免一次性查询所有数据；
• 关联查询：如果存在多表关联（如用户表与订单表），可通过relationship实现关联查询，替代 JOIN 语句；
• 数据库迁移：使用alembic工具管理数据库表结构变更，支持版本控制和回滚。

相比原生 SQL，ORM 框架不仅提升了开发效率，还让代码更具可读性和可维护性，是 Python 后端开发的必备技能。建议结合实际项目多做练习，深入理解会话管理、查询优化等高级特性，让数据库操作更加高效、安全！