🐍 Python Day48：ORM 入门 - SQLAlchemy — 用面向对象操作数据库
🕐 预计用时：2-3 小时 ｜ 🎯 目标：掌握 SQLAlchemy 定义模型、Session、基本 CRUD、关系映射
📖 今日目录
1. 什么是 ORM？
ORM（Object-Relational Mapping）= 用 Python 类代替 SQL 语句操作数据库。
# 不用 ORM：写 SQL
cursor.execute("SELECT * FROM students WHERE age > 20 ORDER BY name")
rows = cursor.fetchall()
for row in rows:
print(row[1]) # 用索引访问
# 用 ORM：写 Python
students = session.query(Student).filter(Student.age > 20).order_by(Student.name).all()
for s in students:
print(s.name) # 用属性访问
# ORM 的好处：
# 1. 不用写 SQL（更 Pythonic）
# 2. 类型安全（编译时检查）
# 3. 数据库无关（切换数据库不改代码）
# 4. 自动防 SQL 注入
2. SQLAlchemy 安装与配置
# pip install sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.orm import declarative_base, sessionmaker
# 创建引擎（连接数据库）
# SQLite
engine = create_engine("sqlite:///school.db", echo=False)
# MySQL
# engine = create_engine("mysql+pymysql://root:123456@localhost/school", echo=False)
# echo=True 会打印所有 SQL 语句（调试用）
# 创建基类（所有模型都继承它）
Base = declarative_base()
# 创建 Session 工厂
Session = sessionmaker(bind=engine)
💡 create_engine 的 URL 格式：
SQLite: sqlite:///文件路径
MySQL: mysql+pymysql://用户:密码@主机:端口/数据库
PostgreSQL: postgresql+psycopg2://用户:密码@主机/数据库
3. 定义模型（Model）
from sqlalchemy import Column, Integer, String, Float, DateTime, ForeignKey
from sqlalchemy.orm import declarative_base
from datetime import datetime
Base = declarative_base()
class Student(Base):
"""学生模型"""
__tablename__ = "students" # 表名
id = Column(Integer, primary_key=True, autoincrement=True)
name = Column(String(100), nullable=False)
age = Column(Integer)
gender = Column(String(10), default="未知")
class_name = Column("class", String(50), nullable=False)
email = Column(String(200), unique=True)
created_at = Column(DateTime, default=datetime.now)
def __repr__(self):
return f"<Student(id={self.id}, name='{self.name}', age={self.age})>"
class Score(Base):
"""成绩模型"""
__tablename__ = "scores"
id = Column(Integer, primary_key=True, autoincrement=True)
student_id = Column(Integer, ForeignKey("students.id"), nullable=False)
subject = Column(String(50), nullable=False)
score = Column(Float)
def __repr__(self):
return f"<Score(subject='{self.subject}', score={self.score})>"
常用字段类型
| | |
|---|
| Integer | |
| String(n) | |
| Float | |
| Boolean | |
| DateTime | |
| Text | |
常用字段参数
| |
|---|
primary_key=True | |
autoincrement=True | |
nullable=False | |
unique=True | |
default=值 | |
ForeignKey("表.列") | |
4. 创建表
from sqlalchemy import create_engine
from sqlalchemy.orm import declarative_base
engine = create_engine("sqlite:///school.db", echo=True)
Base = declarative_base()
# ... 定义模型 ...
# 创建所有表
Base.metadata.create_all(engine)
print("✅ 表创建成功")
# 创建指定表
# Base.metadata.create_all(engine, tables=[Student.__table__])
5. Session — 数据库会话
from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind=engine)
# 方式1: 手动管理
session = Session()
# ... 操作 ...
session.commit()
session.close()
# 方式2: with 语句（推荐）
with Session() as session:
# ... 操作 ...
session.commit()
# 自动关闭
# Session 的生命周期：
# 1. 创建 Session
# 2. 添加/修改/删除对象
# 3. commit() 提交事务
# 4. 出错时 rollback() 回滚
# 5. close() 关闭
6. 增加 (Create)
from sqlalchemy.orm import Session
with Session(bind=engine) as session:
# 添加单个对象
student = Student(name="张三", age=20, gender="男", class_name="一班")
session.add(student)
session.commit()
print(f"插入 ID: {student.id}") # 自动获取生成的 ID
# 添加多个对象
students = [
Student(name="李四", age=22, gender="女", class_name="二班"),
Student(name="王五", age=21, gender="男", class_name="一班"),
Student(name="赵六", age=23, gender="男", class_name="二班"),
Student(name="钱七", age=20, gender="女", class_name="一班"),
]
session.add_all(students)
session.commit()
# 添加成绩
scores = [
Score(student_id=1, subject="数学", score=95),
Score(student_id=1, subject="英语", score=88),
Score(student_id=2, subject="数学", score=80),
Score(student_id=2, subject="英语", score=92),
]
session.add_all(scores)
session.commit()
print(f"✅ 插入完成")
7. 查询 (Read)
with Session(bind=engine) as session:
# 查询所有
students = session.query(Student).all()
for s in students:
print(f" {s.name} | {s.age}岁 | {s.class_name}")
# 查询单个
student = session.query(Student).filter_by(name="张三").first()
print(f"\n查询: {student}")
# 条件查询
adults = session.query(Student).filter(Student.age >= 21).all()
print(f"\n21岁以上: {[s.name for s in adults]}")
# 排序
ordered = session.query(Student).order_by(Student.age.desc()).all()
print(f"按年龄降序: {[s.name for s in ordered]}")
# 分页
page = session.query(Student).limit(2).offset(0).all()
print(f"第1页: {[s.name for s in page]}")
# 只查询特定列
names = session.query(Student.name, Student.age).all()
for name, age in names:
print(f" {name}: {age}岁")
# 统计
count = session.query(Student).count()
print(f"\n总人数: {count}")
# 聚合
from sqlalchemy import func
avg_age = session.query(func.avg(Student.age)).scalar()
print(f"平均年龄: {avg_age:.1f}")
filter 条件运算
# 等于
session.query(Student).filter(Student.name == "张三")
# 不等于
session.query(Student).filter(Student.name != "张三")
# 大于/小于
session.query(Student).filter(Student.age > 20)
session.query(Student).filter(Student.age < 20)
# 包含
session.query(Student).filter(Student.name.like("%张%"))
session.query(Student).filter(Student.name.in_(["张三", "李四"]))
# AND / OR
from sqlalchemy import and_, or_
session.query(Student).filter(and_(Student.age > 20, Student.gender == "男"))
session.query(Student).filter(or_(Student.age > 22, Student.gender == "女"))
# IS NULL
session.query(Student).filter(Student.email.is_(None))
session.query(Student).filter(Student.email.isnot(None))
8. 更新 (Update)
with Session(bind=engine) as session:
# 方式1: 查询后修改
student = session.query(Student).filter_by(name="张三").first()
student.age = 21
session.commit()
# 方式2: 批量更新
session.query(Student).filter(Student.class_name == "一班").update(
{"gender": "男"}
)
session.commit()
# 方式3: 基于原值更新
session.query(Student).filter(Student.name == "张三").update(
{Student.age: Student.age + 1}
)
session.commit()
9. 删除 (Delete)
with Session(bind=engine) as session:
# 删除单个
student = session.query(Student).filter_by(name="赵六").first()
if student:
session.delete(student)
session.commit()
# 批量删除
session.query(Score).filter(Score.score < 60).delete()
session.commit()
10. 今日小结
| | |
|---|
| query(Model).all() | SELECT * FROM table |
| filter(Model.col == val) | WHERE col = val |
| order_by(Model.col) | ORDER BY col |
| .limit(n).offset(m) | LIMIT n OFFSET m |
| .count() | SELECT COUNT(*) |
| session.add(obj) | INSERT INTO ... |
| obj.col = val | UPDATE SET col = val |
| session.delete(obj) | DELETE FROM ... |
🎯 练习建议：
1. 定义 Book 和 Author 模型，练习所有 CRUD 操作
2. 用 SQLAlchemy 重写 Day46 的通讯录管理
3. 实现一个"日志记录器"模型，自动记录创建和修改时间
📚 Day48 完成！明天学习 ORM 进阶 — 关系映射与复杂查询
