简短版本
在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上),当我调用Session.commit
时会出现FlushError
:
sqlalchemy.orm.exc.FlushError
:实例<TestTable at 0x7fc0e0254a10>
具有NULL
身份验证键。如果这是自动生成的值,请检查数据库表是否允许生成新的主键值,并且映射的列对象已配置为期望这些生成的值。还要确保此flush()
不会在不当时间内发生,例如在load()
事件中。
FetchedValue
的文档警告我们,在不支持INSERT
上的RETURNING
子句的方言中,可能会发生这种情况:
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
是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)