特定于方言的 SQLAlchemy 声明性列默认值

Question

精简版

在 SQLAlchemy 的 ORM 列声明中，我如何server_default=sa.FetchedValue()在一种方言和default=somePythonFunction另一种方言上使用，以便我真正的 DBMS 可以使用触发器填充事物，并且我的测试代码可以针对 sqlite 编写？

背景

我正在使用 SQLAlchemy 的声明性 ORM 来处理 Postgres 数据库，但尝试针对 编写单元测试sqlite:///:memory:，并且遇到了在其主键上计算默认值的列的问题。对于一个最小的例子：

CREATE TABLE test_table(
    id VARCHAR PRIMARY KEY NOT NULL
       DEFAULT (lower(hex(randomblob(16))))
)

SQLite 本身对这个表定义 ( sqlfiddle )非常满意，但 SQLAlchemy 似乎无法计算出新创建的行的 ID。

class TestTable(Base):
    __tablename__ = 'test_table'
    id = sa.Column(
              sa.VARCHAR,
              primary_key=True,
              server_default=sa.FetchedValue())

像这样的定义在 postgres 中工作得很好，但在 sqlite 中死了（正如你在 Ideone 上看到的），FlushError当我打电话时Session.commit：

sqlalchemy.orm.exc.FlushError：实例<TestTable at 0x7fc0e0254a10>具有NULL身份密钥。如果这是自动生成的值，请检查数据库表是否允许生成新的主键值，以及映射的 Column 对象是否配置为期望这些生成的值。还要确保这flush()不是在不适当的时间发生，例如在load()事件中。

的文档FetchedValue 警告我们，这可能发生在不支持以下RETURNING条款的方言上INSERT：

对于使用触发器生成主键值的特殊情况，并且使用的数据库不支持 RETURNING 子句，可能需要放弃使用触发器，而是将 SQL 表达式或函数作为“预执行”应用表达：

t = Table('test', meta,
        Column('abc', MyType, default=func.generate_new_value(), 
               primary_key=True)
)

func.generate_new_value是没有定义的其他地方SQLAlchemy的，如此看来，他们打算我不是产生在Python，否则默认写一个单独的功能做一个SQL查询来生成在DBMS的默认值。我可以这样做，但问题是，我只想为 SQLite 这样做，因为FetchedValue在 postgres 上做的正是我想要的。

死胡同

• 子类化Column可能不起作用。我在源代码中找不到任何东西告诉 Column 正在使用什么方言，并且default和server_default字段的行为是在类之外定义的

• 编写一个在真实 DBMS 上手动调用触发器的 Python 函数会产生竞争条件。通过更改隔离级别来避免竞争条件会导致死锁。

我目前的解决方法

不好，因为它破坏了连接到真正 postgres 的集成测试。

import sys
import sqlalchemy as sa

def trigger_column(*a, **kw):
    python_default = kw.pop('python_default')
    if 'unittest' in sys.modules:
        return sa.Column(*a, default=python_default, **kw)
    else
        return sa.Column(*a, server_default=sa.FetchedValue(), **kw)

Answer 1

不是直接回答你的问题，但希望对某人有帮助

我的问题是想根据方言更改排序规则，这是我的解决方案：

from sqlalchemy import Unicode
from sqlalchemy.ext.compiler import compiles

@compiles(Unicode, 'sqlite')    
def compile_unicode(element, compiler, **kw):
    element.collation = None
    return compiler.visit_unicode(element, **kw)

这仅更改 sqlite 的所有 Unicode 列的排序规则。

这是一些文档：http://docs.sqlalchemy.org/en/latest/core/custom_types.html#overriding-type-compilation