SqlAlchemy学习总结

1 安装

pip install sqlalchemy

2 创建Engine

2.1 数据库连接协议：

SQLite: sqlite:///./mydb.db "sqlite:///:memory:"(内存型)
MySQL: mysql://username:password@hostname/database_name
PostgreSQL: postgresql://username:password@hostname/database_name
SQL Server: mssql+pyodbc://username:password@hostname/database_name?driver=ODBC+Driver+17+for+SQL+Server
Oracle: oracle+cx_oracle://username:password@hostname:port/service_name
MariaDB: mariadb://username:password@hostname/database_name
MongoDB: mongodb+srv://username:password@cluster_name/database_name

2.2 示例语法：

from sqlalchemy import create_engine
# echo: python日志记录器输出
engine = create_engine("sqlite+pysqlite:///./mydb.db:", echo=True)

3 创建Connection

# from sqlalchemy import text
# 方式1：
with engine.connect() as conn:result = conn.execute(text("select 'hello world'"))print(result.all())# 方式2：
with session.connection() as conn:result = conn.execute(text("select 'hello world'"))print(result.all())

4 读取表数据

# 以行读取
from sqlalchemy import textwith engine.connect() as conn:result = conn.execute(text("SELECT x, y FROM some_table"))# 元组赋值
for x, y in result:
...# 整数索引
for row in result:
x = row[0]
y = row[1]# 属性名称
for row in result:
print(f"x: {row.x}  y: {row.y}")# 映射访问：
for dict_row in result.mappings():
x = dict_row["x"]
y = dict_row["y"]

5 execute传参

# ":num": 以"冒号"开头来定义占位参数，然后方法的第二个参数以“{}”的形式进行传参
result = conn.execute(text("SELECT x, y FROM some_table WHERE y > :num"), {"num": 2})# 多个参数：以“[{},{},...]”形式传参
conn.execute(text("INSERT INTO some_table (x, y) VALUES (:x, :y)"),[{"x": 11, "y": 12}, {"x": 13, "y": 14}],
)

6 MetaData

6.1 应用场景：

主要和表结构相关：reflect(表结构反射)、orm

# reflect：把数据库中的表结构信息反射到MetaData对象中
# orm：通过模型类创建表到数据库

6.2 reflect：

engine = create_engine('sqlite:///mydb.db')
# 创建 Metadata 对象
metadata_obj = MetaData(bind=engine)
# 从数据库中读取表格结构信息并反映到 Metadata 对象中
metadata_obj.reflect()
# 打印反映后的表格信息
for table in metadata_obj.tables.values():print("Table Name:", table.name)print("Columns:")for column in table.columns:print("\tColumn Name:", column.name)print("\tData Type:", column.type)# 为Table设置MetaData        
from sqlalchemy import Table, Column, Integer, String
user_table = Table("user_account",metadata_obj,Column("id", Integer, primary_key=True),Column("name", String(30)),Column("fullname", String),extend_existing=True
)
# user_table.primary_key
# user_table.c.name
# user_table.c.keys()

6.3 表创建：

metadata_obj.create_all(engine)# 未绑定engine：sqlalchemy.exc.UnboundExecutionError: MetaData object is not bound to an Engine or Connection.  Execution can not proceed without a database to execute against.
metadata_obj = MetaData(bind=engine)# 表已存在：sqlalchemy.exc.InvalidRequestError: Table 'user' is already defined for this MetaData instance.  Specify 'extend_existing=True' to redefine options and columns on an existing Table object.
user_table = Table("user",metadata_obj,...,extend_existing=True
)

7 声明式orm

7.1 1.x版本：

# 注：2.0版本语法参考sqlalchemy(2.0)小结
from sqlalchemy import Column
from sqlalchemy.orm import relationship
from sqlalchemy.orm import DeclarativeBaseclass Base(DeclarativeBase):passclass User(Base):__tablename__ = "user_account"id = Column(Integer, primary_key=True)name = Column(String(30), nullable=False)fullname = Column(String)addresses = relationship("Address", back_populates="user")class Address(Base):__tablename__ = "address"id = Column(Integer, primary_key=True)email_address = Column(String, nullable=False)user_id = Column(ForeignKey("user_account.id"), nullable=False)user = relationship("User", back_populates="addresses")

7.2 创建表：

Base.metadata.create_all(engine)

8 创建对象并持久化

8.1 创建会话：

# 以下两种方式创建会话本质没区别
# 方式1：
from sqlalchemy.orm import sessionmaker
session = sessionmaker(bind=engine)# 方式2：
from sqlalchemy.orm import Session
session = Session(bind=engine)

8.2 持久化：

from sqlalchemy.orm import Sessionwith Session(engine) as session:
spongebob = User(name="spongebob",fullname="Spongebob Squarepants",addresses=[Address(email_address="spongebob@sqlalchemy.org")],
)
sandy = User(name="sandy",fullname="Sandy Cheeks",addresses=[Address(email_address="sandy@sqlalchemy.org"),Address(email_address="sandy@squirrelpower.org"),],
)
patrick = User(name="patrick", fullname="Patrick Star")
session.add_all([spongebob, sandy, patrick])
session.commit()

9 CRUD(sqlalchemy.x)

9.1 简要：

在sqlalchemy包中，已经封装了crud的方法，可以很方便调用，从而不去写sql。
注意仅是将sql语句的方法化，执行这些方法并不是“执行sql”。
from sqlalchemy import insert,delete,update,select# 示例model实例，供下面参考（采用1.x版本语法）
class User(Base):__tablename__ = "users"id = Column(Integer,primary_key=True)name = Column(String(30))fullname = Column(String)

9.2 crud：

# insert
stmt = (insert(User).values(name="jd",fullname="joden")
)# delete
stmt = (delete(User).where(User.name=="jd")
)# update
stmt = (update(User).where(User.name=="u1").values(name="j1")
)# select
stmt = (select(User).where(name=="j1")
)

9.3 执行sql：

# 正如“简要”中提到，执行上述方法并没有执行sql（产生数据库变化）
# 如下语句将执行sql:insert_stmt = (insert(User).values(name="jd",fullname="joden")
)from sqlalchemy import create_engine
# url = "sqlite:///./t1.db"
engine = create_engine(url)
with engine.connect() as conn:conn.execute(insert_stmt)conn.commit()# 另外地，如果是select(查询操作)，也可以使用session.scalar(stmt)/scalars(stmt)来获取以实例为标量的查询集
from sqlalchemy.orm import Session
session = Session(engine)res = session.scalars(stmt)
for row in res:print(row.name,row.fullname)

10 CRUD(session.x)

10.1 简要：

与sqlalchemy.x中的不同的是，session.x中的crud方法会在当前会话中执行，在会话管理中作为数据库待定操作集，但是仍然需要我们去做会话(事务)提交才能真正做出数据库操作。

所以说简单较少的数据库操作可以简单通过sqlalchemy.x中方法，而较多的数据库操作集则用session.x中方法比较好。

10.2 方法：

方法	描述
add(obj)	将对象添加到会话中以进行持久化。
add_all(objs)	将多个对象添加到会话中以进行持久化。
delete(obj)	从会话中删除对象以进行持久化。
query()	创建一个查询对象，用于执行数据库查询。
get(obj,id)	根据id获取单个对象
merge(obj)	将对象合并到会话中以进行持久化。
commit()	提交会话中所有待定的事务。
flush()	将所有待定的数据库操作发送到数据库。
expire(obj)	将对象的属性从缓存中标记为过期。
refresh(obj)	重新加载对象的属性值。
rollback()	回滚会话中所有未提交的事务。
close()	关闭会话。

10.3 简单示例：

# select by id
sandy = session.get(User, 2)
# select
session.query(User).where(name="joden")
session.query(User.name,User.age).where(name="joden")# add 
u1 = User(name="j1",fullname="joden1")
u2 = User(name="j2",fullname="joden2")
# session.add(u1)
session.add_all([u1,u2])# delete
session.delete(u1)# update
u1.name = "jd1"# session commit
session.commit()

10.4 session.flush：

# 应用场景
（1）获取新创建对象的主键值：如果你需要在创建对象后立即获取其分配的主键值，那么在插入操作后执行 flush() 方法是必要的。否则，新创建的对象的主键值将不会在对象上更新，直到事务提交后才会更新。
（2）处理数据库的唯一约束和触发器：在某些情况下，数据库中的唯一约束或触发器可能会影响到插入或更新操作。执行 flush() 可以确保你及时地发现并处理任何由于约束或触发器引发的异常。
（3）在需要确保数据一致性的操作中：在一些需要确保数据一致性的复杂操作中，flush() 可以帮助你将操作分阶段提交到数据库，确保在整个操作过程中数据库的状态保持一致。

u4 = User(name="j4",fullname="joden4")
session.add(u4)
session.flush()
# print(u4.id)
session.commit()

11 execute

在sqlalchemy中使用execute(stmt)来执行sql。
要注意的是，可通过不同的实例来调用execute，下面我们将来了解几者的区别。

（1）engine.connect().execute(stmt)
（2）session.execute(stmt)
（3）session.connection().execute(stmt) # 本质上是engine.connect().execute(stmt)

所以区别仅在于Connection对象还是Session对象执行，很明显通过Connection对象方式需要自己创建管理连接，而Session对象方式直接执行则不需要(但是需要关闭当前“会话”)。

12 session状态

语法	描述
session.dirty	查看当前会话中修改对象
session.new	查看当前会话中添加对象

13 别名

13.1 字段别名：

for row in session.query(User.name.label("name_label")).all():print(row.name_label)

13.2 表别名：

from sqlalchemy.orm import aliaseduser_alias = aliased(User, name="user_alias")for row in session.query(user_alias).all():print(row.name)

14 过滤

14.1 filter_by:

for (name,) in session.query(User.name).filter_by(fullname="joden"):print(name)

14.2 filter：

# 较为灵活，支持python逻辑运算符
for (name,) in session.query(User.name).filter(User.fullname == "joden"):print(name)# 常用操作
query.filter(User.name == "ed")
query.filter(User.name != "ed")
query.filter(User.name.like('%ed%'))
query.filter(User.name.in_(["ed", "wendy", "jack"]))
query.filter(User.name.in_(session.query(User.name))

14.3 链式调用：

# 无论是filter还是filter_by都可以链式调用
session.query(User).filter(User.name == "j1").filter(User.fullname == "joden1")

14.5 and_：

from sqlalchemy import and_
query.filter(and_(User.name == 'ed', User.fullname == 'Ed Jones'))
# 等价下面两种方式
query.filter(User.name == 'ed', User.fullname == 'Ed Jones')
query.filter(User.name == 'ed').filter(User.fullname == 'Ed Jones')

14.6 text：

from sqlalchemy import text# 基础语法
session.query(User).filter(text("id<224"))
# 使用占位参数
session.query(User).filter(text("id<:value and name=:name")).params(value=3, name="jd")

15 join查询

15.1 隐式内连接：

session.query(User, Address).filter(User.id == Address.user_id).filter(Address.email_address == "jack@google.com").all()

15.2 显示内连接：

session.query(User).join(Address).filter(Address.email_address == "jack@google.com")

15.3 外连接：

# LEFT OUTER JOIN
query.outerjoin(User.addresses)  
query.outerjoin(Address)  # select_from 控制左右外连接
query = session.query(User, Address).select_from(Address).join(User)

16 表关系

16.1 一对多：

### 方式一：
# (1) 在两表中各定义
relationship(association_table_name,back_populates=association_table_attr 	# 在关联表中定义的关联属性
)
# (2) 在子表中定义外键属性(字段)
parent_table_id = Column(ForeignKey("parent_table_name.id"))### 方式二：
# (1) 在一表中定义
relationship(association_table_name,backref=association_table_attr
)
# (2) 在子表中定义外键属性(字段)
parent_table_id = Column(ForeignKey("parent_table_name.id"))### 示例代码
class TB1(Base):__tablename__ = 'tb1'id = Column(Integer,primary_key=True)name = Column(String(20))t2 = relationship("TB2",backref="t1")class TB2(Base):__tablename__ = 'tb2'id = Column(Integer,primary_key=True)name = Column(String(20))t1_id = Column(ForeignKey("tb1.id"))### 一对多关联查询
session.query(TB1).first().t2

16.2 多对多：

### 方式一：
association_table = Table("association_table",Base.metadata,Column("left_id", ForeignKey("left_table.id"), primary_key=True),Column("right_id", ForeignKey("right_table.id"), primary_key=True),
)class Parent(Base):__tablename__ = "left_table"id = Column(Integer, primary_key=True)children = relationship("Child", secondary=association_table, back_populates="parents")class Child(Base):__tablename__ = "right_table"id = Column(Integer, primary_key=True)parents = relationship("Parent", secondary=association_table, back_populates="children")### 方式二：
association_table = Table("association_table",Base.metadata,Column("left_id", ForeignKey("left_table.id"), primary_key=True),Column("right_id", ForeignKey("right_table.id"), primary_key=True),
)class Parent(Base):__tablename__ = "left_table"id = Column(Integer, primary_key=True)children = relationship("Child", secondary=association_table, backref="parents")class Child(Base):__tablename__ = "right_table"id = Column(Integer, primary_key=True)

16.3 一对一：

# 一对一通过添加uselist=False参数即可
class Parent(Base):__tablename__ = "parent_table"id = Column(Integer, primary_key=True)child = relationship("Child", back_populates="parent", uselist=False)class Child(Base):__tablename__ = "child_table"id = Column(Integer, primary_key=True)parent_id = Column(Integer, ForeignKey("parent_table.id"))# many-to-one side remains, see tip belowparent = relationship("Parent", back_populates="child")

17 backref和back_populates

17.1 区别：

backref 是 back_populates 的一种简化形式。backref 允许你在定义关系的同时，在另一个表格中自动创建一个反向引用属性，而不需要显式地在另一个表格的 relationship() 中设置 back_populates。

17.2 back_populates：

class Parent(Base):__tablename__ = 'parents'id = Column(Integer, primary_key=True)children = relationship("Child", back_populates="parent")class Child(Base):__tablename__ = 'children'id = Column(Integer, primary_key=True)parent_id = Column(Integer, ForeignKey('parents.id'))parent = relationship("Parent", back_populates="children")

17.3 backref：

class Parent(Base):__tablename__ = 'parents'id = Column(Integer, primary_key=True)children = relationship("Child", backref="parent")class Child(Base):__tablename__ = 'children'id = Column(Integer, primary_key=True)parent_id = Column(Integer, ForeignKey('parents.id'))

18 session.query

18.1 执行简单查询：

from sqlalchemy.orm import sessionmakerSession = sessionmaker(bind=engine)
session = Session()# 查询所有记录
result = session.query(User).all()for instance in session.query(User).order_by(User.id):print(instance.name, instance.fullname)

18.2 常用查询：

# 过滤查询： 你可以使用 `filter()` 方法对查询结果进行过滤。
result = session.query(User).filter(User.age > 25).all()# 排序查询： 使用 `order_by()` 方法对查询结果进行排序。
result = session.query(User).order_by(User.age.desc()).all()# 连接查询： 你可以执行连接查询来检索相关联的对象。
result = session.query(User, Address).join(Address).filter(User.id == Address.user_id).all()# 聚合查询： 使用 `func` 函数进行聚合查询。
from sqlalchemy import func
result = session.query(func.count(User.id)).scalar()# 子查询： 你可以在查询中使用子查询来执行复杂的操作。
subquery = session.query(Address.user_id).filter(Address.city == 'New York')
result = session.query(User).filter(User.id.in_(subquery)).all()

19 sqlalchemy(2.0)

19.1 声明映射类：

# 不同：使用python类型简化类型声明
# 注意：对于字符串类型字段，如果在mysql中创建则必须要给定长度(sqlite、PostgreSQL中则不用)from typing import List
from typing import Optionalfrom sqlalchemy import ForeignKey, String
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship
from sqlalchemy.orm import DeclarativeBaseclass Base(DeclarativeBase):passclass User(Base):__tablename__ = "user_account"id: Mapped[int] = mapped_column(primary_key=True)name: Mapped[str] = mapped_column(String(30))fullname: Mapped[Optional[str]]  # 必要时要补全addresses: Mapped[List["Address"]] = relationship(back_populates="user")def __repr__(self) -> str:return f"User(id={self.id!r}, name={self.name!r}, fullname={self.fullname!r})"class Address(Base):__tablename__ = "address"id: Mapped[int] = mapped_column(primary_key=True)email_address: Mapped[str] # 必要时要补全user_id = mapped_column(ForeignKey("user_account.id"))user: Mapped[User] = relationship(back_populates="addresses")def __repr__(self) -> str:return f"Address(id={self.id!r}, email_address={self.email_address!r})"

19.2 抽离重复类型字段：

# 比如主键用到的比较多且类型一致：整型、主键
# 2.0后可以使用类型注解的形式来对重复字段类型进行抽离定义，从而来简化字段声明from typing_extensions import Annotated
from sqlalchemy.orm import Mappedstr50 =  Annotated[str, mapped_column(String(50))]
intpk = Annotated[int, mapped_column(primary_key=True)]
user_fk = Annotated[int, mapped_column(ForeignKey("user_account.id"))]class User(Base):name:Mapped[str50]

20 数据迁移(alembic)

20.1 安装迁移依赖：

pip install alembic

20.2 迁移须知：

1. 使用alembic迁移配置需要导入基类Base，此时只能通过declarative_base()来创建，而不能通过DeclarativeBase来创建

2. 对于数据库迁移，我们始终需要两个信息：
   (1) 要迁移到的数据库
   (2) 要迁移的表

3. 所以使用alembic初始化我们需要配置以上的两个信息：
   初始化命令(项目根目录下执行，一般与app目录同级)：alembic init alembic

4. 配置两个信息：
   (1) 数据库信息：alembic.ini中的sqlalchemy.url进行配置
   (2) 要迁移的表：因为要迁移的表都与基类Base所关联，并且sqlalchemy创建表是通过Base.metadata来创建表到数据库，所以alembic要求我们配置Base.metadata，在alembic/env.py中的target_metadata进行配置

20.3 数据迁移：

# 安装好依赖，并且定义好所需的表模型后，我们将开始迁移
# 1. 初始化配置：alembic init alembic
# 2. 生成迁移文件：alembic revision --autogenerate -m [message]（该message是迁移备注信息，可以自己定义）
# 3. 开始迁移：alembic upgrade head

# 注：我们可以关注每一步迁移命令产生的变化或文件，从而有利于我们进一步了解迁移的本质