第2章：创建表与增加记录
2.1 定义模型类 = 定义表结构
fromsqlalchemyimportcreate_engine,Column,Integer,String,Floatfromsqlalchemy.ormimportDeclarativeBase,Session# 1. 定义基类classBase(DeclarativeBase):pass# 2. 定义模型类（一张表对应一个类）classStudent(Base):# 表名（必填）__tablename__="students"# 列定义id=Column(Integer,primary_key=True,autoincrement=True)# 主键，自增name=Column(String(50),nullable=False)# 姓名，不能为空age=Column(Integer,default=18)# 年龄，默认18score=Column(Float)# 成绩，可为空# 可选：定义打印时的显示内容def__repr__(self):returnf"<Student(id={self.id}, name={self.name}, age={self.age})>"Column 常用参数：
| 参数 | 说明 | 示例 |
|---|---|---|
primary_key=True | 设为主键 | id |
autoincrement=True | 自动递增 | id |
nullable=False | 不允许为空 | 必填字段 |
unique=True | 值唯一不重复 | 邮箱、用户名 |
default=值 | 默认值 | default=18 |
index=True | 创建索引（加速查询） | 经常被查询的字段 |
常用列类型：
| SQLAlchemy 类型 | 对应 Python 类型 | 说明 |
|---|---|---|
Integer | int | 整数 |
String(长度) | str | 字符串 |
Float | float | 浮点数 |
Boolean | bool | 布尔值 |
DateTime | datetime | 日期时间 |
Text | str | 长文本 |
Enum | enum | 枚举值 |
2.2 创建表
# 创建引擎engine=create_engine("sqlite:///school.db",echo=True)# 根据所有继承 Base 的模型类，在数据库中创建对应的表Base.metadata.create_all(engine)# 注意：如果表已存在，不会重复创建（不会覆盖数据）2.3 添加记录（增 / Create）
# 方式一：逐条添加withSession(engine)assession:s1=Student(name="张三",age=20,score=88.5)session.add(s1)# 加入 sessionsession.commit()# 提交到数据库（必须！）# 方式二：批量添加withSession(engine)assession:students=[Student(name="李四",age=21,score=92.0),Student(name="王五",age=19,score=76.5),Student(name="赵六",age=22,score=85.0),Student(name="钱七",age=20,score=90.0),Student(name='孙八',age=18,score=79.0),Student(name='周九',age=19,score=82.5),Student(name='吴十',age=20),Student(name='郑一',age=21,score=87.0),Student(name='王二',age=22,score=91.5),Student(name='冯三',age=23,score=75.0),Student(name='陈四',age=24),Student(name='褚五',age=25,score=88.0),Student(name='卫六',age=18,score=80.5),Student(name='蒋七',age=19,score=93.0),Student(name='沈八',age=20,score=77.5),Student(name='韩九',age=21),Student(name='杨十',age=22,score=84.0),Student(name='朱一',age=23,score=90.5),Student(name='秦二',age=24,score=72.0),Student(name='尤三',age=25),Student(name='许四',age=18,score=86.5),Student(name='何五',age=19,score=78.0),Student(name='吕六',age=20,score=92.5),Student(name='施七',age=21),Student(name='张八',age=22,score=81.0),Student(name='孔九',age=23,score=76.0),Student(name='曹十',age=24),Student(name='严一',age=25,score=89.5),]session.add_all(students)# 一次性添加多条session.commit()# 方式三：add 后立即刷新获取 IDwithSession(engine)assession:s=Student(name='华二',age=18,score=None)session.add(s)session.flush()# 将 SQL 发送到数据库但不提交事务print(s,s.id,s.name)# 此时可以获取到自动生成的 idsession.commit()# 确认提交add vs flush vs commit 的区别：
add(obj) → 把对象标记为"待插入"（pending 状态） flush() → 把 SQL 发到数据库执行，但不提交事务。此时可以获取自增ID commit() → 提交事务，数据永久写入数据库2.4 完整示例
fromsqlalchemyimportcreate_engine,Column,Integer,String,Floatfromsqlalchemy.ormimportDeclarativeBase,SessionclassBase(DeclarativeBase):passclassStudent(Base):__tablename__="students"id=Column(Integer,primary_key=True,autoincrement=True)name=Column(String(50),nullable=False)age=Column(Integer,default=18)score=Column(Float)def__repr__(self):returnf"<Student(id={self.id}, name={self.name})>"# 创建引擎和表engine=create_engine("sqlite:///school.db",echo=False)Base.metadata.create_all(engine)# 添加学生withSession(engine)assession:session.add_all([Student(name="张三",age=20,score=88.5),Student(name="李四",age=21,score=92.0),Student(name="王五",age=19,score=76.5),Student(name="赵六",age=22,score=85.0),])session.commit()print("学生数据添加成功！")第3章：查询记录
3.1 查询所有记录
withSession(engine)assession:# 查询全部all_students=session.query(Student).all()forsinall_students:print(s)# 查询总数count=session.query(Student).count()print(f"共有{count}名学生")3.2 条件查询
withSession(engine)assession:# 精确匹配zhang=session.query(Student).filter(Student.name=="张三").first()# 多条件（AND）result=session.query(Student).filter(Student.age>=20,Student.score>=80).all()# 多条件（OR）fromsqlalchemyimportor_ result=session.query(Student).filter(or_(Student.name=="张三",Student.name=="李四")).all()# 模糊查询result=session.query(Student).filter(Student.name.like("%三%")# % 是通配符，匹配任意字符).all()# IN 查询result=session.query(Student).filter(Student.name.in_(["张三","李四","王五"])).all()# 范围查询result=session.query(Student).filter(Student.score.between(80,100)).all()# 判空 / 非空result=session.query(Student).filter(Student.score.is_(None)).all()result=session.query(Student).filter(Student.score.isnot(None)).all()3.3 排序
withSession(engine)assession:# 升序students=session.query(Student).order_by(Student.score).all()# 降序students=session.query(Student).order_by(Student.score.desc()).all()# 多级排序：先按年龄升序，同年龄按成绩降序students=session.query(Student).order_by(Student.age,Student.score.desc()).all()3.4 分页（LIMIT / OFFSET）
withSession(engine)assession:# 取前5条top5=session.query(Student).limit(5).all()# 跳过前10条，取5条（第三页）page2=session.query(Student).offset(10).limit(5).all()# 分页公式：第 N 页 = offset((N-1) * page_size).limit(page_size)page=2page_size=10result=session.query(Student)\.offset((page-1)*page_size)\.limit(page_size)\.all()3.5 聚合查询
fromsqlalchemyimportfuncwithSession(engine)assession:# 最大值、最小值、平均值、总和max_score=session.query(func.max(Student.score)).scalar()min_score=session.query(func.min(Student.score)).scalar()avg_score=session.query(func.avg(Student.score)).scalar()sum_score=session.query(func.sum(Student.score)).scalar()# 分组统计：每个年龄有多少人result=session.query(Student.age,func.count(Student.id)).group_by(Student.age).all()# 分组后过滤（HAVING）：人数大于2的年龄result=session.query(Student.age,func.count(Student.id)).group_by(Student.age).having(func.count(Student.id)>2).all()3.6 返回指定列（投影查询）
withSession(engine)assession:# 只查询姓名和成绩result=session.query(Student.name,Student.score).all()forname,scoreinresult:print(f"{name}:{score}")# 查询去重ages=session.query(Student.age).distinct().all()3.7 first() / one() / scalar() 的区别
# .first() — 返回第一条记录，没有则返回 None（最常用）student=session.query(Student).filter(Student.name=="张三").first()# .one() — 必须恰好一条，多了或少了都报错（严格校验时使用）student=session.query(Student).filter(Student.id==1).one()# .one_or_none() — 零或一条，多条则报错student=session.query(Student).filter(Student.name=="陈四").one_or_none()# .scalar() — 返回第一条的第一个字段值count=session.query(func.count(Student.id)).scalar()# .all() — 返回所有记录组成的列表students=session.query(Student).all()第4章：更新与删除
4.1 更新记录（改 / Update）
withSession(engine)assession:# 方式1：先查后改（推荐，有对象可以进一步操作）student=session.query(Student).filter(Student.name=="张三").first()ifstudent:student.age=21student.score=95.0session.commit()# 方式2：批量更新（不需先查出对象）session.query(Student).filter(Student.age<20).update({"age":20},synchronize_session="fetch")session.commit()4.2 删除记录（删 / Delete）
withSession(engine)assession:# 方式1：先查后删student=session.query(Student).filter(Student.name=="王五").first()ifstudent:session.delete(student)session.commit()# 方式2：批量删除session.query(Student).filter(Student.score==None).delete(synchronize_session="fetch")session.commit()4.3 CRUD 完整练习
fromsqlalchemyimportcreate_engine,Column,Integer,String,Floatfromsqlalchemy.ormimportDeclarativeBase,SessionclassBase(DeclarativeBase):passclassProduct(Base):__tablename__="products"id=Column(Integer,primary_key=True,autoincrement=True)name=Column(String(100),nullable=False)price=Column(Float,nullable=False)stock=Column(Integer,default=0)def__repr__(self):returnf"<Product(id={self.id}, name={self.name}, price={self.price}, stock={self.stock})>"engine=create_engine("sqlite:///shop.db",echo=False)Base.metadata.create_all(engine)# 增：添加商品withSession(engine)assession:session.add_all([Product(name="笔记本电脑",price=5999.0,stock=50),Product(name="无线鼠标",price=99.0,stock=200),Product(name="机械键盘",price=399.0,stock=80),Product(name="显示器",price=1999.0,stock=30),Product(name="RTX 5090 32GB（GDDR7）",price=28888.0,stock=0)])session.commit()# 查：查看所有商品withSession(engine)assession:products=session.query(Product).all()forpinproducts:print(f"{p.name}- 价格: ¥{p.price}, 库存:{p.stock}")# 改：鼠标涨价withSession(engine)assession:mouse=session.query(Product).filter(Product.name=="无线鼠标").first()mouse.price=129.0session.commit()# 删：删除库存为0的商品withSession(engine)assession:session.query(Product).filter(Product.stock==0).delete()session.commit()# 查：统计withSession(engine)assession:count=session.query(Product).count()print(f"当前共有{count}种商品")
