11

我的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())

whereGUID是sqlalchemy 文档中描述的自定义类型（完全相同）

现在当我跑

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 ###

但是在应用 upgrade -> 期间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 类型一起工作？

4

6 回答 6

8

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

将其替换为GUID()（在添加 import 行之后from your.models.custom_types import GUID）也可以，但是升级脚本与您的模型代码相关联，这可能不是一件好事。

于 2013-03-27T20:44:42.457 回答
4

我有一个类似的问题并解决如下：

假设您有以下模块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

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

于 2013-08-13T11:29:26.750 回答
1

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

我的 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()

希望这可以帮助！

于 2019-06-11T19:45:11.083 回答
1

跟进@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.")，上面的类型将出现sa.CHAR(...)在 alembic 迁移中。

于 2021-08-10T15:44:44.247 回答
1

简短回答（使用 sqlalchemy 版本1.4.25）：

从文档中：

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

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

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

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

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

长答案：

我在其中BinaryUUID定义了我的自定义类型，uuid_type_mysql.py并将其导入并在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
...

在此之后，对我来说一切都很好，而且不需要手动编辑生成的迁移。

于 2021-10-21T18:34:07.393 回答
0

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

Class GUID(types.TypeDecorator)
    impl = CHAR

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

    # You type logic here.

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

于 2017-05-05T17:23:07.510 回答