-
-
Notifications
You must be signed in to change notification settings - Fork 321
Closed
Labels
Description
moved from sqlalchemy/sqlalchemy#5574
# Insert code here
from sqlalchemy import Column
from sqlalchemy import Float
from sqlalchemy import Integer
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import Session
from sqlalchemy import create_engine
#url = "sqlite:////tmp/sqlite.db" ## different error
url = "mysql://test:test@localhost/test"
# first let's create the old table
BasePrep = declarative_base()
class TestPrep(BasePrep):
__tablename__ = "test123"
id = Column(Integer, primary_key=True)
label_x = Column(Float, nullable=False, server_default="0")
label_y = Column(Float, nullable=False, server_default="0")
enginePrep = create_engine(url)
BasePrep.metadata.create_all(enginePrep)
# this is the new one
Base = declarative_base()
class Test(Base):
__tablename__ = "test123"
id = Column(Integer, primary_key=True)
label_x = Column(Float, default=0) # dies
label_y = Column(Float, nullable=True) # also dies
# now try to migrate this
from alembic.runtime.migration import MigrationContext
from alembic.autogenerate import api
from alembic.operations.base import Operations
engine = create_engine(url)
session = Session(engine)
ctx = MigrationContext(dialect=session.bind.dialect, connection=session.connection(), opts={})
res = api.produce_migrations(ctx,Test.__table__.metadata)
ops = Operations(ctx, ctx.impl)
# If there's an Alembic method to do this without looping, I haven't found it.
def all_ops(op):
if hasattr(op,"ops"):
for o in op.ops:
yield from all_ops(o)
else:
yield op
for op in all_ops(res.upgrade_ops):
print(type(op),vars(op))
ops.invoke(op)
Reactions are currently unavailable