Alembic: How to transfer a custom type to a model?

Question

My User Model

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

where GUID is a custom type as described in sqlalchemy docs (exactly the same)

Now when i run

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

I get my upgrade() as

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

but while applying the update - alembic upgrade head , I see

 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' 

How can I make it work with GUID / custom type here?

Answer 1

You can replace sa.GUID() with sa.CHAR(32) or UUID() (after adding the import line from sqlalchemy.dialects.postgresql import UUID ) depending on the dialect.

Replacing it with GUID() (after adding the import line from your.models.custom_types import GUID ) will also work, but the update script is bound to your model code, which may not be very good.

Answer 2

I had a similar problem and it was solved as follows:

Suppose you have the following my_guid module containing (from the page you have already quoted, with minor naming changes):

 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) 

If you use this GUID in your models, you just need to add three lines to alembic/env.py :

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

It worked for me. Hope this helps!

Answer 3

Using the __repr__ function of the __repr__ attribute impl worked for me for most custom types. I find it cleaner to have a migration definition inside the class, rather than worry about importing into env.py or scripts.py.mako . In addition, it simplifies code movement between modules.

 Class GUID(types.TypeDecorator) impl = CHAR def __repr__(self): return self.impl.__repr__() # You type logic here. 

Automation will result in CHAR(length=XXX) .

Answer 4

My solution uses sqlalchemy_utils.types.uuid.UUIDType , which uses CHAR(32) or BINARY(16) to represent UUID if you are in a database without a UUID type. This must be taken into account during the transfer, which should create CHAR(32)/BINARY(16) for a database without the UUID type and UUIDType for databases with it.

My SQLAlchemy class looks like this:

 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) 

and the actual batch operation is as follows (which supports SQLite, MySQL and 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() 

Hope this helps!