python - 方言特定的 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 中工作得很好，但在带有 FlushError 的 sqlite 中死亡(如您所见 on Ideone )当我调用 Session.commit :

sqlalchemy.orm.exc.FlushError: Instance <TestTable at 0x7fc0e0254a10> has a NULL identity key. If this is an auto-generated value, check that the database table allows generation of new primary key values, and that the mapped Column object is configured to expect these generated values. Ensure also that this flush() is not occurring at an inappropriate time, such as within a load() event.

FetchedValue 的文档warns us that this can happen on dialects that don't support the RETURNING clause on INSERT :

For special situations where triggers are used to generate primary key values, and the database in use does not support the RETURNING clause, it may be necessary to forego the usage of the trigger and instead apply the SQL expression or function as a “pre execute” expression:

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

func.generate_new_value是not defined anywhere else in SQLAlchemy ，所以他们似乎打算我要么在 Python 中生成默认值，要么编写一个单独的函数来执行 SQL 查询以在 DBMS 中生成默认值。我可以这样做，但问题是，我只想为 SQLite 这样做，因为 FetchedValue在 postgres 上做我想要的。

死胡同

• 子类化 Column 可能行不通。我在资源中找不到的任何内容都告诉专栏正在使用什么方言，以及 default 的行为。和 server_default字段是 defined outside the class

• 编写一个在真实 DBMS 上手动调用触发器的 python 函数会产生竞争条件。通过更改 isolation level 来避免竞争条件造成死锁。

我当前的解决方法

不好，因为它破坏了连接到真实 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