Dialect-specific SQLAlchemy declarative Column defaults

Question

Short Version

In SQLAlchemy's ORM column declaration, how can I use server_default=sa.FetchedValue() on one dialect, and default=somePythonFunction on another, so that my real DBMS can populate things with triggers, and my test code can be written against sqlite?

Background

I'm using SQLAlchemy's declarative ORM to work with a Postgres database, but trying to write unit tests against an sqlite:///:memory:, and running into a problem with columns that have computed defaults on their primary keys. For a minimal example:

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

SQLite itself is quite happy with this table definition (sqlfiddle) but SQLAlchemy seems unable to work out the ID of newly created rows.

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

Definitions like this work just fine in postgres, but die in sqlite (as you can see on Ideone) with a FlushError when I call 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.

The documentation for 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 is not defined anywhere else in SQLAlchemy, so it seems they intend I either generate defaults in Python, or else write a separate function to do a SQL query to generate a default value in the DBMS. I can do that, but the problem is, I only want to do that for SQLite, since FetchedValue does exactly what I want on postgres.

Dead Ends

• Subclassing Column probably won't work. Nothing that I can find in the sources ever tells the Column what dialect is being used, and the behavior of the default and server_default fields is defined outside the class

• Writing a python function that calls the triggers by hand on the real DBMS creates a race condition. Avoiding the race condition by changing the isolation level creates a deadlock.

My Current Workaround

Bad because it breaks integration tests that connect to a real 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

Not a direct answer to you question but hopefully helpful to someone

My problem was wanting to change the collation depending on the dialect, this was my solution:

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)

This changes the collation for all Unicode columns only for sqlite.

Here's some documentation: http://docs.sqlalchemy.org/en/latest/core/custom_types.html#overriding-type-compilation