如何在 SQLAlchemy 中表示自定义 PostgreSQL 域？

Question

我开始将 Alembic 合并到我的项目中，该项目已经使用了 SQLAlchemy 表定义。目前，我的数据库架构是在应用程序外部进行管理的，我想将整个架构放入我的表定义文件中。

在 PostgreSQL 中，我使用自定义域来存储电子邮件地址。 PostgreSQL DDL 是：

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

在 SQLAlchemy 中如何表示该域的创建以及将其用作列数据类型？

Answer 1

这可能远不是一个可行的解决方案，但我认为最好的方法是子类

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

显然，这并不完整，但如果你非常想要这个，它应该给你一个很好的起点。 :)

Answer 2

使用 SQLAlchemy 等工具的原因之一是数据库独立性（除了 ORM 之外）。

但是，使用像这样的低级构造（通常是非常特定于数据库的）使得“数据库独立性”不再成为争论，所以我会选择在 alembic 迁移中编写一个简单的

op.execute

。

这通常是一个非常可以接受的权衡，因为它使源代码更更简单，并且更不容易出错。

如果您依赖于仅在一个数据库后端中可用的数据库功能（另一个示例可能是 PostgreSQL 中的

ltree

或

hstore

），那么我在使用迁移时不会发现任何问题，这也会仅适用于目标后端。

所以你可以这样做：

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

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

另一方面，如果您计划支持不同的后端，这将不起作用。

Answer 3

SQLAlchemy 从 v2.0 开始直接支持创建和删除域。

import sqlalchemy as sa
from sqlalchemy

engine = sa.create_engine('postgresql+psycopg2:///so', echo=True)

metadata = sa.MetaData()

UpperChar = DOMAIN('upper_char', sa.String, check='VALUE = UPPER(VALUE)', metadata=metadata)

# For demonstration purposes. Usually we would drop / create via the metadata object.
UpperChar.drop(engine)
UpperChar.create(engine)

tbl = sa.Table(
    't18662846',
    metadata,
    sa.Column('col', UpperChar),
)


tbl.drop(engine, checkfirst=True)
tbl.create(engine)

Alembic 支持似乎有点不完整：从 1.14.0 开始，迁移似乎会创建域，但不会更改或删除它们。