How to represent a custom PostgreSQL domain in SQLAlchemy?

Question

I'm beginning to incorporate Alembic into my project which already uses SQLAlchemy table definitions. At present my DB schema is managed external to my application, and I want to bring the entire schema into my table definitions file.

In PostgreSQL I use a custom domain for storing email addresses. The PostgreSQL DDL is:

CREATE DOMAIN email_address TEXT CHECK (value ~ '.+@.+')

How do I represent the creation of this domain, and the usage of it as a column data type, in SQLAlchemy?

Answer 1

This likely far from a working solution, but I think the best way to do this would be subclass sqlalchemy.schema._CreateDropBase.

from sqlalchemy.schema import _CreateDropBase

class CreateDomain(_CreateDropBase):
    '''Represent a CREATE DOMAIN statement.'''

    __visit_name__ = 'create_domain'

    def __init__(self, element, bind=None, **kw):
        super(CreateDomain, self).__init__(element, bind=bind, **kw)

class DropDomain(_CreateDropBase):
    '''Represent a DROP BASE statement.'''

    __visit_name__ = 'drop_domain'

    def __init__(self, element, bind=None, **kw):
        super(DropDomain, self).__init__(element, bind=bind, **kw)

@compiles(CreateDomain, 'postgresql')
def visit_create_domain(element, compiler, **kw):
    text = '\nCREATE DOMAIN %s AS %s' % (
        compiler.prepare.format_column(element.name),
        compiler.preparer.format_column(element.type_)) # doesn't account for arrays and such I don't think

    default = compiler.get_column_default_string(column)
    if default is not None:
        text += " DEFAULT %s" % default

    return text

Obviously, this is incomplete, but it should give you a good starting point if you want this badly enough. :)

Answer 2

One of the reasons for using something like SQLAlchemy is DB independence (apart from the ORM stuff).

However, using low-level constructs like this which are often very DB specific make "DB independence" a non-argument, so I would opt for writing a simple op.execute in your alembic migration.

This is often a very acceptable trade-off as it makes the source code much simpler and this less error-prone.

If you are relying on features of a DB which are only available in one DB-backend (another example might be ltree or hstore from PostgreSQL), then I don't see any issue in using a migration which will also only work on that targeted back-end.

So you could just do:

def upgrade():
    op.execute("CREATE DOMAIN ...")

def downgrade():
    op.execute("DROP DOMAIN ...")

If on the other hand you plan on supporting different back-ends this won't work.