新手友好、全程实战、可直接复刻运行做Python后端开发SQLAlchemy是绕不开的数据库神器作为Python生态最主流的ORM框架彻底告别手写原生SQL用面向对象的方式操作数据库适配所有主流数据库企业级项目必备一、什么是SQLAlchemy1.1 核心定义SQLAlchemy /ˈælkəmi/艾尔克米炼金术是Python最强大、最流行的ORM对象关系映射框架。核心作用将Python对象与数据库数据表做映射开发者只需操作Python对象无需手写原生SQL兼容混合使用自动转换为数据库语句极大提升开发效率与代码可维护性。1.2 核心优势跨库兼容一套代码无缝适配 MySQL、PostgreSQL、SQLite、Oracle 等主流数据库解耦性强代码与数据库解绑切换数据库无需大规模改代码自动类型转换Python数据类型自动映射数据库字段类型功能强大完善的事务、关联查询、分组聚合、索引约束能力企业级首选FastAPI/Flask/Django 后端项目标配数据库框架1.3 两大核心模块SQLAlchemy整体分为两层学习重点为ORM层Core底层核心基础引擎、连接管理、SQL表达式构建ORM的底层依赖ORM上层封装对象关系映射面向对象操作数据库日常开发核心使用二、SQLAlchemy核心概念必懂2.1 五大核心组件Engine 引擎数据库连接核心入口管理连接池、数据库地址、账号密码所有操作的根基Session 会话数据库交互桥梁所有增删改查操作必须通过Session完成Base 基类所有数据库模型类的父类子类自动映射为数据库表Model 模型Python自定义类对应数据库一张表类属性对应表字段Column 字段定义表字段支持类型、主键、非空、唯一、默认值、索引等约束2.2 ORM映射核心原理ORM对象关系映射核心对应关系Python 模型类 ➡️ 数据库数据表类属性 ➡️ 数据表字段类实例对象 ➡️ 数据表一行数据三、环境搭建与安装3.1 安装依赖根据使用的数据库安装对应驱动SQLite无需额外驱动Python自带# 安装核心框架 pip install sqlalchemy # MySQL驱动二选一 pip install pymysql pip install mysql-connector-python # PostgreSQL驱动 pip install psycopg2-binary # Oracle驱动 pip install cx_Oracle3.2 验证安装成功import sqlalchemy import pymysql # 打印版本无报错即安装成功 print(SQLAlchemy版本, sqlalchemy.__version__) print(pymysql版本, pymysql.__version__)四、数据库前置准备SQLAlchemy仅负责操作数据表不会自动创建数据库需手动提前创建# 登录MySQL mysql -u root -p # 创建测试数据库支持utf8mb4全字符集 CREATE DATABASE IF NOT EXISTS sqlalchemy_demo CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci; # 查看数据库 SHOW DATABASES;五、项目基础配置通用模板新建sqlalchemy_config.py统一管理引擎、会话、基类全局复用from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker from sqlalchemy import create_engine # 1. 数据库连接字符串mysql驱动://账号:密码地址:端口/数据库名?编码 db_url mysqlpymysql://root:123456localhost:3306/sqlalchemy_demo?charsetutf8mb4 # 2. 创建引擎echoTrue 打印执行SQL调试开启生产关闭 engine create_engine(db_url, echoTrue) # 3. 创建会话工厂绑定引擎 SessionFactory sessionmaker(bindengine) # 4. 创建ORM基类所有模型继承此类 Base declarative_base()六、定义数据库模型Model新建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}, create_time{self.create_time})6.1 常用字段类型Integer整型MySQL INTString(n)字符串MySQL VARCHAR(n)DateTime时间日期MySQL DATETIMEFloat浮点型Boolean布尔型MySQL TINYINT(1)6.2 常用字段约束参数作用说明primary_keyTrue设置为主键autoincrementTrue整数主键自增nullableFalse字段非空uniqueTrue字段唯一约束defaultxxx代码层面默认值indexTrue创建普通索引commentxxx字段注释七、自动创建数据表执行以下代码根据模型自动生成数据表重复执行不会报错、不会重复创建from sqlalchemy_config import engine, Base from base_model import User # 根据所有继承Base的模型创建数据表 Base.metadata.create_all(engine) print(✅ 所有数据表创建成功)八、全套CRUD实战操作核心重点统一使用with上下文管理器自动关闭会话避免连接泄露8.1 新增数据 Create支持单条新增、批量对象新增、批量字典新增、2.0新版语法from sqlalchemy_config import SessionFactory from base_model import User from sqlalchemy import insert # 1. 单条新增 with SessionFactory() as session: user User(usernamezhangsan, password123456) session.add(user) session.commit() print(✅ 单条数据新增成功) # 2. 批量对象新增 with SessionFactory() as session: user1 User(usernamelisi, password123456) user2 User(usernamewangwu, password654321) session.add_all([user1, user2]) session.commit() print(✅ 批量对象新增成功) # 3. 批量字典新增高效大数据量 with SessionFactory() as session: user_list [ {username: sunqi, password: 123123}, {username: zhouba, password: 456456} ] session.bulk_insert_mappings(User, user_list) session.commit() print(✅ 批量字典新增成功) # 4. SQLAlchemy2.0 新版insert语法 with SessionFactory() as session: stmt insert(User).values([ {username: zhengshi, password: aaa123}, {username: chenshi, password: bbb456} ]) session.execute(stmt) session.commit() print(✅ 新版语法新增成功)8.2 查询数据 Read最全查询合集8.2.1 基础查询from sqlalchemy_config import SessionFactory from base_model import User with SessionFactory() as session: # 1. 查询所有数据 all_users session.query(User).all() print(所有用户, all_users) # 2. 根据主键查询单条 user session.get(User, 1) print(ID1的用户, user) # 3. 查询第一条数据 first_user session.query(User).first() print(第一条用户数据, first_user) # 4. 查询指定字段 field_users session.query(User.id, User.username, User.create_time).all() print(指定字段数据, field_users)8.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 ! zhangsan).all() # 模糊查询 res3 session.query(User).filter(User.username.like(%zhang%)).all() # IN范围查询 res4 session.query(User).filter(User.id.in_([1,2,3])).all() # 大小比较 res5 session.query(User).filter(User.id 2).all() # 多条件AND res6 session.query(User).filter(and_(User.id 1, User.username.like(z%))).all() # 多条件OR res7 session.query(User).filter(or_(User.id 1, User.username lisi)).all() # 简洁等值查询filter_by res8 session.query(User).filter_by(usernamezhangsan, password123456).first()8.2.3 排序、分页、统计from sqlalchemy_config import SessionFactory from base_model import User with SessionFactory() as session: # 升序排序 asc_users session.query(User).order_by(User.id).all() # 降序排序 desc_users session.query(User).order_by(User.id.desc()).all() # 分页查询第2页每页2条 page_num 2 page_size 2 page_users session.query(User).order_by(User.id).limit(page_size).offset((page_num-1)*page_size).all() # 统计数量 count session.query(User).filter(User.username.like(z%)).count() print(z开头用户数量, count)8.2.4 分组聚合查询from sqlalchemy_config import SessionFactory from sqlalchemy import func, select from base_model import User with SessionFactory() as session: # 分组统计 stmt select( User.username, func.count(User.id).label(user_count) ).group_by(User.username) res session.execute(stmt).mappings().all() print(分组统计结果, res) # 分组筛选 having stmt2 select( User.username, func.count(User.id).label(count) ).group_by(User.username).having(func.count(User.id) 0) res2 session.execute(stmt2).mappings().all() print(分组筛选结果, res2)8.3 更新数据 Updatefrom sqlalchemy_config import SessionFactory from base_model import User from sqlalchemy import update # 1. 单条更新 with SessionFactory() as session: user session.query(User).filter(User.username zhangsan).first() if user: user.password update123456 session.commit() print(✅ 单条更新成功) # 2. 批量更新 with SessionFactory() as session: session.query(User).filter(User.username.like(w%)).update({password: batch123}) session.commit() print(✅ 批量更新成功) # 3. 2.0新版更新语法 with SessionFactory() as session: stmt update(User).where(User.id 3).values(passwordnewpass123) session.execute(stmt) session.commit() print(✅ 新版语法更新成功)8.4 删除数据 Deletefrom sqlalchemy_config import SessionFactory from base_model import User from sqlalchemy import delete # 1. 单条删除 with SessionFactory() as session: user session.query(User).filter(User.username wangwu).first() if user: session.delete(user) session.commit() print(✅ 单条删除成功) # 2. 批量删除 with SessionFactory() as session: session.query(User).filter(User.id 10).delete() session.commit() print(✅ 批量删除成功) # 3. 2.0新版删除语法 with SessionFactory() as session: stmt delete(User).where(User.username.like(c%)) session.execute(stmt) session.commit() print(✅ 新版语法删除成功)九、新手避坑总结所有增删改操作必须执行session.commit()否则数据不入库调试阶段开启echoTrue可查看底层原生SQL快速排错优先使用with上下文管理会话自动释放连接杜绝连接池溢出批量操作优先用bulk_insert_mappings性能远高于循环add数据库需手动创建SQLAlchemy只负责表结构与数据操作十、本章小结本文完整覆盖 SQLAlchemy 入门核心知识点理解ORM思想、五大核心组件作用掌握环境搭建、数据库连接、模型定义、表创建精通全套CRUD操作增删改查、条件、排序、分页、分组聚合适配SQLAlchemy2.0新版语法贴合企业开发规范后续可进阶事务管理、多表关联查询、数据库迁移、FastAPI项目集成码字不易点赞收藏持续更新Python后端实战教程