I'm having issues autogenerating a migration. I have a Materialized

Reproducible test case <div class="highlight highlight-source-python notranslate p

Materialized view change detection fails if upstream view has changed,about olirice/alembic_utils

Comments (6)

olirice commented on September 27, 2024

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

from alembic_utils.

trogers-medasync commented on September 27, 2024

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

from alembic_utils.

trogers-medasync commented on September 27, 2024

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 <module>
    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 "<frozen importlib._bootstrap_external>", line 940, in exec_module
  File "<frozen importlib._bootstrap>", line 241, in _call_with_frames_removed
  File "/path/to/project/migrations/env.py", line 129, in <module>
    run_migrations_online()
  File "/path/to/project/migrations/env.py", line 123, in run_migrations_online
    context.run_migrations()
  File "<string>", 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.

from alembic_utils.

olirice commented on September 27, 2024

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

from alembic_utils.

trogers-medasync commented on September 27, 2024

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?

from alembic_utils.

olirice commented on September 27, 2024

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

alembic_utils/src/alembic_utils/depends.py

Lines 13 to 55 in ed303f5

 def solve_resolution_order(sess: Session, entities): 

 """Solve for an entity resolution order that increases the probability that 

  a migration will suceed if, for example, two new views are created and one 

  refers to the other 

  This strategy will only solve for simple cases 

  """ 

 resolved = [] 

 # Resolve the entities with 0 dependencies first (faster) 

 logger.info("Resolving entities with no dependencies") 

 for entity in entities: 

 try: 

 with simulate_entity(sess, entity): 

 resolved.append(entity) 

 except (sqla_exc.ProgrammingError, sqla_exc.InternalError) as exc: 

 continue 

 # Resolve entities with possible dependencies 

 logger.info("Resolving entities with dependencies. This may take a minute") 

 for _ in range(len(entities)): 

 n_resolved = len(resolved) 

 for entity in entities: 

 if entity in resolved: 

 continue 

 try: 

 with simulate_entity(sess, entity, dependencies=resolved): 

 resolved.append(entity) 

 except (sqla_exc.ProgrammingError, sqla_exc.InternalError): 

 continue 

 if len(resolved) == n_resolved: 

 # No new entities resolved in the last iteration. Exit 

 break 

 for entity in entities: 

 if entity not in resolved: 

 resolved.append(entity) 

 return resolved

alembic_utils/src/alembic_utils/replaceable_entity.py

Lines 291 to 353 in ed303f5

 # Solve resolution order 

 transaction = connection.begin_nested() 

 sess = Session(bind=connection) 

 try: 

 ordered_entities: List[ReplaceableEntity] = solve_resolution_order(sess, entities) 

 finally: 

 sess.rollback() 

 # entities that are receiving a create or update op 

 has_create_or_update_op: List[ReplaceableEntity] = [] 

 # database rendered definitions for the entities we have a local instance for 

 # Note: used for drops 

 local_entities = [] 

 # Required migration OPs, Create/Update/NoOp 

 for entity in ordered_entities: 

 logger.info( 

 "Detecting required migration op %s %s", 

 entity.__class__.__name__, 

 entity.identity, 

 ) 

 if entity.__class__ not in registry.allowed_entity_types: 

 continue 

 if not include_entity(entity, autogen_context, reflected=False): 

 logger.debug( 

 "Ignoring local entity %s %s due to AutogenContext filters", 

 entity.__class__.__name__, 

 entity.identity, 

 ) 

 continue 

 transaction = connection.begin_nested() 

 sess = Session(bind=connection) 

 try: 

 maybe_op = entity.get_required_migration_op(sess, dependencies=has_create_or_update_op) 

 local_db_def = entity.get_database_definition( 

 sess, dependencies=has_create_or_update_op 

 ) 

 local_entities.append(local_db_def) 

 if maybe_op: 

 upgrade_ops.ops.append(maybe_op) 

 has_create_or_update_op.append(entity) 

 logger.info( 

 "Detected %s op for %s %s", 

 maybe_op.__class__.__name__, 

 entity.__class__.__name__, 

 entity.identity, 

 ) 

 else: 

 logger.debug( 

 "Detected NoOp op for %s %s", 

 entity.__class__.__name__, 

 entity.identity, 

 ) 

 finally: 

 sess.rollback()

alembic_utils/src/alembic_utils/replaceable_entity.py

Lines 98 to 122 in ed303f5

 def get_database_definition( 

 self: T, sess: Session, dependencies: Optional[List["ReplaceableEntity"]] = None 

 ) -> T: # $Optional[T]: 

 """Creates the entity in the database, retrieves its 'rendered' then rolls it back""" 

 with simulate_entity(sess, self, dependencies) as sess: 

 # Drop self 

 sess.execute(self.to_sql_statement_drop()) 

 # collect all remaining entities 

 db_entities: List[T] = sorted( 

 self.from_database(sess, schema=self.schema), key=lambda x: x.identity 

 ) 

 with simulate_entity(sess, self, dependencies) as sess: 

 # collect all remaining entities 

 all_w_self: List[T] = sorted( 

 self.from_database(sess, schema=self.schema), key=lambda x: x.identity 

 ) 

 # Find "self" by diffing the before and after 

 for without_self, with_self in zip_longest(db_entities, all_w_self): 

 if without_self is None or without_self.identity != with_self.identity: 

 return with_self 

 raise UnreachableException()

alembic_utils/src/alembic_utils/replaceable_entity.py

Lines 156 to 178 in ed303f5

 def get_required_migration_op( 

 self: T, sess: Session, dependencies: Optional[List["ReplaceableEntity"]] = None 

 ) -> Optional[ReversibleOp]: 

 """Get the migration operation required for autogenerate""" 

 # All entities in the database for self's schema 

 entities_in_database: List[T] = self.from_database(sess, schema=self.schema) 

 db_def = self.get_database_definition(sess, dependencies=dependencies) 

 for x in entities_in_database: 

 if (db_def.identity, normalize_whitespace(db_def.definition)) == ( 

 x.identity, 

 normalize_whitespace(x.definition), 

 ): 

 return None 

 if db_def.identity == x.identity: 

 # Cache the currently live copy to render a RevertOp without hitting DB again 

 self._version_to_replace = x 

 return ReplaceOp(self) 

 return CreateOp(self)

from alembic_utils.

Materialized view change detection fails if upstream view has changed about alembic_utils HOT 6 OPEN

Comments (6)

Related Issues (20)

Recommend Projects

React

Vue.js

Typescript

TensorFlow

Django

Laravel

D3

Recommend Topics

javascript

web

server

Machine learning

Visualization

Game

Recommend Org

Facebook

Microsoft

Google

Alibaba

D3

Tencent

	def solve_resolution_order(sess: Session, entities):
	"""Solve for an entity resolution order that increases the probability that
	a migration will suceed if, for example, two new views are created and one
	refers to the other

	This strategy will only solve for simple cases
	"""

	resolved = []

	# Resolve the entities with 0 dependencies first (faster)
	logger.info("Resolving entities with no dependencies")
	for entity in entities:
	try:
	with simulate_entity(sess, entity):
	resolved.append(entity)
	except (sqla_exc.ProgrammingError, sqla_exc.InternalError) as exc:
	continue

	# Resolve entities with possible dependencies
	logger.info("Resolving entities with dependencies. This may take a minute")
	for _ in range(len(entities)):
	n_resolved = len(resolved)

	for entity in entities:
	if entity in resolved:
	continue

	try:
	with simulate_entity(sess, entity, dependencies=resolved):
	resolved.append(entity)
	except (sqla_exc.ProgrammingError, sqla_exc.InternalError):
	continue

	if len(resolved) == n_resolved:
	# No new entities resolved in the last iteration. Exit
	break

	for entity in entities:
	if entity not in resolved:
	resolved.append(entity)

	return resolved

	# Solve resolution order
	transaction = connection.begin_nested()
	sess = Session(bind=connection)
	try:
	ordered_entities: List[ReplaceableEntity] = solve_resolution_order(sess, entities)
	finally:
	sess.rollback()

	# entities that are receiving a create or update op
	has_create_or_update_op: List[ReplaceableEntity] = []

	# database rendered definitions for the entities we have a local instance for
	# Note: used for drops
	local_entities = []

	# Required migration OPs, Create/Update/NoOp
	for entity in ordered_entities:
	logger.info(
	"Detecting required migration op %s %s",
	entity.__class__.__name__,
	entity.identity,
	)

	if entity.__class__ not in registry.allowed_entity_types:
	continue

	if not include_entity(entity, autogen_context, reflected=False):
	logger.debug(
	"Ignoring local entity %s %s due to AutogenContext filters",
	entity.__class__.__name__,
	entity.identity,
	)
	continue

	transaction = connection.begin_nested()
	sess = Session(bind=connection)
	try:
	maybe_op = entity.get_required_migration_op(sess, dependencies=has_create_or_update_op)

	local_db_def = entity.get_database_definition(
	sess, dependencies=has_create_or_update_op
	)
	local_entities.append(local_db_def)

	if maybe_op:
	upgrade_ops.ops.append(maybe_op)
	has_create_or_update_op.append(entity)

	logger.info(
	"Detected %s op for %s %s",
	maybe_op.__class__.__name__,
	entity.__class__.__name__,
	entity.identity,
	)
	else:
	logger.debug(
	"Detected NoOp op for %s %s",
	entity.__class__.__name__,
	entity.identity,
	)

	finally:
	sess.rollback()

	def get_database_definition(
	self: T, sess: Session, dependencies: Optional[List["ReplaceableEntity"]] = None
	) -> T: # $Optional[T]:
	"""Creates the entity in the database, retrieves its 'rendered' then rolls it back"""
	with simulate_entity(sess, self, dependencies) as sess:
	# Drop self
	sess.execute(self.to_sql_statement_drop())

	# collect all remaining entities
	db_entities: List[T] = sorted(
	self.from_database(sess, schema=self.schema), key=lambda x: x.identity
	)

	with simulate_entity(sess, self, dependencies) as sess:
	# collect all remaining entities
	all_w_self: List[T] = sorted(
	self.from_database(sess, schema=self.schema), key=lambda x: x.identity
	)

	# Find "self" by diffing the before and after
	for without_self, with_self in zip_longest(db_entities, all_w_self):
	if without_self is None or without_self.identity != with_self.identity:
	return with_self

	raise UnreachableException()

	def get_required_migration_op(
	self: T, sess: Session, dependencies: Optional[List["ReplaceableEntity"]] = None
	) -> Optional[ReversibleOp]:
	"""Get the migration operation required for autogenerate"""
	# All entities in the database for self's schema
	entities_in_database: List[T] = self.from_database(sess, schema=self.schema)

	db_def = self.get_database_definition(sess, dependencies=dependencies)

	for x in entities_in_database:

	if (db_def.identity, normalize_whitespace(db_def.definition)) == (
	x.identity,
	normalize_whitespace(x.definition),
	):
	return None

	if db_def.identity == x.identity:
	# Cache the currently live copy to render a RevertOp without hitting DB again
	self._version_to_replace = x
	return ReplaceOp(self)

	return CreateOp(self)