SQLAlchemy
ORM 工具库
https://docs.sqlalchemy.org/en/20/index.html
ORM 快速入门
声明模型
Declare Models
我们定义模块级构造，这些构造将形成我们将从数据库查询的结构。
这种结构被称为声明式映射(Declarative Mapping)，它同时定义了Python对象模型，以及描述特定数据库中存在或将存在的真实SQL表的数据库元数据:
from typing import Optional, List
from sqlalchemy import ForeignKey, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
# mapped_column Mapped 为2.0 新的格式 Python类型和SQL类型之间的关联可以使用类型注释映射自定义。
# 映射从一个基类开始，上面称为base，并通过针对DeclarativeBase类创建一个简单的子类来创建。
空性取决于是否使用了Optional[]类型修饰符。
可以使用右侧mapped_column()指令中的SQLAlchemy类型对象来指示更具体的类型信息，例如上面在User.name列中使用的字符串数据类型。
class Base(DeclarativeBase):
__abstract__ = True
# 定义共有属性
# 每个列的数据类型首先取自与每个映射注释相关联的Python数据类型;
# int表示INTEGER, str表示VARCHAR，等等。
id: Mapped[int] = mapped_column(primary_key=True)
# 定义__repr__方法不是必要的
def __repr__(self) -> str:
return f"{self.__tablename__} (id={self.id!r}"
class User(Base):
__tablename__ = "user"
name: Mapped[str] = mapped_column(String(30))
fullname: Mapped[Optional[str]] # Optional 表示可为空
# relationship 表示两表之间的关系，方便关联查询。List 表示一对多 。非数据库约束。
addresses: Mapped[List["Address"]] = relationship(
back_populates="user", cascade="all, delete-orphan"
)
class Address(Base):
__tablename__ = "address"
email_address: Mapped[str]
user_id: Mapped[int] = mapped_column(ForeignKey("user_account.id"))
user: Mapped["User"] = relationship(back_populates="addresses")
# back_populates 表示反向关系，即关联表中可以通过 addresses 反向查找
类型补充
除了 str int 外 还有时间
from datetime import datetime
created_at: Mapped[datetime] = mapped_column(
insert_default=func.utc_timestamp(), default=None
)
另一种方法
在 SQLAlchemy 中，定义表有两种主要的方式：使用 Table 类和使用继承 declarative_base() 后的基类。
from sqlalchemy import MetaData, Table, Column, Integer, String
metadata = MetaData()
users = Table('users', metadata,
Column('id', Integer, primary_key=True),
Column('name', String),
Column('age', Integer)
)
-----------------------
使用 Table 类：使用 Table 类可以直接定义表的结构，包括表的名称、列名、数据类型等信息。这种方式更加灵活，适用于在代码中动态地定义表结构，或者对现有的数据库进行操作。你可以通过 metadata.create_all() 方法来创建表，也可以使用 engine.execute() 方法直接执行 SQL 语句来创建表。
这两种方式各有优缺点，选择取决于你的具体需求和开发习惯。通常情况下，如果你习惯使用对象来表示数据库中的表，并且希望利用 SQLAlchemy 提供的 ORM（对象关系映射）功能来简化数据库操作，那么推荐使用继承 declarative_base() 后的基类。如果你需要更灵活地控制数据库结构，或者需要对现有数据库进行动态的、基于代码的操作，那么使用 Table 类可能更合适。
表关系
数据本身对表之间约束关系为外键 ForeignKey 这个会在数据库中作为实际字段体现 。ForeignKey 本身也是一个逻辑关系的体验 存在的为从表。一般关联到主表的主键
ORM 中可以只表示表的逻辑关系使用 relationship 这个是在数据中不会实际体现的。 逻辑关系分为 一对多 多对多 一对一 ，其中多对多是通过中间表来实现。
# 外键 约束 表示将user_account 的主键id 作为字段
mapped_column(ForeignKey("user_account.id"))
# 逻辑关系
addresses: Mapped[List["Address"]] = relationship(
back_populates="user", cascade="all, delete-orphan"
)
多对多关系
Base = declarative_base()
# 定义关联表
association_table = Table('association', Base.metadata,
Column('user_id', Integer, ForeignKey('users.id')),
Column('group_id', Integer, ForeignKey('groups.id'))
)
注意 多对多关系 使用 relationship 会有变化
groups = relationship("Group", secondary=association_table, back_populates="users")
创建引擎
Create an Engine
引擎 是一个工厂factory 用来创建数据连接。
from sqlalchemy import create_engine
engine = create_engine("sqlite://", echo=True) # echo=True 表示sql 记录会输出
连接其他数据库
# 连接 mysql pymysql 是需要的驱动库 需要下载
'mysql+pymysql://root:your_password@localhost:3306/your_database?charset=utf8mb4'
发出创建表DDL
Emit CREATE TABLE DDL
# 这个操作将在数据生成sql 并在数据库中创建 对应表
Base.metadata.create_all(engine) # 已经存在则不会创建
数据操作
新增数据
Create Objects and Persis 创建对象和持久化
实例化 表对象 并在数据库中添加数据
from .model import engine, User, Address
from sqlalchemy.orm import Session # 新建连接需要Session 确认更改需要commit
with Session(engine) as session:
spongebob = User(
name="spongebob",
fullname="Spongebob Squarepants",
addresses=[Address(email_address="[email protected]")], # 自动创建关联对象
)
sandy = User(
name="sandy",
fullname="Sandy Cheeks",
addresses=[
Address(email_address="[email protected]"),
Address(email_address="[email protected]"),
],
)
patrick = User(name="patrick", fullname="Patrick Star")
session.add_all([spongebob, sandy, patrick])
session.commit()
查询
Simple SELECT
通过orm 方式也可以代替sql 做查询操作
from sqlalchemy import select
session = Session(engine)
stmt = select(User).where(User.name.in_(["spongebob", "sandy"])) # 相当于生成了查询语句
for user in session.scalars(stmt): # 执行查询是 session.scalars(stmt)
print(user)
多表查询
SELECT with JOIN
stmt = (
select(Address)
.join(Address.user)
.where(User.name == "sandy")
.where(Address.email_address == "[email protected]")
)
sandy_address = session.scalars(stmt).one()
---------------------------------
SELECT address.id, address.email_address, address.user_id
FROM address JOIN user_account ON user_account.id = address.user_id
WHERE user_account.name = ? AND address.email_address = ?
[...] ('sandy', '[email protected]')
改数据
先查再改
stmt = select(User).where(User.name == "patrick")
patrick = session.scalars(stmt).one()
patrick.addresses.append(Address(email_address="[email protected]"))
sandy_address.email_address = "[email protected]"
session.commit()
删除
Some Deletes
sandy = session.get(User, 2)
sandy.addresses.remove(sandy_address) # 确认后删除
session.flush() # 提交 flush 相当于暂存提高 性能 减少 commit 减少资源消耗 这里使用flush 确认remove删除 否则不会删除
session.delete(patrick) # 立刻删除
session.commit()