sqlalchemy-spanner 1.7.0

数据库URL

为了连接数据库，必须在连接创建步骤使用其URL。SQLAlchemy 1.3和1.4版本在方言前缀部分有细微差别。

# for SQLAlchemy 1.3:
spanner:///projects/project-id/instances/instance-id/databases/database-id

# for SQLAlchemy 1.4:
spanner+spanner:///projects/project-id/instances/instance-id/databases/database-id

要直接传递自定义客户端对象以供使用，请按以下方式创建引擎：

engine = create_engine(
    "spanner+spanner:///projects/project-id/instances/instance-id/databases/database-id",
    connect_args={'client': spanner.Client(project="project-id")}
)

创建一个表

from sqlalchemy import (
    Column,
    Integer,
    MetaData,
    String,
    Table,
    create_engine,
)

engine = create_engine(
    "spanner:///projects/project-id/instances/instance-id/databases/database-id"
)
metadata = MetaData(bind=engine)

user = Table(
    "users",
    metadata,
    Column("user_id", Integer, primary_key=True),
    Column("user_name", String(16), nullable=False),
)

metadata.create_all(engine)

插入一行

import uuid

from sqlalchemy import (
    MetaData,
    Table,
    create_engine,
)

engine = create_engine(
    "spanner:///projects/project-id/instances/instance-id/databases/database-id"
)
user = Table("users", MetaData(bind=engine), autoload=True)
user_id = uuid.uuid4().hex[:6].lower()

with engine.begin() as connection:
    connection.execute(user.insert(), {"user_id": user_id, "user_name": "Full Name"})

读取

from sqlalchemy import MetaData, Table, create_engine, select

engine = create_engine(
    "spanner:///projects/project-id/instances/instance-id/databases/database-id"
)
table = Table("users", MetaData(bind=engine), autoload=True)

with engine.begin() as connection:
    for row in connection.execute(select(["*"], from_obj=table)).fetchall():
        print(row)

交错表

team = Table(
    "team",
    metadata,
    Column("team_id", Integer, primary_key=True),
    Column("team_name", String(16), nullable=False),
)
team.create(engine)

client = Table(
    "client",
    metadata,
    Column("team_id", Integer, primary_key=True),
    Column("client_id", Integer, primary_key=True),
    Column("client_name", String(16), nullable=False),
    spanner_interleave_in="team",
    spanner_interleave_on_delete_cascade=True,
)
client.add_is_dependent_on(team)

client.create(engine)

注意：交错表之间存在依赖关系，因此必须先创建父表。在创建具有此功能的表时，请确保在子表上调用add_is_dependent_on()，以请求SQLAlchemy在创建子表之前创建父表。

唯一约束

Cloud Spanner不支持直接创建唯一约束。为了实现列值的唯一性，应使用唯一索引。

而不是直接创建唯一约束

Table(
    'table',
    metadata,
    Column('col1', Integer),
    UniqueConstraint('col1', name='uix_1')
)

创建一个唯一索引

Table(
    'table',
    metadata,
    Column('col1', Integer),
    Index("uix_1", "col1", unique=True),
)

自动提交模式

Spanner方言支持SERIALIZABLE和AUTOCOMMIT隔离级别。SERIALIZABLE是默认级别，其中事务需要手动提交。AUTOCOMMIT模式对应于在查询执行时自动提交查询。

隔离级别更改示例

from sqlalchemy import create_engine

eng = create_engine("spanner:///projects/project-id/instances/instance-id/databases/database-id")
autocommit_engine = eng.execution_options(isolation_level="AUTOCOMMIT")

自动事务重试

在默认的SERIALIZABLE模式下，事务可能会因Aborted异常而失败。这是一种暂时性错误，通常发生在防止并发修改导致数据损坏的情况下。尽管原始事务变得不可操作，但简单的查询重试可以解决问题。

然而，这可能需要手动重复执行一系列操作，这些操作是在失败的交易中执行的。为了简化，Spanner连接API跟踪当前交易中执行的所有操作及其结果校验和。如果交易因Aborted异常而失败，连接API将自动启动一个新交易，并重新运行所有跟踪的操作，检查其结果是否与原始交易中的结果相同。如果数据已更改且结果不同，则交易将被丢弃，因为没有方法可以自动重试。

在AUTOCOMMIT模式下，自动事务重试机制被禁用，因为每个操作都是即时提交的，并且不会发生Aborted异常。

自增ID

由于性能原因，Cloud Spanner不支持自增ID机制（请参阅更多详情）。我们建议您使用Python的uuid模块来生成主键字段，以避免创建单调递增的键。

尽管不鼓励这样做，但如果您确实需要此功能，可以手动模拟如下

with engine.begin() as connection:
    top_id = connection.execute(
        select([user.c.user_id]).order_by(user.c.user_id.desc()).limit(1)
    ).fetchone()
    next_id = top_id[0] + 1 if top_id else 1

    connection.execute(user.insert(), {"user_id": next_id})

查询提示

Spanner方言支持查询提示，这提供了设置额外查询执行参数的能力。用法示例

session = Session(engine)

Base = declarative_base()

class User(Base):
    """Data model."""

    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    name = Column(String(50))


query = session.query(User)
query = query.with_hint(
    selectable=User, text="@{FORCE_INDEX=index_name}"
)
query = query.filter(User.name.in_(["val1", "val2"]))
query.statement.compile(session.bind)

只读事务

默认情况下，由Spanner连接产生的交易处于读写模式。然而，某些应用程序需要授予用户/方法只读访问权限的能力；在这些情况下，Spanner方言支持read_only执行选项，该选项将连接切换到只读模式

with engine.connect().execution_options(read_only=True) as connection:
    connection.execute(select(["*"], from_obj=table)).fetchall()

请注意，执行选项是按需应用的 - 在execute()方法调用之前。

在事务进行时无法更改连接的只读/读写模式 - 您必须首先提交或回滚它。

陈旧读取

要使用Spanner的陈旧读取与SQLAlchemy一起，您可以调整连接执行选项，以获得期望的陈旧值。例如

# maximum staleness
with engine.connect().execution_options(
    read_only=True,
    staleness={"max_staleness": datetime.timedelta(seconds=5)}
) as connection:
    connection.execute(select(["*"], from_obj=table)).fetchall()

# exact staleness
with engine.connect().execution_options(
    read_only=True,
    staleness={"exact_staleness": datetime.timedelta(seconds=5)}
) as connection:
    connection.execute(select(["*"], from_obj=table)).fetchall()

# min read timestamp
with engine.connect().execution_options(
    read_only=True,
    staleness={"min_read_timestamp": datetime.datetime(2021, 11, 17, 12, 55, 30)}
) as connection:
    connection.execute(select(["*"], from_obj=table)).fetchall()

# read timestamp
with engine.connect().execution_options(
    read_only=True,
    staleness={"read_timestamp": datetime.datetime(2021, 11, 17, 12, 55, 30)}
) as connection:
    connection.execute(select(["*"], from_obj=table)).fetchall()

请注意，在将连接返回到连接池时，将丢弃设置的选项。

请求优先级

为了在Cloud Spanner中使用请求优先级功能，SQLAlchemy提供了一个execution_options参数

from google.cloud.spanner_v1 import RequestOptions

with engine.connect().execution_options(
    request_priority=RequestOptions.Priority.PRIORITY_MEDIUM
) as connection:
    connection.execute(select(["*"], from_obj=table)).fetchall()

DDL和事务

DDL语句是在常规事务机制之外执行的，这意味着DDL语句在正常事务回滚时不会回滚。

删除表

默认情况下，Cloud Spanner不删除具有二级索引和/或外键约束的表。然而，在Spanner方言为SQLAlchemy中，这个限制被省略了 - 如果您尝试删除的表有索引/外键，它们将在删除表之前自动删除。

数据类型

数据类型表映射SQLAlchemy类型到Cloud Spanner类型

其他限制

• 不支持WITH RECURSIVE语句。

• 不支持命名架构。

• 不支持临时表。

• 数值类型维度（缩放和精度）是常量。请参阅文档。

在 Spanner 模拟器上运行测试

当在模拟器上执行时，方言测试套件可以在 Spanner 模拟器 上运行。一些与数据类型的 NULL 值相关的测试会被跳过。