Python的SQLAlchemy库学习
1. 安装 SQLAlchemy
使用 pip 安装：
pip install sqlalchemy2. 核心概念
Engine：数据库连接入口，管理连接池。
Session：数据库交互的“句柄”，用于查询和事务管理。
ORM（对象关系映射）：将 Python 类映射到数据库表，对象实例映射到表中的行。
3. 连接数据库
创建引擎并连接数据库（以 SQLite 为例）：
from sqlalchemy import create_engine
# 连接 SQLite（内存数据库）
engine = create_engine('sqlite:///:memory:', echo=True) # echo=True 输出 SQL 日志4. 定义数据模型
使用 Declarative Base 定义模型类：
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import declarative_base
Base = declarative_base()
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
name = Column(String(30))
age = Column(Integer)
def __repr__(self):
return f"<User(name='{self.name}', age={self.age})>"5. 创建数据库表
通过 Base.metadata.create_all() 生成表：
Base.metadata.create_all(engine) # 创建所有继承 Base 的模型对应的表6. 增删改查（CRUD）
6.1 添加数据
from sqlalchemy.orm import sessionmaker
# 创建 Session 类并绑定引擎
Session = sessionmaker(bind=engine)
session = Session() # 实例化一个 Session
# 添加用户
new_user = User(name='Alice', age=30)
session.add(new_user)
session.commit() # 提交事务6.2 查询数据
# 查询所有用户
users = session.query(User).all()
print(users) # 输出: [<User(name='Alice', age=30)>]
# 条件查询（filter_by）
user = session.query(User).filter_by(name='Alice').first()
print(user) # 输出: <User(name='Alice', age=30)>
# 条件查询（filter）
user = session.query(User).filter(User.age > 25).first()
# 执行原生sql
result = session.execute(text("SELECT * FROM t_user"))
for row in result:
print(row)6.3 更新数据
user = session.query(User).filter_by(name='Alice').first()
user.age = 31
session.commit() # 提交更改6.4 删除数据
user = session.query(User).filter_by(name='Alice').first()
session.delete(user)
session.commit()7. 定义表关系
一对多关系（User 和 Address）
from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship
class Address(Base):
__tablename__ = 'addresses'
id = Column(Integer, primary_key=True)
email = Column(String(50))
user_id = Column(Integer, ForeignKey('users.id'))
# 定义反向关系（User.addresses 访问所有地址）
user = relationship("User", back_populates="addresses")
# 在 User 类中添加反向关系
User.addresses = relationship("Address", order_by=Address.id, back_populates="user")操作关联数据
# 添加用户及其地址
user = User(name='Bob', age=25)
user.addresses = [Address(email='bob@example.com'), Address(email='bob2@test.com')]
session.add(user)
session.commit()
# 查询用户及其地址
user = session.query(User).filter_by(name='Bob').first()
print(user.addresses) # 输出: [<Address(email='bob@example.com')>, ...]8. 高级查询
连接查询（Join）
# 查询用户及其地址（显式 Join）
result = session.query(User, Address).join(Address).filter(User.name == 'Bob').all()聚合函数
from sqlalchemy import func
# 计算平均年龄
avg_age = session.query(func.avg(User.age)).scalar()
print(avg_age)分组统计
# 按年龄分组统计用户数
counts = session.query(User.age, func.count(User.id)).group_by(User.age).all()9. 事务管理
使用上下文管理器自动提交或回滚：
# 事务管理
try:
with session.begin():
# 执行一些数据库操作
user1 = User(name='Tom', nickname='Tommy', openid='654321',
status=1, avatar='https://example.com/avatar2.jpg')
session.add(user1)
raise Exception("Simulated error") # 模拟错误，触发回滚
except Exception as e:
# 处理异常
print(f"Transaction failed: {e}")10. 使用 Alembic 迁移数据库
安装 Alembic：
pip install alembic初始化迁移环境：
alembic init migrations修改
alembic.ini和migrations/env.py配置数据库连接。生成迁移脚本：
alembic revision --autogenerate -m "Add new column"执行迁移：
alembic upgrade head
11. 调试技巧
启用 SQL 日志：
create_engine(..., echo=True)捕获异常：
try: session.commit() except SQLAlchemyError as e: session.rollback() print(f"Error: {e}")
学习资源
官方文档: SQLAlchemy Documentation
书籍: 《Essential SQLAlchemy》
教程: SQLAlchemy Tutorial by Mike Bayer
通过逐步实践这些步骤，你将能够熟练使用 SQLAlchemy 进行复杂的数据库操作！
阅读建议
