嘿，各位数据控们，准备好接受一场关于 SQLAlchemy 的深度洗礼了吗？今天咱们不整虚的，直接上干货，教你如何用 SQLAlchemy 的 ORM 和 Core，玩转那些让人头疼的复杂数据库操作。
开场白：为什么要学 SQLAlchemy？
想象一下，你辛辛苦苦用 Python 写了一个程序，结果发现要跟不同的数据库打交道，比如 MySQL、PostgreSQL、SQLite… 每种数据库的语法还不一样，简直让人抓狂！这时候，SQLAlchemy 就闪亮登场了，它就像一个万能翻译器，让你用统一的 Python 代码，操作各种不同的数据库。
SQLAlchemy 提供了两种主要的使用方式：
- ORM（Object Relational Mapper）： 把数据库表映射成 Python 对象，你可以像操作对象一样操作数据库，不用写 SQL 语句，方便快捷。
- Core： 更底层的 SQL 表达式语言，让你直接写 SQL 语句，灵活性更高，可以应对更复杂的场景。
咱们今天就来一起看看，怎么把这两种武器都练得炉火纯青。
第一部分：ORM – 对象关系映射的艺术
ORM 的核心思想就是“对象即表，属性即字段”。咱们先来创建一个简单的例子：
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
# 1. 定义数据库连接
engine = create_engine('sqlite:///:memory:', echo=True) # 使用内存数据库，方便演示
# 2. 定义基类
Base = declarative_base()
# 3. 定义模型类 (表结构)
class User(Base):
__tablename__ = 'users' # 表名
id = Column(Integer, primary_key=True)
name = Column(String(50))
age = Column(Integer)
email = Column(String(100))
def __repr__(self):
return f"<User(name='{self.name}', age={self.age}, email='{self.email}')>"
# 4. 创建表
Base.metadata.create_all(engine)
# 5. 创建 Session
Session = sessionmaker(bind=engine)
session = Session()
# 6. 添加数据
user1 = User(name='Alice', age=30, email='[email protected]')
user2 = User(name='Bob', age=25, email='[email protected]')
session.add_all([user1, user2])
session.commit()
# 7. 查询数据
users = session.query(User).all()
print(users)
# 8. 根据条件查询
user_alice = session.query(User).filter(User.name == 'Alice').first()
print(user_alice)
# 9. 更新数据
user_alice.age = 31
session.commit()
# 10. 删除数据
session.delete(user_alice)
session.commit()
# 11. 关闭 Session
session.close()这段代码做了这些事情：
- 连接数据库：
create_engine创建了一个数据库连接引擎，这里用了 SQLite 的内存数据库，方便演示。 - 定义基类：
declarative_base是所有模型类的基类。 - 定义模型类：
User类代表users表，每个属性代表表中的一个字段。__tablename__指定了表名，Column定义了字段的类型和约束。 - 创建表：
Base.metadata.create_all(engine)根据模型类的定义，创建数据库表。 - 创建 Session：
sessionmaker创建一个 Session 类，Session 对象用于与数据库交互。 - 添加数据： 创建
User对象，然后用session.add或session.add_all添加到数据库。session.commit()提交事务，把数据写入数据库。 - 查询数据：
session.query(User).all()查询所有User对象。 - 根据条件查询：
session.query(User).filter(User.name == 'Alice').first()根据条件查询，filter方法可以传入各种条件表达式。 - 更新数据： 直接修改
User对象的属性，然后session.commit()提交修改。 - 删除数据：
session.delete(user_alice)删除User对象，然后session.commit()提交删除。 - 关闭 Session:
session.close()释放资源。
ORM 的进阶玩法：关系映射
数据库表之间通常存在各种关系，比如一对一、一对多、多对多。ORM 可以轻松处理这些关系。
一对多关系：
比如一个用户可以有多篇文章，那么 User 和 Article 表之间就是一对多关系。
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy.ext.declarative import declarative_base
engine = create_engine('sqlite:///:memory:', echo=True)
Base = declarative_base()
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
name = Column(String(50))
articles = relationship("Article", back_populates="author") # 定义关系
def __repr__(self):
return f"<User(name='{self.name}')>"
class Article(Base):
__tablename__ = 'articles'
id = Column(Integer, primary_key=True)
title = Column(String(100))
content = Column(String(200))
user_id = Column(Integer, ForeignKey('users.id')) # 外键
author = relationship("User", back_populates="articles") # 定义关系
def __repr__(self):
return f"<Article(title='{self.title}')>"
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()
# 创建用户
user1 = User(name='Alice')
# 创建文章，并关联到用户
article1 = Article(title='SQLAlchemy 入门', content='SQLAlchemy 的 ORM 很好用', author=user1)
article2 = Article(title='Python 技巧', content='Python 的各种小技巧', author=user1)
session.add_all([user1, article1, article2])
session.commit()
# 查询用户及其文章
user = session.query(User).filter(User.name == 'Alice').first()
print(user)
for article in user.articles:
print(article)
# 查询文章及其作者
article = session.query(Article).filter(Article.title == 'SQLAlchemy 入门').first()
print(article)
print(article.author)
session.close()关键点：
relationship定义了关系，back_populates指定了反向关系。ForeignKey定义了外键，指向关联表的id字段。- 通过
user.articles可以访问用户的所有文章，通过article.author可以访问文章的作者。
多对多关系：
比如一个学生可以选修多门课程，一门课程也可以被多个学生选修，那么 Student 和 Course 表之间就是多对多关系。需要一个中间表来存储学生和课程的关联关系。
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey, Table
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy.ext.declarative import declarative_base
engine = create_engine('sqlite:///:memory:', echo=True)
Base = declarative_base()
# 定义中间表
association_table = Table('association', Base.metadata,
Column('student_id', Integer, ForeignKey('students.id')),
Column('course_id', Integer, ForeignKey('courses.id'))
)
class Student(Base):
__tablename__ = 'students'
id = Column(Integer, primary_key=True)
name = Column(String(50))
courses = relationship("Course", secondary=association_table, back_populates="students")
def __repr__(self):
return f"<Student(name='{self.name}')>"
class Course(Base):
__tablename__ = 'courses'
id = Column(Integer, primary_key=True)
name = Column(String(50))
students = relationship("Student", secondary=association_table, back_populates="courses")
def __repr__(self):
return f"<Course(name='{self.name}')>"
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()
# 创建学生
student1 = Student(name='Alice')
student2 = Student(name='Bob')
# 创建课程
course1 = Course(name='Python')
course2 = Course(name='SQLAlchemy')
# 关联学生和课程
student1.courses.append(course1)
student1.courses.append(course2)
student2.courses.append(course1)
session.add_all([student1, student2, course1, course2])
session.commit()
# 查询学生及其课程
student = session.query(Student).filter(Student.name == 'Alice').first()
print(student)
for course in student.courses:
print(course)
# 查询课程及其学生
course = session.query(Course).filter(Course.name == 'Python').first()
print(course)
for student in course.students:
print(student)
session.close()关键点：
Table定义了中间表，包含两个外键，分别指向关联表的id字段。secondary指定了中间表。- 通过
student.courses可以访问学生选修的课程，通过course.students可以访问选修该课程的学生。
第二部分：Core – SQL 表达式语言的威力
ORM 很好用，但有些复杂的 SQL 语句，用 ORM 写起来比较麻烦。这时候，SQLAlchemy Core 就派上用场了。
Core 提供了 SQL 表达式语言，让你用 Python 代码构建 SQL 语句。
from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String, select, insert, update, delete
# 1. 定义数据库连接
engine = create_engine('sqlite:///:memory:', echo=True)
# 2. 定义 MetaData
metadata = MetaData()
# 3. 定义表
users_table = Table('users', metadata,
Column('id', Integer, primary_key=True),
Column('name', String(50)),
Column('age', Integer),
Column('email', String(100))
)
# 4. 创建表
metadata.create_all(engine)
# 5. 连接数据库
connection = engine.connect()
# 6. 插入数据
insert_stmt = insert(users_table).values(name='Alice', age=30, email='[email protected]')
connection.execute(insert_stmt)
insert_stmt = insert(users_table)
connection.execute(insert_stmt, [
{'name': 'Bob', 'age': 25, 'email': '[email protected]'},
{'name': 'Charlie', 'age': 35, 'email': '[email protected]'}
])
# 7. 查询数据
select_stmt = select([users_table])
result = connection.execute(select_stmt)
for row in result:
print(row)
# 8. 根据条件查询
select_stmt = select([users_table]).where(users_table.c.name == 'Alice')
result = connection.execute(select_stmt)
for row in result:
print(row)
# 9. 更新数据
update_stmt = update(users_table).where(users_table.c.name == 'Alice').values(age=31)
connection.execute(update_stmt)
# 10. 删除数据
delete_stmt = delete(users_table).where(users_table.c.name == 'Alice')
connection.execute(delete_stmt)
# 11. 关闭连接
connection.close()这段代码做了这些事情：
- 连接数据库：
create_engine创建了一个数据库连接引擎。 - 定义 MetaData：
MetaData是所有表的元数据容器。 - 定义表：
Table定义了表结构，Column定义了字段的类型和约束。 - 创建表：
metadata.create_all(engine)根据表定义，创建数据库表。 - 连接数据库：
engine.connect()创建一个数据库连接。 - 插入数据：
insert函数创建插入语句，values方法指定插入的值。 - 查询数据：
select函数创建查询语句，connection.execute执行查询语句。 - 根据条件查询：
where方法添加查询条件，users_table.c.name表示users表的name字段。 - 更新数据：
update函数创建更新语句，values方法指定更新的值。 - 删除数据：
delete函数创建删除语句。 - 关闭连接：
connection.close()释放资源。
Core 的高级技巧：Join 和 Aggregate
Core 可以轻松处理复杂的 SQL 查询，比如 Join 和 Aggregate。
Join：
from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String, ForeignKey, join, select
engine = create_engine('sqlite:///:memory:', echo=True)
metadata = MetaData()
users_table = Table('users', metadata,
Column('id', Integer, primary_key=True),
Column('name', String(50))
)
addresses_table = Table('addresses', metadata,
Column('id', Integer, primary_key=True),
Column('email', String(100)),
Column('user_id', Integer, ForeignKey('users.id'))
)
metadata.create_all(engine)
connection = engine.connect()
# 插入数据
connection.execute(users_table.insert(), [
{'name': 'Alice'},
{'name': 'Bob'}
])
connection.execute(addresses_table.insert(), [
{'email': '[email protected]', 'user_id': 1},
{'email': '[email protected]', 'user_id': 2}
])
# Join 查询
join_stmt = join(users_table, addresses_table, users_table.c.id == addresses_table.c.user_id)
select_stmt = select([users_table.c.name, addresses_table.c.email]).select_from(join_stmt)
result = connection.execute(select_stmt)
for row in result:
print(row)
connection.close()关键点：
join函数创建 Join 语句，指定关联的表和关联条件。select_from方法指定从哪个 Join 语句中查询数据。
Aggregate：
from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String, func, select
engine = create_engine('sqlite:///:memory:', echo=True)
metadata = MetaData()
users_table = Table('users', metadata,
Column('id', Integer, primary_key=True),
Column('name', String(50)),
Column('age', Integer)
)
metadata.create_all(engine)
connection = engine.connect()
# 插入数据
connection.execute(users_table.insert(), [
{'name': 'Alice', 'age': 30},
{'name': 'Bob', 'age': 25},
{'name': 'Charlie', 'age': 35}
])
# Aggregate 查询
select_stmt = select([func.avg(users_table.c.age)])
result = connection.execute(select_stmt)
for row in result:
print(row)
connection.close()关键点：
func模块提供了各种聚合函数，比如avg、sum、count。
第三部分：ORM + Core – 混合使用，天下无敌
ORM 和 Core 各有优点，可以结合使用，发挥更大的威力。
比如，可以用 ORM 定义模型类，然后用 Core 写复杂的查询语句。
from sqlalchemy import create_engine, Column, Integer, String, func
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
engine = create_engine('sqlite:///:memory:', echo=True)
Base = declarative_base()
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
name = Column(String(50))
age = Column(Integer)
def __repr__(self):
return f"<User(name='{self.name}', age={self.age})>"
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()
# 插入数据
user1 = User(name='Alice', age=30)
user2 = User(name='Bob', age=25)
user3 = User(name='Charlie', age=35)
session.add_all([user1, user2, user3])
session.commit()
# 使用 Core 查询
select_stmt = session.query(func.avg(User.age))
result = select_stmt.scalar()
print(result)
session.close()关键点：
session.query可以传入 Core 的 SQL 表达式。scalar方法返回查询结果的第一个值。
总结：SQLAlchemy 的最佳实践
- ORM 适合简单的 CRUD 操作，Core 适合复杂的 SQL 查询。
- 尽量使用 ORM，只有在 ORM 无法满足需求时，才使用 Core。
- 熟悉 SQL 表达式语言，可以让你更好地理解 ORM 的底层原理。
- 多看 SQLAlchemy 的官方文档，里面有很多高级用法。
最后的彩蛋：一些常用的 SQLAlchemy 技巧
| 技巧 | 说明 |
|---|---|
使用 hybrid_property |
可以把 Python 方法变成模型类的属性，方便计算和访问。 |
使用 event 监听数据库事件 |
可以在数据插入、更新、删除前后，执行自定义的逻辑，比如自动更新时间戳。 |
使用 alembic 管理数据库迁移 |
可以方便地创建和执行数据库迁移脚本，保证数据库结构的一致性。 |
使用 pytest 进行单元测试 |
可以对 SQLAlchemy 代码进行单元测试，保证代码的质量。 |
好了，今天的 SQLAlchemy 讲座就到这里。希望大家能够掌握 SQLAlchemy 的 ORM 和 Core，成为数据库操作的高手！ 记住，熟能生巧，多练习才能真正掌握这些技巧。 祝大家编程愉快！