在Python后端开发中原生SQL手写语句冗余繁杂、可读性低且跨数据库适配难度大。而SQLAlchemy是Python生态最主流的ORM对象关系映射框架能够通过Python面向对象语法操作数据库无需频繁编写原生SQL同时兼容原生SQL混用极大提升开发效率。本文基于SQLAlchemy 2.0版本从零讲解核心概念、环境搭建、模型映射、全套CRUD实战所有代码可直接复制运行适配新手入门与日常开发备查。一、什么是 SQLAlchemySQLAlchemy是一款功能强大的Python ORM框架核心作用是实现Python对象与关系型数据库表的双向映射。简单来说数据库表对应Python类表中字段对应类属性表中数据行对应类的实例。开发者无需深耕SQL语法通过操作Python对象即可完成增删改查、分组聚合、分页排序等数据库操作同时支持原生SQL混合使用兼顾开发效率与灵活性。SQLAlchemy 核心优势跨数据库兼容一套语法适配MySQL、PostgreSQL、SQLite、Oracle等主流数据库项目数据库迁移几乎零成本代码简洁易维护面向对象编程告别冗余SQL语句代码可读性、可维护性大幅提升功能全面强大支持事务管理、复杂条件查询、分组聚合、外键关联、分页排序等企业级数据库能力自动类型转换自动适配Python与数据库数据类型无需手动转换时间、数字、字符串类型生态成熟通用完美适配FastAPI、Flask等主流Python后端框架是中小型项目到企业级项目的通用选型二、SQLAlchemy 核心架构SQLAlchemy整体分为Core核心组件和ORM对象映射层两部分日常开发以ORM层为核心Core层为底层支撑。1. Core 底层核心组件Core是SQLAlchemy的底层引擎负责数据库连接管理、SQL语句构建、事务调度是ORM功能的基础支撑核心四大组件如下Engine数据库引擎数据库连接核心入口通过连接字符串绑定数据库管理连接池、驱动类型、连接参数Session会话数据库交互的唯一桥梁所有增删改查操作都需通过Session执行相当于临时数据库连接会话Base基类所有数据模型类的父类继承Base的Python类会自动映射为数据库数据表Column字段用于定义模型类属性对应数据库表字段可配置字段类型、主键、非空、唯一、默认值等约束2. ORM 对象关系映射ORM对象关系映射是一种编程技术打通面向对象代码与关系型数据库的壁垒核心映射关系Python 模型类 → 数据库数据表模型类属性 → 数据表字段模型类实例 → 数据表单行数据借助ORM开发者可以完全用Python思维操作数据库大幅降低数据库开发门槛实现业务代码与数据库的解耦。三、环境搭建与数据库准备1. 依赖安装需要安装SQLAlchemy核心库与对应数据库驱动本文以MySQL为例# 安装核心框架 pip install sqlalchemy # 安装MySQL驱动二选一推荐pymysql pip install pymysql其他数据库驱动补充PostgreSQL安装psycopg2-binarySQLite无需额外安装Python原生支持。安装验证导入库并打印版本无报错即为成功。import sqlalchemy import pymysql print(sqlalchemy.__version__) print(pymysql.__version__)2. 数据库准备SQLAlchemy仅负责操作数据表不会自动创建数据库需手动提前创建# 登录MySQL后执行 CREATE DATABASE IF NOT EXISTS sqlalchemy_demo CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci; # 查看数据库确认创建成功 SHOW DATABASES;四、初始化配置与模型创建1. 全局配置封装新建sqlalchemy_config.py统一封装引擎、会话工厂、基类避免代码冗余from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker from sqlalchemy import create_engine # 数据库连接字符串驱动://账号:密码地址:端口/数据库?编码 db_url mysqlpymysql://root:123456localhost:3306/sqlalchemy_demo?charsetutf8mb4 # 创建引擎echoTrue开启SQL日志打印方便调试 engine create_engine(db_url, echoTrue) # 创建会话工厂绑定数据库引擎 SessionFactory sessionmaker(bindengine) # 创建ORM基类 Base declarative_base()2. 数据模型定义新建base_model.py定义数据表模型配置字段类型与约束from sqlalchemy import Column, Integer, String, DateTime from datetime import datetime from sqlalchemy_config import Base # 用户表模型 class User(Base): # 指定数据库表名 __tablename__ user # 主键ID自增、非空 id Column(Integer, primary_keyTrue, autoincrementTrue, comment用户ID) # 用户名最大50字符、非空、唯一 username Column(String(50), nullableFalse, uniqueTrue, comment用户名) # 密码最大100字符、非空 password Column(String(100), nullableFalse, comment登录密码) # 创建时间默认当前时间 create_time Column(DateTime, defaultdatetime.now, comment创建时间) # 重写打印方法方便调试查看数据 def __repr__(self): return fUser(id{self.id}, username{self.username})3. 自动创建数据表通过基类方法自动根据模型创建表重复执行不会报错、不会重复建表from sqlalchemy_config import engine, Base from base_model import User # 自动创建所有模型对应的数据表 Base.metadata.create_all(engine) print(数据表创建成功)五、ORM 全套 CRUD 实战所有数据库操作均通过SessionFactory创建会话采用with语句自动关闭会话避免连接泄露。1. 新增数据Create支持单条新增、批量新增、字典批量插入多种方式from sqlalchemy_config import SessionFactory from base_model import User # 1. 单条新增 with SessionFactory() as session: user User(usernamezhangsan, password123456) session.add(user) session.commit() # 2. 多条对象新增 with SessionFactory() as session: user1 User(usernamelisi, password123456) user2 User(usernamewangwu, password654321) session.add_all([user1, user2]) session.commit() # 3. 字典批量新增无需创建对象高效 with SessionFactory() as session: data_list [ {username: sunqi, password: 123123}, {username: zhouba, password: 456456} ] session.bulk_insert_mappings(User, data_list) session.commit()2. 查询数据ReadSQLAlchemy查询功能丰富支持基础查询、条件查询、排序分页、分组聚合等场景。2.1 基础查询from sqlalchemy_config import SessionFactory from base_model import User with SessionFactory() as session: # 查询所有数据 all_user session.query(User).all() # 查询单条数据主键查询 user session.get(User, 1) # 查询第一条数据 first_user session.query(User).first() # 查询指定字段 field_data session.query(User.id, User.username).all()2.2 条件查询from sqlalchemy_config import SessionFactory from base_model import User from sqlalchemy import and_, or_ with SessionFactory() as session: # 等值查询 res1 session.query(User).filter(User.username zhangsan).first() # 模糊查询 res2 session.query(User).filter(User.username.like(%z%)).all() # 范围查询 res3 session.query(User).filter(User.id.in_([1,2,3])).all() # 多条件AND查询 res4 session.query(User).filter(and_(User.id 1, User.username.like(z%))).first() # 多条件OR查询 res5 session.query(User).filter(or_(User.id 2, User.username lisi)).all() # 简易多条件查询filter_by仅支持等值 res6 session.query(User).filter_by(usernamelisi, password123456).first()2.3 排序与分页from sqlalchemy_config import SessionFactory from base_model import User with SessionFactory() as session: # 升序、降序排序 asc_data session.query(User).order_by(User.id).all() desc_data session.query(User).order_by(User.id.desc()).all() # 分页查询第2页每页2条 page 2 size 2 page_data session.query(User).order_by(User.id).limit(size).offset((page-1)*size).all() # 统计数据总数 total session.query(User).count()2.4 分组聚合查询结合func实现统计、求和、平均值等聚合操作支持分组筛选havingfrom sqlalchemy_config import SessionFactory from sqlalchemy import func, select from base_model import Account with SessionFactory() as session: # 按年龄分组统计每组人数 group_res session.query(Account.age, func.count(Account.id)).group_by(Account.age).all() # 新版2.0语法字典格式返回结果 stmt select(Account.age, func.count(Account.id).label(count)).group_by(Account.age) res_dict session.execute(stmt).mappings().all() # 分组后筛选只展示人数大于1的分组 having_res session.query(Account.age, func.count(Account.id)).group_by(Account.age).having(func.count(Account.id)1).all()3. 更新数据Update支持单条数据修改、批量数据更新修改后必须commit提交生效from sqlalchemy_config import SessionFactory from base_model import User # 单条更新 with SessionFactory() as session: user session.query(User).filter(User.username zhangsan).first() if user: user.password new123456 session.commit() # 批量更新 with SessionFactory() as session: session.query(User).filter(User.username.like(z%)).update({password: batch123}) session.commit()4. 删除数据Deletefrom sqlalchemy_config import SessionFactory from base_model import User # 单条删除 with SessionFactory() as session: user session.query(User).filter(User.username zhouba).first() if user: session.delete(user) session.commit() # 批量删除 with SessionFactory() as session: session.query(User).filter(User.id 10).delete() session.commit()六、开发常用注意事项事务必须提交新增、修改、删除操作必须执行session.commit()否则数据不会入库会话及时关闭推荐使用with上下文管理会话自动释放数据库连接避免连接耗尽调试开启echo开发环境开启echoTrue可查看底层执行的原生SQL方便排错分组查询规则聚合查询中查询字段只能是分组字段和聚合函数不可查询普通字段区分filter与filter_byfilter支持所有条件语法filter_by仅支持等值查询语法更简洁