SQLAlchemy

ORM

ORM(Object-Relational Mapping, 对象关系映射) 是一种编程方式

映射

ORM通过定义类(Class)和表(Table)的映射关系
数据库表 users <---> 编程语言中的类 User
表中的一行记录 <---> 一个对象实例
表的字段 <---> 对象的属性

实例:

from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_baseBase = declarative_base()# 定义 User 类（映射到数据库的 users 表）
class User(Base):__tablename__ = 'users'  # 表名id = Column(Integer, primary_key=True)  # 字段映射name = Column(String(50))age = Column(Integer)# 操作数据库（无需直接写 SQL）
user = User(name="Alice", age=25)  # 创建对象（对应插入一条记录）

ORM的作用

• 自动生成sql避免手写
• 同一套代码可以用在多个数据库 只需要修改ORM配置

SQLAlchemy

SQLAlchemy分为两层 ORM和Core
ORM 隐藏DBAPI细节 提供更高层次的抽象
Core 直接使用DBAPI

建立连接-引擎(engine)

无论是ORM和Core都需要用到engine

from sqlalchemy import create_engine
engine = create_engine("sqlite+pysqlite:///:memory:")

1. sqlite 用哪个数据库
2. pysqlite DBAPI 如果省略会用指定的数据库的默认DBAPI
3. /:memory: 定位

使用事务和DBAPI

获取连接并提交更改

1. 逐个提交

   from sqlalchemy import tect
   with engine.connect() as conn:result = conn.execute(text("select 'hello world'"))print(result.all())conn.commit()   # 提交
   

   在代码块中使用conn.commit()提交事务在SQLAlchemy被称为逐个提交

2. 一次开始(推荐)
   声明connect代码块为事务代码块 使用engine.begin()

   with engine.begin() as conn:result = conn.execute(....)
   

   如果没有发生异常则提交(commit) 如果发生异常就回滚(rollback)

• 代码中的result最好在该with代码块中使用
• DBAPI核心步骤: 创建连接 → 创建游标 → 执行 SQL → 提交/回滚 → 关闭连接
• 使用with上下文可以防止忘记关闭连接或者游标耗尽资源

语句执行基础

上面的result是一个Result对象，表示结果行的可迭代对象

1. 元组赋值

   for column1, column2, column3 ... in result:...
   

2. 整数索引

   for row in result:coloum1 = row[1]column2 = row[2]...
   

3. 属性名称

   for row in result:column1 = row.column1_namecolumn2 = row.column2_name...
   

4. 映射访问

   for dict_row in result.mappings():column1 = dict_row["column1_name"]column2 = dict_row["column2_name"]...
   

发送参数

1. 单个参数

   with engine.connect() as conn:result = conn.execute(text("SELECT x, y FROM some_table WHERE y > :y"), {"y": 2})for row in result:print(f"x: {row.x}  y: {row.y}")
   

2. 多个参数

   with engine.connect() as conn:conn.execute(text("INSERT INTO some_table (x, y) VALUES (:x, :y)"),[{"x": 11, "y": 12}, {"x": 13, "y": 14}],)conn.commit()
   

   参数是一个列表，说明要对列表中的每一个参数集都执行一次SQL语句
   这种风格称为executemany

在语句中有RETURNING时后者不支持返回RETURNING的值

使用ORM执行

使用ORM的时候，基本的事务/数据库交互对象被称为Session，Session实际上在内部使用Connection 并用其来发出sql语句

from sqlalchemy.orm import Session
stmt = text("SELECT x, y FROM some_table WHERE y > :y ORDER BY x, y")
with Session(engine) as session:result = session.execute(stmt, {"y": 6})for row in result:print(f"x: {row.x}  y: {row.y}")

使用数据库元数据

使用Table对象设置MetaData

MetaData可以理解为整个数据库的一个目录，记录了:
1. 有哪些表(table)
2. 每个表的结构（列名、数据类型、约束等）
3. 表之间的关系（如外键）

什么时候创建MetaData?
一般在最开始创建一个MetaData,一般情况下只需要一个

from sqlalchemy import MetaData, Table, Column, Integer, String
user_table = Table("user_account",metadata_obj,Column("id", Integer, primary_key=True),Column("name", String(30)),Column("fullname", String),
)

Table的组件

• Table 表示数据库表并将自身分配给MetaData集合
• Column 表示数据库表中的列并将自身分配给Table, Column通常包含名称和类型 通常通过父Table.c访问
• Integer, String 类型，可以传递给Column

声明简单约束

上面的Column指定了一个primary_key，可以通过Table.primary_key访问

from sqlalchemy import ForeignKey
address_table = Table("address",metadata_obj,Column("id", Integer, primary_key=True),Column("user_id", ForeignKey("user_account.id"), nullable=False),Column("email_address", String, nullable=False),
)

ForeighKey是外键约束 nullable代表SQL中NOT NULL的约束

向数据库发送DDL (Data Definition Language, 数据定义语言)

已经构建了一个metadata_obj 包含了两个表 每个表中都有Column
将engine传递给metadata_obj就可以创建数据库

metadata_obj.create_all(engine)

另外MetaData.drop_all()可以删除所有架构

使用ORM声明形式定义MetaData

建立声明基类

from sqlalchemy.orm import DeclarativeBase
class Base(DeclarativeBase):pass

声明映射类

from typing import List
from typing import Optional
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationshipclass 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})"

User和Address被称为映射类

• 每个类都引用一个Table对象 __tablename__是表名
• 这种形式被称为声明式表配置
• 使用mapped_column()构造指示Table中的列，结合基于Mapped类型的类型注解, 此对象将生成应用于Table构造的Column对象
• 对于具有简单数据类型且没有其他选项的列,可以仅指示Mapped类型注解，使用简单的Python类型（如 int 和 str）来表示 Integer 和 String
• 可以使用 <type> | None代表可选，或者显式地使用mapped_column.nullable参数
• 显式类型注解的使用是完全可选的,也可以在没有注解的情况下使用 mapped_column()
• 如果我们不声明自己的 __init__() 方法，则会自动为类提供一个 __init__() 方法。此方法的默认形式接受所有属性名称作为可选关键字参数, 如: sandy = User(name="sandy", fullname="Sandy Cheeks")

从ORM映射向数据库发送DDL

Base.metadata.create_all(engine)

处理数据

使用INSERT语句

使用 Core 和 ORM 进行批量操作的时候 SQL INSERT语句直接使用insert()生成 此函数生成Insert的心实例，代表SQL语句中的INSERT

insert() SQL 表达式构造

from sqlalchemy import insert
stmt = insert(user_table).values(name="spongebob", fullname="Spongebob Squarepants")

可以直接用print()查看正在生成的内容
字符串化形式是通过生成对象的 Compiled 形式来创建的, 其中包含语句的数据库特定字符串 SQL 表示形式, 可以使用.compiled()获得该对象

compiled = stmt.compiled()

要查看name和fullname绑定的参数 使用compiled.params

执行语句

将一行INSERT到user_table中

with engine.begin() as conn:result = conn.execute(stmt)

返回值是一个元组：(主键值, )

INSERT通常自动生成"values"子句

如果不使用.values()将得到表中每一行的insert

INSERT INTO user_account (id, name, fullname) VALUES (:id, :name, :fullname)

with engine.connect() as conn:result = conn.execute(insert(user_table),[{"name": "sandy", "fullname": "Sandy Cheeks"},{"name": "patrick", "fullname": "Patrick Star"},],)conn.commit()

会将两条都插入，是一个executemany

INSERT...RETURNING

insert_stmt = insert(address_table).returning(address_table.c.id, address_table.c.email_address
)
print(insert_stmt)

显式地使用.returning()方法构建RETURNING子句

使用SELECT语句

对于ORM和Core select()函数生成一个SELECT构造 传递给ORM的Session.execute()或者Core的Connection.execute() 发出SELECT事务后返回Result对象

select() SQL表达式构造

from sqlalchemy import select
stmt = select(user_table).where(user_table.c.name == "spongebob")
print(stmt)结果：
SELECT user_account.id, user_account.name, user_account.fullname
FROM user_account
WHERE user_account.name = :name_1

执行后的返回值依旧是Result对象

设置COLUMN和FORM子句

select()接受位置元素，这些元素可以是任意个Column或者Table, 以及各种兼容的对象
如下：

>>> print(select(user_table))SELECT user_account.id, user_account.name, user_account.fullname
FROM user_account

>>> print(select(user_table.c.name, user_table.c.fullname))SELECT user_account.name, user_account.fullname
FROM user_account

使用FormClause.c的集合时这种写法也是正确的

>>> print(select(user_table.c["name", "fullname"]))SELECT user_account.name, user_account.fullname
FROM user_account

选择 ORM 实体和列

选择ORM实体User的呈现结果和选择user_table一样

>>> print(select(User))SELECT user_account.id, user_account.name, user_account.fullname
FROM user_account

当使用Session.execute()执行上面的语句时，当选择完整实体时而不是user_table,一个重要的区别就是 实体本身作为每行中的单个元素返回

或者使用ORM实体中的属性作为不同列

>>> print(select(User.name, User.fullname))SELECT user_account.name, user_account.fullname
FROM user_account

也可以混合使用属性和完整实体

>>> session.execute(
...     select(User.name, Address).where(User.id == Address.user_id).order_by(Address.id)
... ).all()SELECT user_account.name, address.id, address.email_address, address.user_id
FROM user_account, address
WHERE user_account.id = address.user_id ORDER BY address.id
[...] ()[('spongebob', Address(id=1, email_address='spongebob@sqlalchemy.org')),
('sandy', Address(id=2, email_address='sandy@sqlalchemy.org')),
('sandy', Address(id=3, email_address='sandy@squirrelpower.org'))]

从标记的SQL表达式中选择

>>> from sqlalchemy import func, cast
>>> stmt = select(
...     ("Username: " + user_table.c.name).label("username"),
... ).order_by(user_table.c.name)
>>> with engine.connect() as conn:
...     for row in conn.execute(stmt):
...         print(f"{row.username}")Username: patrick
Username: sandy
Username: spongebob

给特定的形式Username: user_table.c.name一个标签username

WHERE子句

使用Select.where()来生成子句

select(user_table).where(user_table.c.name == "squidward")

多个where子句

1. 多次调用Select.where()方法
   如Select.where().where().where()
2. 一个Select.where()也接受多个条件语句,
   如：

   select(address_table.c.email_address).where(user_table.c.name == "squidward",address_table.c.user_id == user_table.c.id,
   )
   

3. 使用AND和OR
   提供了and_()和or_()函数

   from sqlalchemy import and_, or_
   print(select(Address.email_address).where(and_(or_(User.name == "squidward", User.name == "sandy"),Address.user_id == User.id,))
   )
   

针对单个实体的简单"相等"比较，可以使用Select.filter_by()方法, 如：

print(select(User).filter_by(name="spongebob", fullname="Spongebob Squarepants"))

显式的FORM子句和JOIN

先前的FORM子句在Select中都是由传递的元素推断出来的
要将两个表join在一起有两种办法

1. 显式地指定左边和右边的表 Select.join_from()

   select(user_table.c.name, address_table.c.email_address).join_from(user_table, address_table
   )
   

2. 指定右边的表Select.join(),左边的表会自动推断出来

   select(address_table.c.email_address).select_from(user_table).join(address_table)
   

在简单的外键情况下，join 的 ON 子句也会为我们推断出来

使用UPDATE和DELETE语句

update() SQL表达式构造

update()针对单个表发出UPDATE且不返回任何行 但是可以使用RETURNING
基本的update语句

update(user_table).where(user_table.c.name == "patrick").values(fullname="Patrick the Star")

为了使用executemany的风格 可以使用bindparam()方法

from sqlalchemy import bindparam
stmt = (update(user_table).where(user_table.c.name == bindparam("oldname")).values(name=bindparam("newname"))
)
with engine.begin() as conn:conn.execute(stmt,[{"oldname": "jack", "newname": "ed"},{"oldname": "wendy", "newname": "mary"},{"oldname": "jim", "newname": "jake"},],)

其中将字符串替换成bindparam("参数的名称")

delete() SQL表达式构造

delete()和update()构造相似，传统上不返回值 但是可以使用RETURNING

from sqlalchemy import delete
stmt = delete(user_table).where(user_table.c.name == "patrick")
print(stmt)

使用ORM进行数据操作

使用ORM工作单元模式插入行

使用ORM的时候 Session会先记录所有要执行的并在一起执行 这个过程称为一个工作单元

类的实例代表行

之前的实例使用字典传递给Insert以指示想要的数据，而在ORM中直接使用 声明式表配置 的Python类 。 类级别中， User和Address类充当定义相应数据库表外观的位置 也可充当要拓展的数据对象
以下两个User对象，每个都充当一个要insert的潜在数据库行

squidward = User(name="squidward", fullname="Squidward Tentacles")
krabs = User(name="ehkrabs", fullname="Eugene H. Krabs")

目前这两个对象都是处在瞬态，没有id

瞬态（Transient） 是描述对象生命周期状态的一个关键术语，它表示对象尚未与数据库建立任何关联，是一个“全新”且“临时”存在的实体

向Session添加对象

with Session(engine) as session:session.add(squidward)session.add(krabs)

向Session添加这两个User对象 目前这两个对象是处于一个挂起的状态 有id,但未插入
当有挂起对象的时候可以使用Session.new查看挂起对象

刷新

Session使用工作单元的模式 意味着它一次累积一个更改，但直到需要时才真正将其传达给数据库。 这使其能够根据给定的一组挂起更改，更好地决定如何在事务中发出 SQL DML(Data Manipulation Language)
但是如果需要推送以支持下一次推送可以条用Session.flush()手动刷新
刷新首先调用 Session 来发出 SQL，因此它创建了一个新事务并为这两个对象发出了适当的 INSERT 语句。 该事务现在保持打开状态

自动生成的主键属性

一旦插入行，所创建的两个User对象就处于持久的状态 有id，已经插入

从identity map中按主键获取对象

对象的主键标识对于Session非常重要

some_squidward = session.get(User, 4)
some_squidward

返回：User(id=4, name='squidward', fullname='Squidward Tentacles')

identity map在特定Session对象的范围内，为每个特定的数据库标识维护一个特定的 Python 对象的唯一实例
因此some_squidward和squidward是同一个对象

提交

session.commit()

当session关闭后 上面的User对象的属性都会过期

使用工作单元模式更新 ORM 对象

User对象sandy的主键为2
如果更改此对象的属性，Session会跟踪此更改
当 Session 下次发出刷新时，将发出一个 UPDATE，该 UPDATE 将更新数据库中的此值 nb

使用工作单元模式删除 ORM 对象

patrick = session.get(User, 3)
session.delete(patrick)

现在被删除的 patrick 对象实例不再被认为在 Session 中是持久的

回滚

在之前的示例中，将 sandy 对象的 .fullname 更改为 "Sandy Squirrel"，但如果想回滚此更改。调用 Session.rollback() 不仅会回滚事务，还会使当前与此 Session 关联的所有对象过期，这将导致它们在下次访问时使用称为延迟加载 的过程自行刷新

关闭Session

1. 释放所有连接资源到连接池
2. 从 Session 中驱逐所有对象