Example of inline autogenerated use?

18 views

Skip to first unread message

Jason Pascucci

unread,

Nov 5, 2020, 2:14:52 AM11/5/20

to sqlalchemy-alembic

I couldn't lay my hands on an example of using MigrationContext(connect=<blah>) and running the results of produce_migrations(Base.metadata, mc).

I got close, I think, evaling the results of render_python_code (which failed due to spaces, but I can wrap it into a def, I guess), but I think there's got to be a better way, I'm just not seeing it.

(NB: I can guarantee that the migrations will always be simple alters to add only, &c, so not using the whole infrastructure lets it be consistent, and due to the nature of the project, I'd like to make it simple)

Thanks

JRP

Mike Bayer

unread,

Nov 5, 2020, 8:28:03 AM11/5/20

to 'Carol Guo' via sqlalchemy-alembic

Not quite given in an example, I guess I could add as a recipe but then people will be using it which as you've noted isn't reliable in the general sense, let's put together how to get the MigrationScript, which it looks like you have, with then how to make the Operations object programmatically, which is at https://alembic.sqlalchemy.org/en/latest/ops.html#alembic.operations.Operations, then there's a method invoke() which is at https://alembic.sqlalchemy.org/en/latest/ops.html#alembic.operations.Operations.invoke , so we can use that to run the ops. We have to traverse them too where I'll use a little bit of assumptions to make it succinct, seems to work for simple operations, like this:

migrations = produce_migrations(mc, metadata)

operations = Operations(mc)

stack = [migrations.upgrade_ops]

while stack:

elem = stack.pop(0)

if hasattr(elem, "ops"):

stack.extend(elem.ops)

else:

# work around Alembic issue #753

if hasattr(elem, "column"):

elem.column = elem.column.copy()

operations.invoke(elem)

full example below

from alembic.migration import MigrationContext

from alembic.autogenerate import produce_migrations

from sqlalchemy.schema import SchemaItem

from sqlalchemy.types import TypeEngine

from sqlalchemy import create_engine, MetaData, Column, Integer, String, Table

import pprint

from alembic.operations import Operations

engine = create_engine("mysql://scott:tiger@localhost/test", echo=True)

with engine.connect() as conn:

m = MetaData()

m.reflect(conn)

m.drop_all(conn)

conn.execute(

"""

create table foo (

id integer not null primary key,

old_data varchar(50),

x integer

)"""

)

conn.execute(

"""

create table bar (

data varchar(50)

)"""

)

metadata = MetaData()

Table(

"foo",

metadata,

Column("id", Integer, primary_key=True),

Column("data", Integer),

Column("x", Integer, nullable=False),

)

Table("bat", metadata, Column("info", String(100)))

mc = MigrationContext.configure(engine.connect())