Materialized view change detection fails if upstream view has changed

[trogers-medasync]

I'm having issues autogenerating a migration.

I have a Materialized view (mat_view) that is dependent on another view (view1) which is dependent on a view (view2) that changed.

view2's changes are found and a replace op is properly detected as needed. view1 is correctly identified as not needing changes. During change detection of mat_view however, it throws a sqlalchemy.exc.ProgrammingError saying mat_view doesn't exist, and during the handling of that error, the db api throws another error saying that view1 (the one that depends on the one that changed) doesn't exist.

I've made sure that the views are registered in dependent order.

I've been able to work around it by generating a migration with mat_view removed from the registered entities, which generates a replace op for view2 and a drop op for mat_view. I then manually remove the drop op for mat_view and it applies fine. That is only feasible at small scales though.

[olirice]

In complex situations like this with multiple nested dependencies there are many things that can go wrong

As a rule of thumb, only change 1 view/matview at a time between auto-gen runs. You'll end up with more migrations, but it'll succeed more of the time.

If that doesn't work in this case, I'd recommend importing each entity.to_sql_statement_create_or_replcae() to produce the SQL upgrade script and manually pasting them in the right order in a (non-autogen) migration

I've made sure that the views are registered in dependent order.

the order in register_entity doesn't have an impact in resolution order, it is solved for during autogen

[trogers-medasync]

The error happens even if there is only one view that is changed for the migration being generated. In the case I'm describing, only view1 needs changed, but an error is thrown when mat_view is being analyzed for changes (which don't exist because the mat view hasn't changed). The last info message before the error is thrown is: Detecting required migration op PGMaterializedView PGMaterializedView: public.mat_view

[trogers-medasync]

This also happens if its just a plain view that depends on a view that depends on another that changed. When View3 is dependent on View2 which is dependent on View1, and only View1 changed so needs replaced, this is the output I get when I run an autogenerate command.

CLI output

``` INFO [alembic_utils.replaceable_entity] Detecting required migration op PGView PGView: public.View1 INFO [alembic_utils.replaceable_entity] Detected ReplaceOp op for PGView PGView: public.View1 INFO [alembic_utils.replaceable_entity] Detecting required migration op PGView PGView: public.View2 INFO [alembic_utils.replaceable_entity] Detecting required migration op PGView PGView: public.View3 Traceback (most recent call last): File "/path/to/project/.venv/lib/python3.11/site-packages/pg8000/legacy.py", line 250, in execute self._context = self._c.execute_simple(operation) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/path/to/project/.venv/lib/python3.11/site-packages/pg8000/core.py", line 658, in execute_simple self.handle_messages(context) File "/path/to/project/.venv/lib/python3.11/site-packages/pg8000/core.py", line 807, in handle_messages raise context.error pg8000.exceptions.DatabaseError: {'S': 'ERROR', 'V': 'ERROR', 'C': '42P01', 'M': 'view "View3" does not exist', 'F': 'tablecmds.c', 'L': '1195', 'R': 'DropErrorMsgNonExistent'} During handling of the above exception, another exception occurred: Traceback (most recent call last): File "/path/to/project/.venv/lib/python3.11/site-packages/sqlalchemy/engine/base.py", line 1905, in _execute_context self.dialect.do_execute( File "/path/to/project/.venv/lib/python3.11/site-packages/sqlalchemy/engine/default.py", line 736, in do_execute cursor.execute(statement, parameters) File "/path/to/project/.venv/lib/python3.11/site-packages/pg8000/legacy.py", line 280, in execute raise cls(msg) pg8000.dbapi.ProgrammingError: {'S': 'ERROR', 'V': 'ERROR', 'C': '42P01', 'M': 'view "View3" does not exist', 'F': 'tablecmds.c', 'L': '1195', 'R': 'DropErrorMsgNonExistent'} The above exception was the direct cause of the following exception: Traceback (most recent call last): File "/path/to/project/.venv/lib/python3.11/site-packages/alembic_utils/simulate.py", line 47, in simulate_entity sess.execute(entity.to_sql_statement_drop(cascade=True)) File "/path/to/project/.venv/lib/python3.11/site-packages/sqlalchemy/orm/session.py", line 1717, in execute result = conn._execute_20(statement, params or {}, execution_options) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/path/to/project/.venv/lib/python3.11/site-packages/sqlalchemy/engine/base.py", line 1710, in _execute_20 return meth(self, args_10style, kwargs_10style, execution_options) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/path/to/project/.venv/lib/python3.11/site-packages/sqlalchemy/sql/elements.py", line 334, in _execute_on_connection return connection._execute_clauseelement( ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/path/to/project/.venv/lib/python3.11/site-packages/sqlalchemy/engine/base.py", line 1577, in _execute_clauseelement ret = self._execute_context( ^^^^^^^^^^^^^^^^^^^^^^ File "/path/to/project/.venv/lib/python3.11/site-packages/sqlalchemy/engine/base.py", line 1948, in _execute_context self._handle_dbapi_exception( File "/path/to/project/.venv/lib/python3.11/site-packages/sqlalchemy/engine/base.py", line 2129, in _handle_dbapi_exception util.raise_( File "/path/to/project/.venv/lib/python3.11/site-packages/sqlalchemy/util/compat.py", line 211, in raise_ raise exception File "/path/to/project/.venv/lib/python3.11/site-packages/sqlalchemy/engine/base.py", line 1905, in _execute_context self.dialect.do_execute( File "/path/to/project/.venv/lib/python3.11/site-packages/sqlalchemy/engine/default.py", line 736, in do_execute cursor.execute(statement, parameters) File "/path/to/project/.venv/lib/python3.11/site-packages/pg8000/legacy.py", line 280, in execute raise cls(msg) sqlalchemy.exc.ProgrammingError: (pg8000.dbapi.ProgrammingError) {'S': 'ERROR', 'V': 'ERROR', 'C': '42P01', 'M': 'view "View3" does not exist', 'F': 'tablecmds.c', 'L': '1195', 'R': 'DropErrorMsgNonExistent'} [SQL: DROP VIEW "public"."View3" cascade] (Background on this error at: https://sqlalche.me/e/14/f405) During handling of the above exception, another exception occurred: Traceback (most recent call last): File "/path/to/project/.venv/lib/python3.11/site-packages/pg8000/legacy.py", line 250, in execute self._context = self._c.execute_simple(operation) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/path/to/project/.venv/lib/python3.11/site-packages/pg8000/core.py", line 658, in execute_simple self.handle_messages(context) File "/path/to/project/.venv/lib/python3.11/site-packages/pg8000/core.py", line 807, in handle_messages raise context.error pg8000.exceptions.DatabaseError: {'S': 'ERROR', 'V': 'ERROR', 'C': '42P01', 'M': 'relation "View2" does not exist', 'P': '237', 'F': 'parse_relation.c', 'L': '1191', 'R': 'parserOpenTable'} During handling of the above exception, another exception occurred: Traceback (most recent call last): File "/path/to/project/.venv/lib/python3.11/site-packages/sqlalchemy/engine/base.py", line 1905, in _execute_context self.dialect.do_execute( File "/path/to/project/.venv/lib/python3.11/site-packages/sqlalchemy/engine/default.py", line 736, in do_execute cursor.execute(statement, parameters) File "/path/to/project/.venv/lib/python3.11/site-packages/pg8000/legacy.py", line 280, in execute raise cls(msg) pg8000.dbapi.ProgrammingError: {'S': 'ERROR', 'V': 'ERROR', 'C': '42P01', 'M': 'relation "View2" does not exist', 'P': '237', 'F': 'parse_relation.c', 'L': '1191', 'R': 'parserOpenTable'} The above exception was the direct cause of the following exception: Traceback (most recent call last): File "/path/to/project/.venv/bin/alembic", line 8, in sys.exit(main()) ^^^^^^ File "/path/to/project/.venv/lib/python3.11/site-packages/alembic/config.py", line 630, in main CommandLine(prog=prog).main(argv=argv) File "/path/to/project/.venv/lib/python3.11/site-packages/alembic/config.py", line 624, in main self.run_cmd(cfg, options) File "/path/to/project/.venv/lib/python3.11/site-packages/alembic/config.py", line 601, in run_cmd fn( File "/path/to/project/.venv/lib/python3.11/site-packages/alembic/command.py", line 233, in revision script_directory.run_env() File "/path/to/project/.venv/lib/python3.11/site-packages/alembic/script/base.py", line 578, in run_env util.load_python_file(self.dir, "env.py") File "/path/to/project/.venv/lib/python3.11/site-packages/alembic/util/pyfiles.py", line 93, in load_python_file module = load_module_py(module_id, path) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/path/to/project/.venv/lib/python3.11/site-packages/alembic/util/pyfiles.py", line 109, in load_module_py spec.loader.exec_module(module) # type: ignore ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "", line 940, in exec_module File "", line 241, in _call_with_frames_removed File "/path/to/project/migrations/env.py", line 129, in run_migrations_online() File "/path/to/project/migrations/env.py", line 123, in run_migrations_online context.run_migrations() File "", line 8, in run_migrations File "/path/to/project/.venv/lib/python3.11/site-packages/alembic/runtime/environment.py", line 922, in run_migrations self.get_context().run_migrations(**kw) File "/path/to/project/.venv/lib/python3.11/site-packages/alembic/runtime/migration.py", line 612, in run_migrations for step in self._migrations_fn(heads, self): ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/path/to/project/.venv/lib/python3.11/site-packages/alembic/command.py", line 209, in retrieve_migrations revision_context.run_autogenerate(rev, context) File "/path/to/project/.venv/lib/python3.11/site-packages/alembic/autogenerate/api.py", line 560, in run_autogenerate self._run_environment(rev, migration_context, True) File "/path/to/project/.venv/lib/python3.11/site-packages/alembic/autogenerate/api.py", line 607, in _run_environment compare._populate_migration_script( File "/path/to/project/.venv/lib/python3.11/site-packages/alembic/autogenerate/compare.py", line 59, in _populate_migration_script _produce_net_changes(autogen_context, upgrade_ops) File "/path/to/project/.venv/lib/python3.11/site-packages/alembic/autogenerate/compare.py", line 92, in _produce_net_changes comparators.dispatch("schema", autogen_context.dialect.name)( File "/path/to/project/.venv/lib/python3.11/site-packages/alembic/util/langhelpers.py", line 268, in go fn(*arg, **kw) File "/path/to/project/.venv/lib/python3.11/site-packages/alembic_utils/replaceable_entity.py", line 328, in compare_registered_entities maybe_op = entity.get_required_migration_op(sess, dependencies=has_create_or_update_op) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/path/to/project/.venv/lib/python3.11/site-packages/alembic_utils/replaceable_entity.py", line 163, in get_required_migration_op db_def = self.get_database_definition(sess, dependencies=dependencies) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/path/to/project/.venv/lib/python3.11/site-packages/alembic_utils/replaceable_entity.py", line 102, in get_database_definition with simulate_entity(sess, self, dependencies) as sess: File "/home/tim/.pyenv/versions/3.11.2/lib/python3.11/contextlib.py", line 137, in __enter__ return next(self.gen) ^^^^^^^^^^^^^^ File "/path/to/project/.venv/lib/python3.11/site-packages/alembic_utils/simulate.py", line 62, in simulate_entity sess.execute(entity.to_sql_statement_create()) File "/path/to/project/.venv/lib/python3.11/site-packages/sqlalchemy/orm/session.py", line 1717, in execute result = conn._execute_20(statement, params or {}, execution_options) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/path/to/project/.venv/lib/python3.11/site-packages/sqlalchemy/engine/base.py", line 1710, in _execute_20 return meth(self, args_10style, kwargs_10style, execution_options) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/path/to/project/.venv/lib/python3.11/site-packages/sqlalchemy/sql/elements.py", line 334, in _execute_on_connection return connection._execute_clauseelement( ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/path/to/project/.venv/lib/python3.11/site-packages/sqlalchemy/engine/base.py", line 1577, in _execute_clauseelement ret = self._execute_context( ^^^^^^^^^^^^^^^^^^^^^^ File "/path/to/project/.venv/lib/python3.11/site-packages/sqlalchemy/engine/base.py", line 1948, in _execute_context self._handle_dbapi_exception( File "/path/to/project/.venv/lib/python3.11/site-packages/sqlalchemy/engine/base.py", line 2129, in _handle_dbapi_exception util.raise_( File "/path/to/project/.venv/lib/python3.11/site-packages/sqlalchemy/util/compat.py", line 211, in raise_ raise exception File "/path/to/project/.venv/lib/python3.11/site-packages/sqlalchemy/engine/base.py", line 1905, in _execute_context self.dialect.do_execute( File "/path/to/project/.venv/lib/python3.11/site-packages/sqlalchemy/engine/default.py", line 736, in do_execute cursor.execute(statement, parameters) File "/path/to/project/.venv/lib/python3.11/site-packages/pg8000/legacy.py", line 280, in execute raise cls(msg) sqlalchemy.exc.ProgrammingError: (pg8000.dbapi.ProgrammingError) {'S': 'ERROR', 'V': 'ERROR', 'C': '42P01', 'M': 'relation "View2" does not exist', 'P': '237', 'F': 'parse_relation.c', 'L': '1191', 'R': 'parserOpenTable'} [SQL: CREATE VIEW "public"."View3" AS SELECT "View2".id, "View2".field_1, "View2".field_2 FROM "View2" JOIN "Table1" ON "View2".table1_id = "Table1".id GROUP BY "View2".field_3;] (Background on this error at: https://sqlalche.me/e/14/f405) ```

If I remove View3 from the registered entities, the migration succeeds and a drop op is created in addition to the replace op for View1, which is expected.

[olirice]

Reproducible test case

import pytest
from sqlalchemy.exc import ProgrammingError

from alembic_utils.exceptions import SQLParseFailure
from alembic_utils.pg_view import PGView
from alembic_utils.replaceable_entity import register_entities
from alembic_utils.testbase import TEST_VERSIONS_ROOT, run_alembic_command

def test_update_nested_view_revision(engine) -> None:
    TEST_VIEW_1 = PGView(schema="public", signature="view1", definition="select 1 as one")
    TEST_VIEW_2 = PGView(
        schema="public", signature="view2", definition="select one from public.view1"
    )
    TEST_VIEW_3 = PGView(
        schema="public", signature="view3", definition="select one from public.view2"
    )

    # Create the view outside of a revision
    with engine.begin() as connection:
        connection.execute(TEST_VIEW_1.to_sql_statement_create())
        connection.execute(TEST_VIEW_2.to_sql_statement_create())
        connection.execute(TEST_VIEW_3.to_sql_statement_create())

    # Update definition of TO_UPPER
    UPDATED_TEST_VIEW_1 = PGView(TEST_VIEW_1.schema, TEST_VIEW_1.signature, "select 2 as one")

    register_entities([UPDATED_TEST_VIEW_1, TEST_VIEW_2, TEST_VIEW_3], entity_types=[PGView])

    # Autogenerate a new migration
    # It should detect the change we made and produce a "replace_function" statement
    output = run_alembic_command(
        engine=engine,
        command="revision",
        command_kwargs={"autogenerate": True, "rev_id": "2", "message": "replace"},
    )

    migration_replace_path = TEST_VERSIONS_ROOT / "2_replace.py"

    with migration_replace_path.open() as migration_file:
        import pdb

        pdb.set_trace()
        migration_contents = migration_file.read()

    assert "op.replace_entity" in migration_contents
    assert "op.create_entity" not in migration_contents
    assert "op.drop_entity" not in migration_contents
    assert "from alembic_utils.pg_view import PGView" in migration_contents

    # Execute upgrade
    run_alembic_command(engine=engine, command="upgrade", command_kwargs={"revision": "head"})
    # Execute Downgrade
    run_alembic_command(engine=engine, command="downgrade", command_kwargs={"revision": "base"})

Output

=================================== FAILURES ===================================
_______________________ test_update_nested_view_revision _______________________

self = <sqlalchemy.engine.base.Connection object at 0x1079755a0>
dialect = <sqlalchemy.dialects.postgresql.psycopg2.PGDialect_psycopg2 object at 0x1079f16f0>
constructor = <bound method DefaultExecutionContext._init_compiled of <class 'sqlalchemy.dialects.postgresql.psycopg2.PGExecutionContext_psycopg2'>>
statement = 'DROP VIEW "public"."view3" cascade', parameters = {}
execution_options = immutabledict({'autocommit': symbol('PARSE_AUTOCOMMIT'), 'future_result': True})
args = (<sqlalchemy.dialects.postgresql.psycopg2.PGCompiler_psycopg2 object at 0x107999510>, [{}], <sqlalchemy.sql.elements.TextClause object at 0x107a04fd0>, [])
kw = {'cache_hit': symbol('CACHE_HIT')}
branched = <sqlalchemy.engine.base.Connection object at 0x1079755a0>
conn = <sqlalchemy.pool.base._ConnectionFairy object at 0x1079f0fa0>
context = <sqlalchemy.dialects.postgresql.psycopg2.PGExecutionContext_psycopg2 object at 0x107a056c0>
cursor = <cursor object at 0x107595d50; closed: -1>, evt_handled = False

Trace

src/test/test_pg_view_deep.py:32: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
src/alembic_utils/testbase.py:48: in run_alembic_command
    command_func(alembic_cfg, **command_kwargs)
venv/lib/python3.10/site-packages/alembic/command.py:236: in revision
    script_directory.run_env()
venv/lib/python3.10/site-packages/alembic/script/base.py:582: in run_env
    util.load_python_file(self.dir, "env.py")
venv/lib/python3.10/site-packages/alembic/util/pyfiles.py:94: in load_python_file
    module = load_module_py(module_id, path)
venv/lib/python3.10/site-packages/alembic/util/pyfiles.py:110: in load_module_py
    spec.loader.exec_module(module)  # type: ignore
<frozen importlib._bootstrap_external>:883: in exec_module
    ???
<frozen importlib._bootstrap>:241: in _call_with_frames_removed
    ???
src/test/alembic_config/env.py:75: in <module>
    run_migrations_online()
src/test/alembic_config/env.py:72: in run_migrations_online
    context.run_migrations()
<string>:8: in run_migrations
    ???
venv/lib/python3.10/site-packages/alembic/runtime/environment.py:928: in run_migrations
    self.get_context().run_migrations(**kw)
venv/lib/python3.10/site-packages/alembic/runtime/migration.py:615: in run_migrations
    for step in self._migrations_fn(heads, self):
venv/lib/python3.10/site-packages/alembic/command.py:212: in retrieve_migrations
    revision_context.run_autogenerate(rev, context)
venv/lib/python3.10/site-packages/alembic/autogenerate/api.py:562: in run_autogenerate
    self._run_environment(rev, migration_context, True)
venv/lib/python3.10/site-packages/alembic/autogenerate/api.py:609: in _run_environment
    compare._populate_migration_script(
venv/lib/python3.10/site-packages/alembic/autogenerate/compare.py:59: in _populate_migration_script
    _produce_net_changes(autogen_context, upgrade_ops)
venv/lib/python3.10/site-packages/alembic/autogenerate/compare.py:93: in _produce_net_changes
    comparators.dispatch("schema", autogen_context.dialect.name)(
venv/lib/python3.10/site-packages/alembic/util/langhelpers.py:269: in go
    fn(*arg, **kw)
src/alembic_utils/replaceable_entity.py:328: in compare_registered_entities
    maybe_op = entity.get_required_migration_op(sess, dependencies=has_create_or_update_op)
src/alembic_utils/replaceable_entity.py:164: in get_required_migration_op
    db_def = self.get_database_definition(sess, dependencies=dependencies)
src/alembic_utils/replaceable_entity.py:102: in get_database_definition
    with simulate_entity(sess, self, dependencies) as sess:
../../../.pyenv/versions/3.10.3/Library/Frameworks/Python.framework/Versions/3.10/lib/python3.10/contextlib.py:135: in __enter__
    return next(self.gen)
src/alembic_utils/simulate.py:62: in simulate_entity
    sess.execute(entity.to_sql_statement_create())
venv/lib/python3.10/site-packages/sqlalchemy/orm/session.py:1653: in execute
    result = conn._execute_20(statement, params or {}, execution_options)
venv/lib/python3.10/site-packages/sqlalchemy/engine/base.py:1520: in _execute_20
    return meth(self, args_10style, kwargs_10style, execution_options)
venv/lib/python3.10/site-packages/sqlalchemy/sql/elements.py:313: in _execute_on_connection
    return connection._execute_clauseelement(
venv/lib/python3.10/site-packages/sqlalchemy/engine/base.py:1389: in _execute_clauseelement
    ret = self._execute_context(
venv/lib/python3.10/site-packages/sqlalchemy/engine/base.py:1748: in _execute_context
    self._handle_dbapi_exception(
venv/lib/python3.10/site-packages/sqlalchemy/engine/base.py:1929: in _handle_dbapi_exception
    util.raise_(
venv/lib/python3.10/site-packages/sqlalchemy/util/compat.py:198: in raise_
    raise exception
venv/lib/python3.10/site-packages/sqlalchemy/engine/base.py:1705: in _execute_context
    self.dialect.do_execute(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <sqlalchemy.dialects.postgresql.psycopg2.PGDialect_psycopg2 object at 0x1079f16f0>
cursor = <cursor object at 0x107595e40; closed: -1>
statement = 'CREATE VIEW "public"."view3" AS select one from public.view2;'
parameters = {}
context = <sqlalchemy.dialects.postgresql.psycopg2.PGExecutionContext_psycopg2 object at 0x107a05450>

    def do_execute(self, cursor, statement, parameters, context=None):
>       cursor.execute(statement, parameters)
E       sqlalchemy.exc.ProgrammingError: (psycopg2.errors.UndefinedTable) relation "public.view2" does not exist
E       LINE 1: CREATE VIEW "public"."view3" AS select one from public.view2...
E                                                               ^
E       
E       [SQL: CREATE VIEW "public"."view3" AS select one from public.view2;]
E       (Background on this error at: http://sqlalche.me/e/14/f405)

venv/lib/python3.10/site-packages/sqlalchemy/engine/default.py:681: ProgrammingError

I had a look and transitive dependencies are not detected during dependency resolution.

I'd suggest resolving that migration manually using the steps from my previous comment

[trogers-medasync]

Would it be a difficult change to update the dependency resolution mechanism to identify transitive dependencies? I could attempt that change if you're open to PRs?

[olirice]

I'm definitely open to PRs but it is a very complex problem to improve dependency resolution without making resolution time balloon out of control

Here are good places to review

https://github.com/olirice/alembic_utils/blob/ed303f52635b9fee6fdb3b9abdddfa38bed1bcb4/src/alembic_utils/depends.py#L13-L55

https://github.com/olirice/alembic_utils/blob/ed303f52635b9fee6fdb3b9abdddfa38bed1bcb4/src/alembic_utils/replaceable_entity.py#L291-L353

https://github.com/olirice/alembic_utils/blob/ed303f52635b9fee6fdb3b9abdddfa38bed1bcb4/src/alembic_utils/replaceable_entity.py#L98-L122

https://github.com/olirice/alembic_utils/blob/ed303f52635b9fee6fdb3b9abdddfa38bed1bcb4/src/alembic_utils/replaceable_entity.py#L156-L178