如何在 SQLALchemy Core 中`SET CONSTRAINTS ... DEFERRED`

Question

我正在使用 SQLAlchemy 和 PostgreSQL。Postgres 支持执行延迟约束，这允许我们将检查表上的约束推迟到事务结束。

例如，在 SQLAlchemy 中，我可能会定义一个这样的表：

t_group_categories = Table('group_categories', metadata,
    Column('id', Integer, primary_key=True),
    Column('group_id', Integer, ForeignKey('groups.id', deferrable=True))
)

SQLAlchemy 将生成一个CREATE TABLE类似于以下内容的语句：

CREATE TABLE group_categories
(
  id serial NOT NULL,
  group_id integer,
  CONSTRAINT group_categories_pkey PRIMARY KEY (id),
  CONSTRAINT group_categories_group_id_fkey FOREIGN KEY (group_id)
      REFERENCES groups (id) MATCH SIMPLE
      ON UPDATE NO ACTION ON DELETE CASCADE DEFERRABLE INITIALLY IMMEDIATE
)

据我了解，这DEFERRABLE INITIALLY IMMEDIATE意味着FOREIGN KEY约束将表现为不可延迟约束，除非另有明确说明，这正是我想要的。

问题是我似乎找不到任何关于如何让 SQLAlchemy 核心SET CONSTRAINTS ... DEFERRED在事务内部实际发出命令的信息。例如，假设我有以下代码：

connection = engine.connect()
...
with connection.begin() as transaction:
    # Create a group
    r = connection.execute(
        t_groups.insert().values(...)
    )
    group_id = r.inserted_primary_key

    # Assign a category to the group (ERROR!)
    r2 = connection.execute(
        t_group_categories.insert().values(group_id=group_id, ...)
    )

第一个块只是创建一个新组。然后第二个块尝试为我们刚刚创建的组分配一个类别。问题是，如果没有 special SET CONSTRAINTS ... DEFERRED，我们实际上无法在group_categories不违反表上的外键约束的情况下创建条目，因为事务尚未提交。

在这种情况下，我想做的是将约束检查推迟到事务提交之后。我如何才能真正将约束检查推迟到事务完成之后？

---

笔记：

• 在这个问题 如何设置约束SQLAlchemy的表达式语言递延？是相似的，但 OP 有兴趣使用DEFERRABLE INITIALLY DEFERRED，我不想这样做。相反（如果可能），我希望将我的约束保留为DEFERRABLE INITIALLY IMMEDIATE并明确标记需要推迟约束的实例。
• SQLAlchemy 是生成DEFERRABLE INITIALLY DEFERRED约束的那个，所以我希望/假设它具有在另一侧实际使用此约束的表达方式（即发出SET CONSTRAINTS ... DEFERRED.

---

更新：

• 这样做connection.execute("SET CONSTRAINTS ALL DEFERRED")似乎也没有任何效果；我仍然收到 IntegrityError。
• 这样做connection.execute("SET CONSTRAINTS group_categories_group_id_fkey DEFERRED")的事务块里面也给回一个IntegrityError。

Answer 1

我可以通过使用元数据命名约定来推迟约束，以确保约束具有可在SET CONSTRAINTS语句中使用的名称。

convention = {
    'ix': 'ix_%(column_0_label)s',
    'uq': 'uq_%(table_name)s_%(column_0_name)s',
    'ck': 'ck_%(table_name)s_%(constraint_name)s',
    'fk': 'fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s',
    'pk': 'pk_%(table_name)s',
}

metadata = sa.MetaData(naming_convention=convention)

...
# Specify "initially immediate" 
t_group_categories = sa.Table(
    'group_categories',
    metadata,
    sa.Column('id', sa.Integer, primary_key=True),
    sa.Column(
        'group_id',
        sa.Integer,
        sa.ForeignKey('groups.id', deferrable=True, initially='IMMEDIATE'),
    ),
)

engine = sa.create_engine('postgresql:///test', echo=True, future=True)
metadata.drop_all(engine)
metadata.create_all(engine)

# If the table had multiple FK constraints we would need to find the right one,
# but here we can just unpack the sole contraint.
constraint, = t_group_categories.foreign_key_constraints

with engine.begin() as conn:
    conn.execute(sa.text(f'SET CONSTRAINTS "{constraint.name}" DEFERRED'))
    # Passing id before the group is created will trigger an immediate constraint.
    conn.execute(t_group_categories.insert().values(group_id=1))
    conn.execute(t_groups.insert())