Alembic：如何在模型中迁移自定义类型？

Question

我的User 模特是

class User(UserMixin, db.Model):
    __tablename__ = 'users'
    # noinspection PyShadowingBuiltins
    uuid = Column('uuid', GUID(), default=uuid.uuid4, primary_key=True,
                  unique=True)
    email = Column('email', String, nullable=False, unique=True)
    _password = Column('password', String, nullable=False)
    created_on = Column('created_on', sa.types.DateTime(timezone=True),
                        default=datetime.utcnow())
    last_login = Column('last_login', sa.types.DateTime(timezone=True),
                        onupdate=datetime.utcnow())

其中GUID 是sqlalchemy docs 中描述的自定义类型（完全相同）

现在当我跑步时

alembic revision --autogenerate -m "Added initial table"

我的upgrade() 是

def upgrade():
    ### commands auto generated by Alembic - please adjust! ###
    op.create_table('users',
    sa.Column('uuid', sa.GUID(), nullable=False),
    sa.Column('email', sa.String(), nullable=False),
    sa.Column('password', sa.String(), nullable=False),
    sa.Column('created_on', sa.DateTime(timezone=True), nullable=True),
    sa.Column('last_login', sa.DateTime(timezone=True), nullable=True),
    sa.PrimaryKeyConstraint('uuid'),
    sa.UniqueConstraint('email'),
    sa.UniqueConstraint('uuid')
    )
    ### end Alembic commands ###

但是在应用升级过程中-> alembic upgrade head，我看到了

File "alembic/versions/49cc74d0da9d_added_initial_table.py", line 20, in upgrade
    sa.Column('uuid', sa.GUID(), nullable=False),
AttributeError: 'module' object has no attribute 'GUID'

我怎样才能使它与GUID/custom type here 一起工作？

Answer 1

简答（使用sqlalchemy版本1.4.25）：

来自documentation：

对于用户定义的类型，即不在其中的任何自定义类型 sqlalchemy。模块命名空间，默认情况下 Alembic 将使用 自定义类型的 __module__ 值：

Column("my_column", myapp.models.utils.types.MyCustomType())

上述类型的导入必须再次出现在 手动迁移，或者将其添加到 script.py.mako。

因此，将您的my_module 导入script.py.mako（并且您可能需要在models.py 以外的文件中定义您的自定义类型）：

...
from alembic import op
import sqlalchemy as sa
import my_module
${imports if imports else ""}
...

长答案：

我在uuid_type_mysql.py 中定义了我的自定义类型BinaryUUID，我将它导入models.py 并在那里使用它：

models.py:

...
from .uuid_type_mysql import BinaryUUID
...

在使用flask db migrate 生成迁移后，我会在迁移文件中得到这个：

...
sa.Column('public_id', my_module.uuid_type_mysql.BinaryUUID(length=16), nullable=False),
...

问题是迁移不知道my_module，因为它没有被导入。

按照文档中的建议将import my_module 添加到script.py.mako 后，现在模块已导入到迁移文件中：

...
from alembic import op
import sqlalchemy as sa
import my_module
...

在此之后，我的一切工作正常，而且不需要手动编辑生成的迁移。

Answer 2

跟进@Red-Tune-84的解决方案

class GUID(types.TypeDecorator)
  impl = CHAR

  def __repr__(self):
    return self.impl.__repr__()

  # You type logic here.

确实有效，但您可能需要并行设置env.py 配置user_module_prefix。

例如，使用context.configure(..., user_module_prefix="sa.")，上面的类型将在alembic迁移中显示为sa.CHAR(...)。

Answer 3

我的解决方案使用sqlalchemy_utils.types.uuid.UUIDType，如果您在没有UUID 类型的数据库上，它使用CHAR(32) 或BINARY(16) 来表示UUID。您需要在迁移中考虑到这一点，它必须在没有 UUID 类型的数据库上创建一个 CHAR(32)/BINARY(16)，并在有它的数据库上创建一个 UUIDType。

我的 SQLAlchemy 类如下所示：

from sqlalchemy_utils.types.uuid import UUIDType
from sqlalchemy import CHAR, Column, Integer

Base = declarative_base()

def get_uuid():
    return str(uuid.uuid4())

class Dashboard(Base):
    __tablename__ = 'dashboards'
    id = Column(Integer, primary_key=True)
    uuid = Column(UUIDType(binary=False), default=get_uuid)

实际的批处理操作是这样的（支持 SQLite、MySQL 和 Postgres）：

from superset import db # Sets up a SQLAlchemy connection

def upgrade():
    bind = op.get_bind()
    session = db.Session(bind=bind)
    db_type = session.bind.dialect.name

    def add_uuid_column(col_name, _type):
        """Add a uuid column to a given table"""
        with op.batch_alter_table(col_name) as batch_op:
            batch_op.add_column(Column('uuid', UUIDType(binary=False), default=get_uuid))
        for s in session.query(_type):
            s.uuid = get_uuid()
            session.merge(s)

        if db_type != 'postgresql':
            with op.batch_alter_table(col_name) as batch_op:
                batch_op.alter_column('uuid', existing_type=CHAR(32),
                                      new_column_name='uuid', nullable=False)
                batch_op.create_unique_constraint('uq_uuid', ['uuid'])

        session.commit()

add_uuid_column('dashboards', Dashboard)
session.close()

希望这会有所帮助！

Answer 4

使用impl 属性类的__repr__ 函数对我来说适用于大多数自定义类型。我发现将迁移定义包含在类中更简洁，而不是担心将导入放入您的env.py 或scripts.py.mako。此外，它还可以轻松地在模块之间移动代码。

Class GUID(types.TypeDecorator)
    impl = CHAR

    def __repr__(self):
        return self.impl.__repr__()

    # You type logic here.

自动迁移将产生CHAR(length=XXX)。

Answer 5

我遇到了类似的问题，解决方法如下：

假设您有以下模块 my_guid，包含（来自您已经引用的页面，稍作命名修改）：

import uuid as uuid_package
from sqlalchemy.dialects.postgresql import UUID as PG_UUID
from sqlalchemy import TypeDecorator, CHAR

class GUID(TypeDecorator):
    impl = CHAR

    def load_dialect_impl(self, dialect):
        if dialect.name == 'postgresql':
            return dialect.type_descriptor(PG_UUID())
        else:
            return dialect.type_descriptor(CHAR(32))

    def process_bind_param(self, value, dialect):
        if value is None:
            return value
        elif dialect.name == 'postgresql':
            return str(value)
        else:
            if not isinstance(value, uuid_package.UUID):
                return "%.32x" % uuid_package.UUID(value)
            else:
                # hexstring
                return "%.32x" % value

    def process_result_value(self, value, dialect):
        if value is None:
            return value
        else:
            return uuid_package.UUID(value)

如果您在模型中使用此 GUID，您只需在 alembic/env.py 处添加三行：

from my_guid import GUID
import sqlalchemy as sa
sa.GUID = GUID

这对我有用。希望对您有所帮助！

Answer 6

您可以根据方言将sa.GUID() 替换为sa.CHAR(32) 或UUID()（在添加导入行from sqlalchemy.dialects.postgresql import UUID 之后）。

用GUID() 替换它（在添加导入行from your.models.custom_types import GUID 之后）也可以，但是升级脚本会绑定到您的模型代码，这可能不是一件好事。