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?

This question seems a little bit more advanced, maybe you can find better help at the SQLAlchemy list. For the usage of the type I'd recommend finding a way to subclass `TEXT` and changing its name to `email_address`. This would be the most simple way (but I don't know how to do it). The latter part could probably be done by literal SQL, because I'm not sure SQLAlchemy has an interface for creating types that don't exist yet. — javex, Sep 06 '13 at 21:05
@javex for now I'm creating the domain by issuing custom DDL and I've subclassed `UserDefinedType` to return `email_address` for its column type definition. Not ideal but it works well enough. I may check out the SQLA list, thank you for this suggestion. — skyler, Sep 09 '13 at 18:37

score 1 · Answer 1 · answered Nov 02 '15 at 15:24

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. :)

score 1 · Answer 2 · answered Jul 01 '20 at 10:22

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.

How to represent a custom PostgreSQL domain in SQLAlchemy?

2 Answers2

Linked