How to use insert_before/insert_after using a rewriter.

[jtknights08]

``I inherited some migration code and we semi routinely need to add columns to existing tables. The instructions for the migration says to add the insert_before or insert_after once alembic has been ran but I was hoping to fold this in since I know the order of the columns based on a list which will be easy to programmatically add.

The downside is I do not understand how this rewriter works nor what its asking me for. I tried using an example from the code and have tried many different ways.

Here is an example seemed close/simple:

from alembic.autogenerate import rewriter
from alembic.operations import ops

writer = rewriter.Rewriter()

@writer.rewrites(ops.AddColumnOp)
def add_column(context, revision, op):
    ops.AddColumnOp.batch_add_column(op.table_name, op.column_name, insert_before='Table1')
    return [op, ops.AlterColumnOp(op.table_name, op.column_name)]

...
                context.configure(
                    url=rec["url"],
                    output_buffer=buffer_,
                    target_metadata=target_metadata.get(name),
                    literal_binds=True,
                    dialect_opts={"paramstyle": "named"},
                    include_name=include_name,
                    process_revision_directives=writer,
                    render_as_batch=True,
                    compare_type=True
                )

My result:

    with op.batch_alter_table('User', schema=None) as batch_op:
        batch_op.add_column(sa.Column('fake', sa.String(length=20), nullable=False))

If there is any more info needed let me know.

[zzzeek]

I think you just want to take the AddColumnOp you were given and set op.insert_before

# ... fragmented env.py script ....

from alembic.autogenerate import rewriter
from alembic.operations import ops

writer = rewriter.Rewriter()

@writer.rewrites(ops.AddColumnOp)
def add_column(context, revision, op):
    op.insert_before = "column_one"
    return op

note insert_before refers to the name of a column, not a table

[zzzeek]

this extension API is unfortunately read the source code stuff.

it looks like the insert_before thing is in the objects kw collection:

op.kw['insert_before'] = 'colname'

[jtknights08]

Same result using:

@writer.rewrites(ops.AddColumnOp)
def add_column(context, revision, op):
    print(op)
    print(op.column)
    print(dir(op))
    print(op.kw)
    op.kw['insert_before'] = 'column1'
    print(op.kw)

    return op

<alembic.operations.ops.AddColumnOp object at 0x7fdcfc174fd0>
User.fake
['annotations', 'class', 'delattr', 'dict', 'dir', 'doc', 'eq', 'format', 'ge', 'getattribute', 'gt', 'hash', 'init', 'init_subclass', 'le', 'lt', 'module', 'ne', 'new', 'reduce', 'reduce_ex', 'repr', 'setattr', 'sizeof', 'str', 'subclasshook', 'weakref', '_mutations', 'add_column', 'batch_add_column', 'column', 'from_column', 'from_column_and_tablename', 'info', 'kw', 'reverse', 'schema', 'table_name', 'to_column', 'to_diff_tuple']
{}
{'insert_before': 'column1'}

Is it possible I'm not calling recreate="always" in the correct spot?

Is there something I can import so I can see "op" path and type hinting?

[zzzeek]

it's mostly possible that this attribute is not supported by autogenerate right now, so when it's set on the op object, the render step doesn't produce it in the Python code anyway.

the only workaround to making this automated would be rewriting the renderer at

alembic/alembic/autogenerate/render.py

Line 434 in 679555c

@renderers.dispatch_for(ops.AddColumnOp)

to include this attribute.

If I had the actual issue you have, I would likely not be trying to automate this and instead would include a test suite that ensures the columns for tables are rendered in the way that you expect, after they are created, using inspect. that way you can get all the columns of a table and check them for whatever conditions are needed, which will provide an alert that developers need to use this include_column parameter in the way that's required.

[jtknights08]

Thank you for your help, as I mentioned in my other answer, we adopted alembic to reduce human error with very large datasets and schemas. I will at least create a pull request with my own solution that way if the community wants it then its in.

[zzzeek]

well we'd want "insert_before" / "insert_after" to be rendered by alembic's built-in column renderer, that's mostly the bug here

[jtknights08]

I came up with my own solution for this. We have substantial amount of tables/columns and we adopted alembic to reduce human error so the more we can do programmatically the better. Here is the solution I did, I will also do a pull request sometime for this once I've run it enough.

But here is the code you would put in your env.py.

def _add_column_custom(autogen_context, op):
    kwargs = None
    if op.kw:
        kwargs = [(key, str(value)) for key, value in op.kw.items()]
    schema, tname, column = op.schema, op.table_name, op.column
    if autogen_context._has_batch:
        template = "%(prefix)sadd_column(%(column)s)"

    else:
        template = "%(prefix)sadd_column(%(tname)r, %(column)s"
        if schema:
            template += ", schema=%(schema)r"
        template += ")"
    print(kwargs)
    text = template % {
        "prefix": _alembic_autogenerate_prefix(autogen_context),
        "tname": tname,
        "column": _render_column(column, autogen_context, kwargs),
        "schema": schema,
    }
    return text

def _render_column_custom(column, autogen_context, kwargs=None):
    rendered = _user_defined_render("column", column, autogen_context)
    if rendered is not False:
        return rendered

    args = []
    opts = []

    if column.server_default:

        rendered = _render_server_default(  # type:ignore[assignment]
            column.server_default, autogen_context
        )
        if rendered:
            if _should_render_server_default_positionally(
                column.server_default
            ):
                args.append(rendered)
            else:
                opts.append(("server_default", rendered))

    if (
        column.autoincrement is not None
        and column.autoincrement != sqla_compat.AUTOINCREMENT_DEFAULT
    ):
        opts.append(("autoincrement", column.autoincrement))

    if column.nullable is not None:
        opts.append(("nullable", column.nullable))

    if column.system:
        opts.append(("system", column.system))

    comment = column.comment
    if comment:
        opts.append(("comment", "%r" % comment))

    if kwargs:
        opts += kwargs

    return "%(prefix)sColumn(%(name)r, %(type)s, %(args)s%(kwargs)s)" % {
        "prefix": _sqlalchemy_autogenerate_prefix(autogen_context),
        "name": _ident(column.name),
        "type": _repr_type(column.type, autogen_context),
        "args": ", ".join([str(arg) for arg in args]) + ", " if args else "",
        "kwargs": (
            ", ".join(
                ["%s=%s" % (kwname, val) for kwname, val in opts]
                + [
                    "%s=%s"
                    % (key, _render_potential_expr(val, autogen_context))
                    for key, val in sqla_compat._column_kwargs(column).items()
                ]
            )
        ),
    }

_render_column = _render_column_custom
_add_column = _add_column_custom