摘要SQLAlchemy是Python中最流行的ORM框架把数据库表映射成Python对象让你用操作对象的方式操作数据库不用手写SQL。本文从Engine、Session、Base讲到CRUD的完整实现。一、什么是SQLAlchemySQLAlchemy是Python中最流行的ORM框架。ORMObject-Relational Mapping把Python对象和数据库表对应起来类对应表、属性对应字段、实例对应一行数据。核心优势一套代码支持MySQL、PostgreSQL、SQLite、Oracle等主流数据库切换数据库只需改连接字符串不用改业务代码。二、核心概念SQLAlchemy分两层Core底层引擎和ORM上层映射。我们重点学ORM。Engine数据库连接的入口通过连接字符串创建决定了连什么数据库、用什么账号密码。Session和数据库交互的桥梁所有增删改查都通过Session完成。Base所有模型类的父类。继承Base的类会自动映射成数据库表。ModelPython类对应一张表。类属性对应表字段类实例对应一行数据。Column定义字段可指定类型、主键、非空、默认值等约束。三、快速开始安装依赖bashpip install sqlalchemy pymysql其他数据库驱动PostgreSQLpip install psycopg2-binarySQLitePython自带无需额外安装Oraclepip install cx_Oracle创建Engine连接数据库pythonfrom sqlalchemy import create_engine # 格式mysqlpymysql://用户名:密码主机:端口/数据库名 engine create_engine( mysqlpymysql://root:123456localhost:3306/sqlalchemy_demo, echoTrue # 打印生成的SQL调试用 )创建Base和Modelpythonfrom sqlalchemy.orm import declarative_base from sqlalchemy import Column, Integer, String, DateTime, Boolean from datetime import datetime Base declarative_base() class User(Base): __tablename__ user # 对应数据库表名 id Column(Integer, primary_keyTrue, autoincrementTrue, comment用户ID) username Column(String(50), nullableFalse, uniqueTrue, comment用户名) password Column(String(100), nullableFalse, comment密码) email Column(String(100), comment邮箱) is_active Column(Boolean, defaultTrue, comment是否激活) created_at Column(DateTime, defaultdatetime.now, comment创建时间)常用字段类型Integer整数、String(n)字符串、DateTime时间、Float浮点数、Boolean布尔常用约束primary_key主键、autoincrement自增、nullable非空、unique唯一、default默认值创建数据表python# 根据所有继承Base的模型创建表 Base.metadata.create_all(engine)执行后在数据库中看到user表。创建Sessionpythonfrom sqlalchemy.orm import sessionmaker SessionFactory sessionmaker(bindengine) session SessionFactory()用with自动管理关闭pythonwith SessionFactory() as session: # 操作数据库 pass四、CRUD操作Create新增pythonfrom sqlalchemy_config import SessionFactory from base_model import User with SessionFactory() as session: # 创建实例 new_user User( usernamezhangsan, password123456, emailzhangsanexample.com ) # 添加到会话 session.add(new_user) # 批量添加 session.add_all([ User(usernamelisi, password123456), User(usernamewangwu, password123456) ]) # 提交事务 session.commit()Read查询基础查询pythonwith SessionFactory() as session: # 查询全部 users session.query(User).all() # 查询第一条 user session.query(User).first() # 按条件查询 user session.query(User).filter(User.username zhangsan).first()条件查询python# 模糊查询用户名以zh开头 users session.query(User).filter(User.username.like(zh%)).all() # 多条件且 users session.query(User).filter( User.is_active True, User.id 5 ).all() # 多条件或 from sqlalchemy import or_ users session.query(User).filter( or_(User.username zhangsan, User.username lisi) ).all() # 范围查询id在2到5之间 users session.query(User).filter(User.id.between(2, 5)).all() # 包含用户名在列表中 users session.query(User).filter(User.username.in_([zhangsan, lisi])).all()排序和分页python# 升序 users session.query(User).order_by(User.id).all() # 降序 users session.query(User).order_by(User.id.desc()).all() # 分页跳过3条取2条 users session.query(User).offset(3).limit(2).all()分组和聚合pythonfrom sqlalchemy import func # 统计总数 count session.query(func.count(User.id)).scalar() # 按是否激活分组统计 results session.query( User.is_active, func.count(User.id).label(count) ).group_by(User.is_active).all()Update更新pythonwith SessionFactory() as session: # 单条更新查出来改属性 user session.query(User).filter(User.username zhangsan).first() if user: user.password new_password session.commit() # 批量更新 session.query(User).filter(User.username.like(w%)).update( {password: common_password} ) session.commit() # SQLAlchemy 2.0语法 from sqlalchemy import update stmt update(User).where(User.username.like(w%)).values( passwordsqlalchemy2.0_password ) session.execute(stmt) session.commit()Delete删除pythonwith SessionFactory() as session: # 单条删除 user session.query(User).filter(User.username wangwu).first() if user: session.delete(user) session.commit() # 批量删除 session.query(User).filter(User.id 3).delete() session.commit() # SQLAlchemy 2.0语法 from sqlalchemy import delete stmt delete(User).where(User.id 18) session.execute(stmt) session.commit()五、抽取通用配置项目中把配置抽出来统一管理避免到处写连接字符串。sqlalchemy_config.pypythonfrom sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker, declarative_base # 连接配置 DATABASE_URL mysqlpymysql://root:123456localhost:3306/sqlalchemy_demo # 创建Engine engine create_engine(DATABASE_URL, echoFalse) # 创建Session工厂 SessionFactory sessionmaker(bindengine) # 创建Base Base declarative_base()base_model.pypythonfrom sqlalchemy import Column, Integer, String, DateTime, Boolean from sqlalchemy_config import Base from datetime import datetime class User(Base): __tablename__ user id Column(Integer, primary_keyTrue, autoincrementTrue) username Column(String(50), nullableFalse, uniqueTrue) password Column(String(100), nullableFalse) email Column(String(100)) is_active Column(Boolean, defaultTrue) created_at Column(DateTime, defaultdatetime.now)使用pythonfrom sqlalchemy_config import SessionFactory from base_model import User with SessionFactory() as session: users session.query(User).all() for u in users: print(u.username, u.email)六、常用字段类型和约束速查字段类型类型对应MySQL说明IntegerINT整数String(n)VARCHAR(n)变长字符串TextTEXT长文本DateTimeDATETIME日期时间FloatFLOAT浮点数BooleanTINYINT(1)布尔值字段约束参数说明primary_keyTrue主键autoincrementTrue自增nullableFalse非空uniqueTrue唯一default值默认值comment说明字段注释七、常见问题表已存在时会重复创建吗不会。Base.metadata.create_all(engine)会检查表是否存在已存在则跳过。Session用完后要关闭吗要。用with SessionFactory() as session自动管理或用完手动session.close()。修改数据后必须commit吗必须。增删改都要session.commit()才能同步到数据库。查询不需要。filter和filter_by有什么区别filter_by只能用于等值条件写法简单filter_by(usernamezhangsan)filter支持所有条件表达式更灵活filter(User.username zhangsan)