2023-09-06发表2023-09-29更新Notes

SQLAlchemy完全入门

之前写小网站应用练手的时候接触到了这个框架的用法。学习的时候发现各种博客参差不齐，大多只是搬了一套用法，没有特别能讲清楚的那种。官方文档很完善，但其中有些东西写得比较抽象，也有很多事项还是自己钻研过之后才理解的。在这里按自己的思路简单记录一下。

这些大部分都是咱自己的理解，大概会有一些不准确的地方。目前还没写完，以后一有时间了就会再添加一些。

SQLAlchemy是python中常用的ORM层框架。它的位置处于DBAPI和web应用之间，自身并不包含连接数据库的功能，需要配合数据库驱动(DBAPI)使用。

008_001

对于一般的python的DBAPI，进行数据库操作需要建立一个连接，再从连接中获取一个游标(cursor)，再用游标执行SQL语句并从游标中获取结果。整个流程很直观。

而对于一个SQLAlchemy实例，通常需要：

engine: 通过DBAPI获取数据库连接(实际上是一个连接池)。即让SQLAlchemy知道如何连接到我们的数据库。
Metadata: 保存数据库中schema信息的集合。
Table: 数据库的表的对象。可以自己定义，或者通过engine从数据库中已经存在的表中反射。当然同时也有Column作为列的对象。
Mapped Class: 映射类。把数据库表映射成类。
Session: 构建一个绑定到engine的session，是最终用来进行各种数据库操作的接口。

最终使用SQLAlchemy具体方法其实多种多样。新旧版本之间也有些区别，有些方法是旧版的常规使用方法但新版已经不推荐使用了(但还在兼容)，有些非常规的方法在简单的情况下也可以正常使用(比如直接用Table实例在Session中进行数据库操作而不进行类映射)。看别的博客和说明的时候会发现怎么用的都有，很容易就绕晕了。但其实提到的方法在目前版本下基本上都可以正常使用，但用不用得明白又是另外一回事了。

不管怎样，我们只要分别理解以上几个部分的用法和作用，即可理解这些使用方法分别是怎么回事。

0x01 安装

这个不用多说，直接用pip安装。

pip install SQLAlchemy

当然上面提到过需要和DBAPI一起使用，所以DBAPI当然也得安装。咱用mysql多一点，使用的是mysql官方的mysql-connector。并不是SQLAlchemy推荐的mysql驱动，但咱用着好像也没什么问题

pip install mysql-connector-python

0x02 连接数据库

在SQLAlchemy中建立数据库连接需要用engine。首先安装数据库对应的DBAPI，然后用create_engine()函数创建一个连接engine。不同于python驱动，SQLAlchemy中engine连接参数的指定使用的是类似于JDBC的一种Database URL。总结一下就像这样：

<dialect>+<DBAPI>://<user>:<password>@<ip>:<port>/<schema>?<arg_key>=<value>&<arg_key>=<value>..

不同的数据库的差别被称为dialect(方言)。SQLAlchemy即是通过区分不同的dialect和DBAPI来同时兼容众多不同的数据库的。因此DATABASE URL最开头首先需要指定数据库dialect和使用的DBAPI。//后的内容会解析成python的DBAPI所识别的kwargs参数，原样传给DBAPI中建立连接池的方法，获取的当然也是一个连接池(并非单个连接)。DBAPI所支持的其他的参数，包括SSL连接相关的参数直接作为URL的参数加在?后即可。

顺便Database URL也是支持URL Encode的，特殊字符像网站URL一样escape一下就可以正常传了。

from sqlalchemy import create_engine

engine=create_engine("mysql+mysqlconnector://user:password@octanepi:3306/test?ssl_verify_cert=True&ssl_cert=client.crt&ssl_key=client_key.pem&ssl_ca=ca-chain.crt&pool_size=10")

如此就能得到一个使用mysql数据库和mysql-connector-python的DBAPI通过SSL连接到octanepi的mysql数据库服务的一个名为”test”的schema的engine实例，对应的就是DBAPI中的mysql.pooling.MySQLConnectionPool()方法。之后对数据库的操作都建立在这个engine实例的基础上即可。3306端口号可以省略。pool_size是DBAPI中的方法中指定连接池的连接数的参数，当然也可以直接在Database URL中指定，影响的就是engine对应的连接池大小。

0x03 Raw SQL

除了ORM查询方法外，SQLAlchemy也预留了直接使用SQL语句的接口。有了数据库的连接，我们就可以用raw sql方法直接进行操作了。其实就是相当于隔了一层SQLAlchemy core让DBAPI运行我们自己构建的SQL语句，没有涉及到SQLAlchemy的ORM功能，也就当然不用进行表和类的声明了。

from sqlalchemy.sql import text

conn = engine.connect()
print(conn.execute(text("SELECT * FROM information LIMIT 1")).fetchall())
print(conn.execute(text("SELECT * FROM information WHERE id=:id LIMIT 1"), {"id":1}).fetchall()) #参数化查询
conn.close()

注意我们自己写的SQL语句字符串需要先传入text()，但如果是SQLAlchemy Expression等生成的SQL语句就不用了。

这种方法乍一看没什么用，但这个方式还是可以利用到SQLAlchemy统一管理数据库连接的功能(SQLAlchemy自动使用连接池)。用这种方式在SQLAlchemy框架下直接从现有的sql语句中插入数据库初始数据是一个很好的选择。

0x04 构建表对象(Table)

在使用ORM之前，我们首先需要初始化一个Table对象(咱先叫它表对象吧orz)。可以从数据库中已经存在的表中加载，也可以通过定义一个python类而构建一个新的表。

如果我们需要的表是要新建的，则我们可以选择直接用声明式映射新建一个映射类而直接跳过此节构建表对象的过程。具体内容在0x05节的##声明式映射.

当然也可以选择先按照此节内容，构建一个新的表对象再用下一节的命令式/混合式映射从该表对象构建映射类。

metadata

在SQLAlchemy中metadata相当于表对象的一个容器和目录。表对象都要附加到metadata实例上。对于同一个schema的表，应当使用同一个metadata实例。

反射表

如果我们需要用到的表是数据库中存在的表，那么我们可以直接从该表直接创建一个SQLAlchemy中的表对象。这个过程叫做Reflect(反射)

表对象是需要基于Metadata的，所以首先我们需要构建一个MetaData实例。然后利用这个metadata实例借助engine从数据库中指定的一个存在的表中获取一个Table实例即可。

from sqlalchemy import Table, MetaData

metadata = MetaData()
metadata.reflect(bind=engine)
info_table = Table("information", metadata, autoload_with=engine)
status_table = Table("status", metadata, autoload_with=engine)

定义表

如果我们需要的数据库表在原数据库中不存在，而我们需要新建一个，则可以选择通过继承Table类作为表。表中的列通过在Table中传入Column实例即可定义。同时也要注意表对象是需要基于Metadata的，因此我们也需要一个metadata实例并将这个表对象注册到其中，随后这个表就可以由metadata实例借助engine在原数据库中新建了。

from sqlalchemy import Table, MetaData, Column, Integer, String, ForeignKey

metadata = MetaData()
info_table = Table(
  "information",
  metadata,
  Column("id", Integer, primary_key=True), # 使用Integer类构建int数据类型
  Column("Name", String(45)), # 使用String类构建VARCGHAR数据类型，需要指定大小
  Column("Gender", String(45))
)

# 在原数据库中新建表(如某张表已经存在则不会新建)
metadata.create_all(bind=engine)

使用表构建SQL语句

只构建一个表对象并没有完全做完从关系数据库映射到类的操作，但此时数据库表已经可以用表对象来表示了，因此我们也可以通过SQLAlchemy2.0下的方法用表对象构建SQL语句并执行。

column对象在Table对象column(可以简写为c)属性下。用select()和insert()等函数对表对象进行操作即可构建相应的SQL语句，随后可以被执行：

from sqlalchemy import select
print(select(info.c.id, info.c.Name).filter(info.c.id==12))
# SELECT information.id, information."Name" 
# FROM information
# WHERE information.id = :id_1
# 虽然这里是以:id_1参数的形式print出来，但如果直接执行则:id_1会作为12执行

通过这种方法就可以生成的SQL语句了，已经足够实现大部分的操作，通常应该使用Session执行。(详见0x06节的##通过ORM方法构造SQL语句并使用Session执行)

0x05 构建映射类(Mapped Class)

创建一个映射类的基本方法主要有两种style，一种是Declarative Mapping(声明式映射)，另一种是Imperative Mapping(命令式映射)(大概是这么翻译?…)。

声明式映射(Declarative Mapping)

声明式映射通过继承DeclarativeBase直接声明一个映射类，不需要单独构建表对象(Table)，也就不需要0x04中的内容和操作。需要注意这种方法虽然没有单独构建一个表对象，但其对应的表对象还是在声明映射类的同时被创建了，并存在于.__table__下。

from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase): pass

class Info(Base):
    __tablename__ = "information"
    id = mapped_column(Integer, primary_key=True)
    Name = mapped_column(String(45))
    Gender = mapped_column(String(45))

# 在原数据库中新建映射类对应的表
Base.metadata.create_all(bind=engine)

在从DeclarativeBase继承来的Base下默认包含一个MetaData实例Base.metadata。而继承Base得来的映射类所对应的表对象也会自动注册在该metadata下，并可以由该metadata在原数据库中新建该表。因此在创建多个映射类时应当尽量继承同一个Base，以保持metadata一致。

命令/混合式映射

命令/混合式映射都是先构建一个表对象(Table)，再通过命令式/声明式方法将已经构建的表对象再转化成映射类。虽然多了一步，但区分了表对象和映射类，也更有些额外的操作空间。而且想要利用从数据库中存在的表，即反射表，就需要使用这两种方法。但这两种映射产生的映射类的属性不一定在类中显式声明过(但可以手动声明)，在调用属性时会有一些不便之处。

所谓通过声明式映射的方法从表对象映射类其实更像混合式映射的定义，所以咱把它归到混合式映射中。

命令式(Imperative Mapping)

命令式映射用map_imperatively()方法直接将表对象映射到一个新类中作为一个映射类。

from sqlalchemy.orm import registry

# 默认0x03节的操作得到一个名为"info_table"的表对象
class Info: pass

registry().map_imperatively(Info, info_table)

混合式(Hybrid Declarative Mapping)

这种是使用声明式映射的方法，但在声明映射类时不在映射类中声明定义列，而通过直接定义__table__来映射一个已经存在的表对象。

from sqlalchemy.orm import DeclarativeBaseNoMeta

# 默认0x03节的操作得到一个名为"info_table"的表对象
class Base(DeclarativeBaseNoMeta): pass

class info(Base):
    __table__ = info_table

这里需要注意，已有的表对象是属于之前我们自己声明的MetaData实例的，不需要基类再默认包含一个MetaData属性了，因此在这里使用的是默认不自带MetaData的DeclarativeBaseNoMeta而非声明式映射使用的是DeclarativeBase

多表映射(Multiple Tables Mapping)

多表映射(大概能这么翻译?…)就是把多个表对象用join()以混合式映射的方式映射到同一个映射类中，从而可以让一个映射类根据表之间指定的关系来同时联系多个表，在某些情况下用起来很合适(内建join)。当然使用的前提是在表对象中已经存在了ForeignKey的关系。

from sqlalchemy import join
from sqlalchemy.orm import DeclarativeBaseNoMeta

# 默认0x03节的操作得到两个分别名为"table1"和"table2"的表对象
class Base(DeclarativeBaseNoMeta): pass
joined_table = join(table1, table2, isouter=True)

class joined_class(Base):
    __table__=joined_table

注意join方法是有顺序的，传入两个Table对象则前者为left后者为right。默认方式为inner join。设置isouter和full可以指定outer/full方式。

如果映射的表中有相同的列名，那么混合映射生成的属性会产生冲突。遇到这种情况手动指定映射关系即可。

from sqlalchemy import join
from sqlalchemy.orm import DeclarativeBaseNoMeta, column_property

# student_info: ID, Name, Gender
# student_status: StudentID, Name, Status
class Base(DeclarativeBaseNoMeta): pass
joined_table = join(student_info, student_status, isouter=True)

class joined_class(Base):
    __table__=joined_table
    StatusName=column_property(student_status.c.Name)

以上示例中student_info和student_status都有一个相同的名为Name的列，如果将两个表对象映射到同一个映射类，Name属性会产生冲突。因此可以在混合映射时通过单独声明来将原student_info的Name列映射为映射类中的StatusName属性。同时，student_info的Name列在映射类中依旧保持为Name属性。

0x06 用Session进行数据库操作

在实际Flask的应用中，我们通常使用scoped_session()。它会为每个调用它的线程分配一个单独的session，从而实现线程安全。

from sqlalchemy.orm import scoped_session, sessionmaker

db_session=scoped_session(sessionmaker(bind=engine, autocommit=False, autoflush=False))

使用session的方法有很多。

使用query_property进行查询

如果已经有了一个映射类，则可以直接给映射类附加一个query_property()从而让它具有查询的功能。

# 默认0x05节的操作得到一个名为"Info"的映射类
Info.query = db_session.query_property()

也可以在构建映射类的同时就把query_property附加到映射类里。总之，让query_property成为映射类的一个子属性即可。

该query_property也属于scoped_session，因此也是内建线程安全的。

这种方式把session的query_property和映射类封装在了一起，可以很方便地进行查、删、改。

result = Info.query.filter(Info.Gender=="female").first()
print(result.id, result.Name)

如此得到的查询结果以映射类的实例存在。如果需要获取全部结果，可以用.all()来获取一个包含所有查询结果映射类实例的列表。如果只需要获取其中一个，可以用.first()方法直接得到一个单独的映射类实例。我们可以通过调用映射类实例包含的列对象来读取结果中各列的值。同时我们也可以在查询对象中附加.with_entities()方法并在其中指定列对象，使查询结果转化为包含对应列对象值的元组。

print(Info.query.with_entities(Info.id, Info.Name, Info.Gender).filter(Info.Gender=="female").all())
# [(2, 'Alice', 'female')]

删和改的操作：


Info.query.filter(Info.id == 2).update({Info.Gender: "male"}) # update
print(Info.query.with_entities(Info.id, Info.Name, Info.Gender).filter(Info.id == 20).all())
# [(2, 'Alice', 'male')]
Info.query.filter(Info.id == 2).delete() # delete
print(Info.query.with_entities(Info.id, Info.Name).filter(Info.id == 20).all())
# []
db_session.commit()

调用Session.query进行查询

这个部分和query_property()类似，只是映射类不用包含session的query_property属性，对Info.query.的调用变成了对session.query(Info).的调用。

print(db_session.query(Info).filter(...).all())

需要注意的是以上调用query()的方法是SQLAlchemy 1.x中的常规方法，并不是较新的SQLAlchemy 2.0中推荐的方法了。但在2.x版本中这两种方法依然可用，并且不会被移除。

通过映射类新增数据

想要向一个表中新增数据，只需要构建一个创建一个相应映射类的实例，将该实例添加到session提交即可。

row1 = Info(id=3,Name="Bob",Gender="male")
row2 = Info(id=4,Name="Eve",Gender="female")
db_session.add(row1)
db_session.add(row2)
db_session.commit()

通过ORM方法构造SQL语句并使用Session执行

此方法是2.0推荐使用的方法。通过用select()、insert()、update()等方法构造映射类的SQL语句，再传入Session执行并返回结果，让增删查改的操作更统一，使用方法也更贴近于dbapi了。

构造映射类的SQL语句对象：

from sqlalchemy import select, delete, update

stmt1 = select(Info).where(Info.id == 1) # select
stmt2 = delete(Info).where(Info.id == 2) # delete
stmt3 = update(Info).where(Info.ID == 3).values(Name="Eve") # update

如此我们就可以构造sqlalchemy中的sql语句对象。将该对象传入session.execute()或者session.scalars()方法中即可执行并返回结果

result1 = db_session.scalars(stmt1).first()
db_session.execute(stmt2)
db_session.execute(stmt3)
dn_session.commit()
print(result1.Name)

向表中新增数据的一种方法：

from sqlalchemy import insert

row1 = {"id": 3, "Name": "Bob", "Gender": "male"}
row2 = {"id": 4, "Name": "Eve", "Gender": "female"}
db_session.execute(insert(Info), [row1, row2])
db_session.commit()

当然可以用上一节##通过映射类新增数据的构造对象实例的方法向表中新增数据

获取查询结果

使用session进行查询获取的结果有两种方式，使用scalar方法返回的是映射类的实例，而非scalar的方法返回的是row的实例.

0x07 事务

事务在数据库的操作中是一个重要的概念。在SQLAlchemy中，事务是通过Session管理的。调用Session.commit()和Session.rollback()就可以简单地进行事务的提交或者回滚。

try:
    db_session.add(row1)
    db_session.delete(row2)
    db_session.execute(...)
except: db_session.rollback()
else: db_session.commit()
finally: db_session.remove()

事务也可以通过python的上下文管理器来更简洁且直观地进行管理，更推荐：

with scoped_session(sessionmaker(bind=engine, autocommit=false)) as db_session, db_session.begin():
    db_session.add(row1)
    db_session.delete(row2)
    db_session.execute(...)

如果上下文管理器内的数据库操作没有问题，则上下文管理器就会自动commit并关闭session。反之则会自动回滚。

0x08 在flask框架下的应用

SQLAlchemy可以与flask框架配合使用。由于scoped_session()是内建线程安全的，并且SQLAlchemy默认使用的是连接池。因此只要将一个scoped_session()构建的session实例加入到flask应用即可。下面是咱写的一种简单的使用方法，用起来没什么大问题：

# database.py
from sqlalchemy import create_engine, Table, MetaData
from sqlalchemy.orm import scoped_session, sessionmaker, DeclarativeBaseNoMeta

class orm_connector:
    def __init__(self, database_uri:str,) -> None:
        self.engine = create_engine(database_uri)
        self.metadata = MetaData()
        self.metadata.reflect(bind=self.engine)

    def get_class(self, table_name:str): # 根据指定的表名从数据库中反射表为映射类，使用混合式映射
        class Base(DeclarativeBaseNoMeta): pass
        class mapped_class(Base):
            __table__ = Table(table_name, self.metadata, autoload_with=self.engine)
        return mapped_class

    def get_session(self):
        return scoped_session(sessionmaker(bind=self.engine, autocommit=False, autoflush=False))

# __init__.py
from flask import Flask, request
from .database import orm_connector

def create_app():
    app = Flask(__name__, instance_relative_config=True)
    app.config.from_pyfile("config.cfg") # 加载instance下的配置文件

    orm = orm_connector(app.config["SQLALCHEMY_DATABASE_URL"])
    db_session = orm.get_session() # 获取session
    Info = orm.get_class("information") # 对应表的获取映射类

    @app.route("/")
    def index():
        db_session.execute(...)
        ...
        db_session.commit()
        ...
    ...

    @app.teardown_appcontext
    def shutdown_session(exception=None):
        db_session.remove() # 回收session

注意teardown_appcontext下的方法中是必要的。该方法在请求结束时触发，会关闭session并把连接归还到连接池中。

#Python