SqlAlchemy 2.0 中文文档（五十二）（4）

SqlAlchemy 2.0 中文文档（五十二）（3）https://developer.aliyun.com/article/1563142

PyODBC

通过 PyODBC 驱动程序支持 Microsoft SQL Server 数据库。

DBAPI

PyODBC 的文档和下载信息（如果适用）可在以下网址获取：pypi.org/project/pyodbc/

连接

连接字符串：

mssql+pyodbc://<username>:<password>@<dsnname>

连接到 PyODBC

此处的 URL 将被翻译为 PyODBC 连接字符串，详见ConnectionStrings。

DSN 连接

ODBC 中的 DSN 连接意味着在客户端机器上配置了预先存在的 ODBC  数据源。然后，应用程序指定此数据源的名称，其中包括诸如正在使用的特定 ODBC  驱动程序以及数据库的网络地址等细节。假设在客户端上配置了数据源，则基本的基于 DSN 的连接如下所示：

engine = create_engine("mssql+pyodbc://scott:tiger@some_dsn")

将上述内容传递给 PyODBC 的连接字符串如下：

DSN=some_dsn;UID=scott;PWD=tiger

如果省略了用户名和密码，则 DSN 表单还将向 ODBC 字符串添加Trusted_Connection=yes指令。

主机名连接

PyODBC 也支持基于主机名的连接。这通常比 DSN 更容易使用，并且具有另一个优势，即可以在 URL 中本地指定要连接到的特定数据库名称，而不是将其固定为数据源配置的一部分。

在使用主机名连接时，还必须在 URL 的查询参数中指定驱动程序名称。由于这些名称通常包含空格，因此必须对名称进行 URL 编码，这意味着使用加号代替空格：

engine = create_engine("mssql+pyodbc://scott:tiger@myhost:port/databasename?driver=ODBC+Driver+17+for+SQL+Server")

driver关键字对于 pyodbc 方言非常重要，必须以小写形式指定。

查询字符串中传递的任何其他名称都将通过 pyodbc 连接字符串传递，例如 authentication、TrustServerCertificate 等。 多个关键字参数必须用与号（&）分隔；这些参数在生成 pyodbc 连接字符串时将被转换为分号：

e = create_engine(
    "mssql+pyodbc://scott:tiger@mssql2017:1433/test?"
    "driver=ODBC+Driver+18+for+SQL+Server&TrustServerCertificate=yes"
    "&authentication=ActiveDirectoryIntegrated"
)

可以使用 URL 构造相等的 URL：

from sqlalchemy.engine import URL
connection_url = URL.create(
    "mssql+pyodbc",
    username="scott",
    password="tiger",
    host="mssql2017",
    port=1433,
    database="test",
    query={
        "driver": "ODBC Driver 18 for SQL Server",
        "TrustServerCertificate": "yes",
        "authentication": "ActiveDirectoryIntegrated",
    },
)

通过确切的 Pyodbc 字符串传递

PyODBC 连接字符串也可以直接以 pyodbc 的格式发送，如PyODBC 文档中所述，使用参数 odbc_connect。 URL 对象可以帮助简化此过程：

from sqlalchemy.engine import URL
connection_string = "DRIVER={SQL Server Native Client 10.0};SERVER=dagger;DATABASE=test;UID=user;PWD=password"
connection_url = URL.create("mssql+pyodbc", query={"odbc_connect": connection_string})
engine = create_engine(connection_url)

使用访问令牌连接到数据库

一些数据库服务器仅允许使用访问令牌进行登录。 例如，SQL Server 允许使用 Azure Active Directory 令牌连接到数据库。 这需要使用 azure-identity 库创建凭据对象。 关于身份验证步骤的更多信息可以在 Microsoft 文档中找到。

获得引擎后，每次请求连接都需要将凭据发送到 pyodbc.connect。 一种方法是在引擎上设置事件侦听器，该事件侦听器将凭据令牌添加到方言的连接调用中。 关于这一点的更多讨论可以在 生成动态身份验证令牌中找到。 尤其对于 SQL Server，这将作为由 Microsoft 描述的 ODBC 连接属性传递的数据结构。

下面的代码片段将创建一个引擎，该引擎使用 Azure 凭据连接到 Azure SQL 数据库：

import struct
from sqlalchemy import create_engine, event
from sqlalchemy.engine.url import URL
from azure import identity
SQL_COPT_SS_ACCESS_TOKEN = 1256  # Connection option for access tokens, as defined in msodbcsql.h
TOKEN_URL = "https://database.windows.net/"  # The token URL for any Azure SQL database
connection_string = "mssql+pyodbc://@my-server.database.windows.net/myDb?driver=ODBC+Driver+17+for+SQL+Server"
engine = create_engine(connection_string)
azure_credentials = identity.DefaultAzureCredential()
@event.listens_for(engine, "do_connect")
def provide_token(dialect, conn_rec, cargs, cparams):
    # remove the "Trusted_Connection" parameter that SQLAlchemy adds
    cargs[0] = cargs[0].replace(";Trusted_Connection=Yes", "")
    # create token credential
    raw_token = azure_credentials.get_token(TOKEN_URL).token.encode("utf-16-le")
    token_struct = struct.pack(f"<I{len(raw_token)}s", len(raw_token), raw_token)
    # apply it to keyword arguments
    cparams["attrs_before"] = {SQL_COPT_SS_ACCESS_TOKEN: token_struct}

提示

当没有用户名或密码时，SQLAlchemy pyodbc 方言当前会添加 Trusted_Connection 令牌。 根据 Microsoft 的用于 Azure 访问令牌的文档，当使用访问令牌时，连接字符串不得包含 UID、PWD、Authentication 或 Trusted_Connection 参数，需要将其删除。 #### 在 Azure Synapse Analytics 上避免事务相关的异常

Azure Synapse Analytics 在事务处理方面与普通 SQL Server 有显着差异；在某些情况下，Synapse 事务中的错误可能导致服务器端任意终止，从而导致 DBAPI 的 .rollback() 方法 (以及 .commit()) 失败。该问题阻止了允许 .rollback() 在没有事务存在时静默通过的常规 DBAPI 合同，因为驱动程序不期望出现此条件。此故障的症状是，在某些操作失败后尝试发出 .rollback() 后，异常消息类似于‘No corresponding transaction found. (111214)’。

可以通过向 SQL Server 方言传递 ignore_no_transaction_on_rollback=True 参数来处理此特定情况，方法是通过create_engine() 函数如下所示：

engine = create_engine(connection_url, ignore_no_transaction_on_rollback=True)

使用上述参数，方言将捕获在 connection.rollback() 期间引发的 ProgrammingError 异常，并在错误消息中包含代码 111214 时发出警告，但不会引发异常。

版本 1.4.40 中的新功能：添加了 ignore_no_transaction_on_rollback=True 参数。

为 Azure SQL 数据仓库 (DW) 连接启用自动提交

Azure SQL 数据仓库不支持事务，这可能会导致 SQLAlchemy 的“autobegin”（以及隐式提交/回滚）行为出现问题。我们可以通过在 pyodbc 和 engine 级别启用自动提交来避免这些问题：

connection_url = sa.engine.URL.create(
    "mssql+pyodbc",
    username="scott",
    password="tiger",
    host="dw.azure.example.com",
    database="mydb",
    query={
        "driver": "ODBC Driver 17 for SQL Server",
        "autocommit": "True",
    },
)
engine = create_engine(connection_url).execution_options(
    isolation_level="AUTOCOMMIT"
)

避免将大字符串参数发送为 TEXT/NTEXT

出于历史原因，默认情况下，Microsoft 的 SQL Server ODBC 驱动程序将长字符串参数（大于 4000 个 SBCS  字符或 2000 个 Unicode 字符）发送为 TEXT/NTEXT 值。多年来，TEXT 和 NTEXT 已经被弃用，并且开始在新版本的  SQL_Server/Azure 中引起兼容性问题。例如，参见此问题。

从 ODBC 驱动程序 18 开始，我们可以通过 LongAsMax=Yes 连接字符串参数覆盖传统行为，并将长字符串作为 varchar(max)/nvarchar(max) 传递：

connection_url = sa.engine.URL.create(
    "mssql+pyodbc",
    username="scott",
    password="tiger",
    host="mssqlserver.example.com",
    database="mydb",
    query={
        "driver": "ODBC Driver 18 for SQL Server",
        "LongAsMax": "Yes",
    },
)

Pyodbc 连接池 / 连接关闭行为

PyODBC 默认使用内部连接池，这意味着连接的生命周期比在 SQLAlchemy 本身中更长。由于 SQLAlchemy 有自己的连接池行为，通常最好禁用此行为。此行为只能在创建任何连接之前在 PyODBC 模块级别全局禁用：

import pyodbc
pyodbc.pooling = False
# don't use the engine before pooling is set to False
engine = create_engine("mssql+pyodbc://user:pass@dsn")

如果将此变量保留在默认值 True，应用程序将继续保持活动数据库连接，即使 SQLAlchemy 引擎本身完全丢弃连接或引擎被处理掉。

另请参阅

连接池 - 在 PyODBC 文档中。

驱动程序 / Unicode 支持

PyODBC 最适合与微软 ODBC 驱动程序一起使用，特别是在 Python 2 和 Python 3 上都支持 Unicode 的领域。

不建议在 Linux 或 OSX 上使用 FreeTDS ODBC 驱动程序与 PyODBC 一起使用；在这个领域，包括在微软为  Linux 和 OSX 提供 ODBC 驱动程序之前，历史上存在许多与 Unicode 相关的问题。现在微软为所有平台提供驱动程序，对于  PyODBC 支持，建议使用这些驱动程序。FreeTDS 仍然适用于非 ODBC 驱动程序，例如 pymssql，在那里它的工作非常出色。

行计数支持

至于 Pyodbc 与 SQLAlchemy ORM 的“版本化行”功能之前的限制，在 SQLAlchemy 2.0.5 版中已经解决。请参阅 Rowcount Support / ORM Versioning 中的说明。

快速执行多次模式

PyODBC 驱动程序包括对执行 DBAPI executemany() 调用时大大减少往返次数的“快速执行多次”模式的支持，当使用微软 ODBC 驱动程序时，对于内存中适合的有限大小批次。通过在 DBAPI 游标上设置属性 .fast_executemany 来启用此功能，当要使用 executemany 调用时。SQLAlchemy PyODBC SQL Server 方言通过将 fast_executemany 参数传递给 create_engine() 来支持此参数，仅当使用微软 ODBC 驱动程序时：

engine = create_engine(
    "mssql+pyodbc://scott:tiger@mssql2017:1433/test?driver=ODBC+Driver+17+for+SQL+Server",
    fast_executemany=True)

从版本 2.0.9 开始更改：- fast_executemany 参数现在具有其预期的效果，这使得 PyODBC 功能在执行具有多个参数集的所有 INSERT 语句时生效，不包括 RETURNING。之前，SQLAlchemy 2.0 的 insertmanyvalues 功能通常会导致即使指定了，也大多数情况下不使用fast_executemany。

1.3 版中的新功能。

另请参阅

快速执行多次 - 在 github ### 设置输入大小支持

从版本 2.0 开始，pyodbc cursor.setinputsizes() 方法用于所有语句执行，除了当 fast_executemany=True 时，不支持cursor.executemany()调用（假设 insertmanyvalues 已启用，“fastexecutemany”不管怎样都不会对 INSERT 语句产生影响）。

通过将 use_setinputsizes=False 传递给 create_engine() 可以禁用cursor.setinputsizes()的使用。

当 use_setinputsizes 保持默认值 True 时，可以通过使用 DialectEvents.do_setinputsizes() 钩子来自定义传递给 cursor.setinputsizes() 的每种类型符号。请参阅该方法以获取用法示例。

从 2.0 版本开始更改：mssql+pyodbc 方言现在默认为在所有语句执行中使用use_setinputsizes=True，但 fast_executemany=True 时除外，快速执行多次cursor.executemany() 调用。该行为可以通过将 use_setinputsizes=False 传递给 create_engine() 来关闭。 ## pymssql

通过 pymssql 驱动程序支持 Microsoft SQL Server 数据库。

连接

连接字符串：

mssql+pymssql://<username>:<password>@<freetds_name>/?charset=utf8

pymssql 是一个提供围绕 FreeTDS 的 Python DBAPI 接口的 Python 模块。

从 2.0.5 版本开始更改：pymssql 已恢复到 SQLAlchemy 的持续集成测试 ## aioodbc

通过 aioodbc 驱动程序支持 Microsoft SQL Server 数据库。

DBAPI

aioodbc 的文档和下载信息（如果适用）可在此处获取：pypi.org/project/aioodbc/

连接

连接字符串：

mssql+aioodbc://<username>:<password>@<dsnname>

以 asyncio 样式支持 SQL Server 数据库，使用 aioodbc 驱动程序，它本身是 pyodbc 的线程包装器。

从 2.0.23 版本开始新增：添加了在 pyodbc 和通用 aio* 方言架构之上构建的 mssql+aioodbc 方言。

使用特殊的 asyncio 中介层，aioodbc 方言可用作 SQLAlchemy asyncio 扩展包的后端。

该驱动程序的大多数行为和注意事项与在 SQL Server 上使用的 pyodbc 方言相同；有关一般背景，请参阅 PyODBC。

该方言通常仅应与 create_async_engine() 引擎创建函数一起使用；否则，连接样式与在 pyodbc 部分文档中记录的相同：

from sqlalchemy.ext.asyncio import create_async_engine
engine = create_async_engine(
    "mssql+aioodbc://scott:tiger@mssql2017:1433/test?"
    "driver=ODBC+Driver+18+for+SQL+Server&TrustServerCertificate=yes"
)

支持 Microsoft SQL Server 数据库。

以下表格总结了数据库发布版本的当前支持级别。

支持的 Microsoft SQL Server 版本

DBAPI 支持

提供以下方言/DBAPI 选项。请参阅各个 DBAPI 部分以获取连接信息。

• PyODBC
  
• pymssql
  
• aioodbc
  

外部方言

除了具有本地 SQLAlchemy 支持的上述 DBAPI 层之外，还有用于其他与 SQL Server 兼容的 DBAPI 层的第三方方言。请参阅 方言 页面上的“外部方言”列表。

自动递增行为 / IDENTITY 列

SQL Server 使用IDENTITY构造提供所谓的“自动增量”行为，该构造可以放置在表中的任何单个整数列上。SQLAlchemy 将IDENTITY考虑在其整数主键列的默认“autoincrement”行为中，该行为在Column.autoincrement中描述。这意味着默认情况下，Table中的第一个整数主键列将被视为标识列 - 除非它与Sequence相关联 - 并且将生成 DDL 如下：

from sqlalchemy import Table, MetaData, Column, Integer
m = MetaData()
t = Table('t', m,
        Column('id', Integer, primary_key=True),
        Column('x', Integer))
m.create_all(engine)

上述示例将生成 DDL 如下：

CREATE  TABLE  t  (
  id  INTEGER  NOT  NULL  IDENTITY,
  x  INTEGER  NULL,
  PRIMARY  KEY  (id)
)

对于不希望使用此默认生成的IDENTITY的情况，在第一个整数主键列上指定Column.autoincrement标志为False：

m = MetaData()
t = Table('t', m,
        Column('id', Integer, primary_key=True, autoincrement=False),
        Column('x', Integer))
m.create_all(engine)

要将IDENTITY关键字添加到非主键列，请在所需的Column对象上指定Column.autoincrement标志为True，并确保在任何整数主键列上将Column.autoincrement设置为False：

m = MetaData()
t = Table('t', m,
        Column('id', Integer, primary_key=True, autoincrement=False),
        Column('x', Integer, autoincrement=True))
m.create_all(engine)

自版本 1.4 更改：在Column中添加了Identity构造，用于指定IDENTITY的起始值和增量参数。这些参数取代了使用Sequence对象来指定这些值。

自版本 1.4 弃用：Column的mssql_identity_start和mssql_identity_increment参数已弃用，应该用Identity对象替换。指定两种配置IDENTITY的方式将导致编译错误。这些选项也不再作为Inspector.get_columns()中dialect_options键的一部分返回。请改为使用identity键中的信息。

自版本 1.3 起弃用：使用Sequence指定 IDENTITY 特性已被弃用，并将在将来的版本中删除。请使用Identity对象参数Identity.start和Identity.increment。

从版本 1.4 开始更改：移除了使用Sequence对象修改 IDENTITY 特性的能力。现在，Sequence对象仅操作真正的 T-SQL SEQUENCE 类型。

注意

表上只能有一个 IDENTITY 列。当使用autoincrement=True启用 IDENTITY 关键字时，SQLAlchemy 不会阻止多个列同时指定该选项。相反，SQL Server 数据库将拒绝CREATE TABLE语句。

注意

尝试为标记为 IDENTITY 的列提供值的 INSERT 语句将被 SQL Server 拒绝。为了接受该值，必须启用会话级选项“SET IDENTITY_INSERT”。当使用核心Insert构造时，SQLAlchemy  SQL Server 方言将在执行指定 IDENTITY 列的值时自动执行此操作；如果执行为 IDENTITY  列指定了一个值，则“IDENTITY_INSERT”选项将在该语句调用的范围内启用。然而，这种情况的性能不高，不应该依赖于常规使用。如果表实际上不需要  IDENTITY 行为在其整数主键列中，创建表时应禁用该关键字，方法是确保autoincrement=False被设置。

控制“开始”和“增量”

通过将参数Identity.start和Identity.increment传递给Identity对象提供了对“开始”和“增量”值的特定控制：

from sqlalchemy import Table, Integer, Column, Identity
test = Table(
    'test', metadata,
    Column(
        'id',
        Integer,
        primary_key=True,
        Identity(start=100, increment=10)
    ),
    Column('name', String(20))
)

上述Table对象的 CREATE TABLE 将是：

CREATE  TABLE  test  (
  id  INTEGER  NOT  NULL  IDENTITY(100,10)  PRIMARY  KEY,
  name  VARCHAR(20)  NULL,
  )

注意

Identity对象支持许多其他参数，除了start和increment之外。这些参数不受 SQL Server 支持，在生成 CREATE TABLE ddl 时将被忽略。

版本 1.3.19 中的更改：在 SQL Server 下，现在使用 Identity 对象来影响 Column 的 IDENTITY 生成器。之前使用的是 Sequence 对象。由于 SQL Server 现在支持将实际序列作为一个独立的构造，因此 Sequence 将从 SQLAlchemy 版本 1.4 开始以正常方式运作。

使用非整数数值类型的 IDENTITY

SQL Server 还允许将 IDENTITY 用于 NUMERIC 列。为了在 SQLAlchemy 中平滑实现这种模式，在列的主要数据类型应保持为 Integer，但是可以使用 TypeEngine.with_variant() 来指定部署到 SQL Server 数据库的底层实现类型为 Numeric：

from sqlalchemy import Column
from sqlalchemy import Integer
from sqlalchemy import Numeric
from sqlalchemy import String
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class TestTable(Base):
    __tablename__ = "test"
    id = Column(
        Integer().with_variant(Numeric(10, 0), "mssql"),
        primary_key=True,
        autoincrement=True,
    )
    name = Column(String)

在上面的示例中，Integer().with_variant() 提供了明确的使用信息，准确描述了代码的意图。将 autoincrement 仅适用于 Integer 的一般限制建立在元数据级别而不是每个方言级别。

使用上述模式时，从插入行返回的主键标识符（也是将分配给类似于上面的 TestTable 的 ORM 对象的值）将是 Decimal() 的实例，而不是使用 SQL Server 时的 int。通过将 False 传递给 Numeric.asdecimal，可以将 Numeric 类型的数值返回类型更改为浮点数。要将上述 Numeric(10, 0) 的返回类型规范化为返回 Python 整数（在 Python 3 中也支持“长”整数值），请使用 TypeDecorator 如下所示：

from sqlalchemy import TypeDecorator
class NumericAsInteger(TypeDecorator):
  '''normalize floating point return values into ints'''
    impl = Numeric(10, 0, asdecimal=False)
    cache_ok = True
    def process_result_value(self, value, dialect):
        if value is not None:
            value = int(value)
        return value
class TestTable(Base):
    __tablename__ = "test"
    id = Column(
        Integer().with_variant(NumericAsInteger, "mssql"),
        primary_key=True,
        autoincrement=True,
    )
    name = Column(String)

插入行为

在 INSERT 时处理 IDENTITY 列涉及两个关键技术。最常见的是能够获取给定 IDENTITY 列的“最后插入值”，SQLAlchemy 在许多情况下都会隐式执行这个过程，最重要的是在 ORM 中。

获取此值的过程有几种变体：

• 在绝大多数情况下，在 SQL Server 上与 INSERT 语句一起使用 RETURNING 以获取新生成的主键值：

INSERT  INTO  t  (x)  OUTPUT  inserted.id  VALUES  (?)

• 从 SQLAlchemy 2.0 开始，默认还使用 INSERT 语句的“插入多个值”行为功能来优化多行 INSERT 语句；对于 SQL Server，该功能适用于 RETURNING 和非 RETURNING INSERT 语句。
  从版本  2.0.10 开始更改：由于行排序问题，SQLAlchemy 版本 2.0.9 暂时禁用了 SQL Server 的 INSERT  语句的“插入多个值”行为功能。从 2.0.10 开始，该功能已重新启用，并对工作单元对 RETURNING 的排序要求进行了特殊处理。
• 当RETURNING不可用或通过implicit_returning=False禁用时，将使用scope_identity()函数或@@identity变量；后端的行为各不相同：

• 使用 PyODBC 时，短语; select scope_identity()将被附加到插入语句的末尾；为了接收值，将获取第二个结果集。给定一个表如下：

t = Table(
    't',
    metadata,
    Column('id', Integer, primary_key=True),
    Column('x', Integer),
    implicit_returning=False
)

• 插入操作看起来像是：

INSERT  INTO  t  (x)  VALUES  (?);  select  scope_identity()

• 其他方言，如 pymssql，在 INSERT 语句后调用SELECT scope_identity() AS lastrowid。如果将标志use_scope_identity=False传递给create_engine()，则将改为使用语句SELECT @@identity AS lastrowid。
  

包含IDENTITY列的表将禁止明确引用标识列的插入语句。SQLAlchemy 方言将检测到当使用核心insert()构造创建的 INSERT 构造引用标识列时（而不是普通的字符串 SQL），在这种情况下，将在插入语句执行之前发出SET IDENTITY_INSERT ON，并在执行后发出SET IDENTITY_INSERT OFF。给定此示例：

m = MetaData()
t = Table('t', m, Column('id', Integer, primary_key=True),
                Column('x', Integer))
m.create_all(engine)
with engine.begin() as conn:
    conn.execute(t.insert(), {'id': 1, 'x':1}, {'id':2, 'x':2})

上述列将使用 IDENTITY 创建，但我们发出的 INSERT 语句指定了显式值。在回显输出中，我们可以看到 SQLAlchemy 如何处理这个问题：

CREATE  TABLE  t  (
  id  INTEGER  NOT  NULL  IDENTITY(1,1),
  x  INTEGER  NULL,
  PRIMARY  KEY  (id)
)
COMMIT
SET  IDENTITY_INSERT  t  ON
INSERT  INTO  t  (id,  x)  VALUES  (?,  ?)
((1,  1),  (2,  2))
SET  IDENTITY_INSERT  t  OFF
COMMIT

这是一个适用于测试和批量插入场景的辅助用例。

控制“开始”和“增量”

使用传递给Identity对象的Identity.start和Identity.increment参数提供对IDENTITY生成器的“开始”和“增量”值的特定控制：

from sqlalchemy import Table, Integer, Column, Identity
test = Table(
    'test', metadata,
    Column(
        'id',
        Integer,
        primary_key=True,
        Identity(start=100, increment=10)
    ),
    Column('name', String(20))
)

上述Table对象的 CREATE TABLE 将是：

CREATE  TABLE  test  (
  id  INTEGER  NOT  NULL  IDENTITY(100,10)  PRIMARY  KEY,
  name  VARCHAR(20)  NULL,
  )

注意

Identity对象除了start和increment之外还支持许多其他参数。这些参数在 SQL Server 中不受支持，在生成 CREATE TABLE ddl 时将被忽略。

从版本 1.3.19 开始更改：Identity对象现在用于影响 SQL Server 下的Column的IDENTITY生成器。以前，使用的是Sequence对象。由于 SQL Server 现在支持真实的序列作为单独的构造，因此从 SQLAlchemy 版本 1.4 开始，Sequence将以正常的方式运行。

SqlAlchemy 2.0 中文文档（五十二）（5）https://developer.aliyun.com/article/1563144