Create table and dependent view on same Alembic migration

[fernandoandreotti]

Hi there,

First of all, great work on this library. Really convenient for creating views/materialised views.

The problem I am facing is the same as this one described in the alembic_utilspackage:
olirice/alembic_utils#79

Specifically, I would like to, for each new table I create, generate an accompanying view. And for that I would like migrations to be able to introspect and find the dependencies from view->table. This because my tables are append-only, and I would like to have a latest view of those.

class Foo(Base):
    __tablename__ = 'foo'
    id = Column(types.Integer(), primary_key=True)


@view(Base)
class HighFoo:
    __tablename__ = "high_foo"
    __view__ = select(Foo.__table__).where(Foo.__table__.c.id >= 10)

Then with register_alembic_events I register those in my alembic env.py

Is this something solvable with the approach you've taken in this package? Am I missing something on my end?

[DanCardin]

I'm guessing this lib will largely have similar issues, generally, to alembic_utils for certain kinds of entities. Views in particular are difficult because postgres will internally normalize the SQL you give it, which makes actually attempting to create the view (and roll back the creation) somewhat necessary in many cases.

With that said, I think for this usecase, I can probably help. For a pure creation, we know it's being created ahead of time and dont need the complex live-database comparison. I suspect that it's really only for updates that it becomes important

[fernandoandreotti]

That would be awesome! Let me know if you need help testing.

[DanCardin]

Sure, if you want/are able to test out dc/create-table-and-view before I merge it, to make sure it addresses your particular case (my test case just copies your failing example above), feel free.

I'll wait a few hours, or until you respond affirmative and merge it anyways :P

[fernandoandreotti]

Amazing! Many thanks for the speedy response. 🚀

I did test it after pip installed this branch straight from github. On the first attempt I got a dependency missing sqlglot. Stack trace below. Wondering if this is because I just pip installed from GH.

After installing this package migration worked without any issues! 🥳

Here is an edited version of the migration:

op.create_table('foo',
    sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
    sa.Column('created_at', sa.DateTime(), nullable=False),
    sa.PrimaryKeyConstraint('id', name=op.f('pk_foo')),
    schema='test'
    )
op.create_index(op.f('ix_foo_id'), 'foo', ['id'], unique=False, schema='test') # I had this index on my Mixin table_args
op.execute("""CREATE VIEW high_foo AS SELECT [test.foo.id](http://test.foo.id/), test.foo.created_at FROM test.foo WHERE [test.foo.id](http://test.foo.id/) >= 10;""")
op.execute("""DROP VIEW <ALL MY CURRENT PGView/PGMaterializedViews ARE DROPPED>""") 

⚠️ The last row is dropping all my PGView / PGMaterialized views. I guess this is fine as one would pick one or the other library

Stack trace from first run of running pytest_alembic:

_____________________________________________________________________________________ test_model_definitions_match_ddl _____________________________________________________________________________________

self = View(name='high_foo', definition=<sqlalchemy.sql.selectable.Select object at 0x12ecf9f10>, schema=None, materialized=False, constraints=None)
conn = <sqlalchemy.engine.base.Connection object at 0x12f42e4f0>, using_connection = False

    def render_definition(self, conn: Connection, using_connection: bool = True):
        dialect = conn.engine.dialect
   
        compiled_definition = self.compile_definition(dialect)
   
        if using_connection and [dialect.name](http://dialect.name/) == "postgresql":
            from sqlalchemy_declarative_extensions.dialects import get_view
   
            with conn.begin_nested() as trans:
                try:
                    random_name = "v" + uuid.uuid4().hex
                    conn.execute(
                        text(f"CREATE VIEW {random_name} AS {compiled_definition}")
                    )
                    view = get_view(conn, random_name)
                    definition1 = view.definition
   
                    # Optimization, the view query **can** change if we re-run it,
                    # but if it's not changed from the first iteration, we assume it wont.
                    if definition1 == compiled_definition:
                        return escape_params(compiled_definition)
   
                    # Re-generate the view, it **can** not produce the same text twice.
                    random_name = "v" + uuid.uuid4().hex
                    conn.execute(text(f"CREATE VIEW {random_name} AS {definition1}"))
                    view = get_view(conn, random_name)
                    definition2 = view.definition
                    return escape_params(definition2)
                finally:
                    trans.rollback()
        else:
            # Fall back to library-based normalization, which cannot be perfect.
            try:
>               import sqlglot
E               ModuleNotFoundError: No module named 'sqlglot'

.venv/lib/python3.9/site-packages/sqlalchemy_declarative_extensions/view/base.py:208: ModuleNotFoundError

During handling of the above exception, another exception occurred:

alembic_runner = MigrationContext(command_executor=CommandExecutor(alembic_config=<alembic.config.Config object at 0x12f3ba850>, stdout...c_config=None, before_revision_data=None, at_revision_data=None, minimum_downgrade_revision=None, skip_revisions=None))

    @pytest.mark.alembic()
    def test_model_definitions_match_ddl(alembic_runner):
        """Assert that the state of the migrations matches the state of the models describing the DDL.
   
        In general, the set of migrations in the history should coalesce into DDL which is described
        by the current set of models. Therefore, a call to `revision --autogenerate` should always
        generate an empty migration (e.g. find no difference between your database (i.e. migrations
        history) and your models).
        """
   
        def verify_is_empty_revision(migration_context, __, directives):
            script = directives[0]
   
            migration_is_empty = script.upgrade_ops.is_empty()
            if not migration_is_empty:
                autogen_context = AutogenContext(migration_context)
                rendered_upgrade = _render_cmd_body(script.upgrade_ops, autogen_context)
   
                if not migration_is_empty:
                    message = (
                        "The models describing the DDL of your database are out of sync with the set of "
                        "steps described in the revision history. This usually means that someone has "
                        "made manual changes to the database's DDL, or some model has been changed "
                        "without also generating a migration to describe that change."
                    )
                    raise AlembicTestFailure(
                        message,
                        context=[
                            (
                                "The upgrade which would have been generated would look like",
                                rendered_upgrade,
                            )
                        ],
                    )
   
        test_upgrade(alembic_runner)
>       alembic_runner.generate_revision(
            message="test revision",
            autogenerate=True,
            prevent_file_generation=True,
            process_revision_directives=verify_is_empty_revision,
        )

.venv/lib/python3.9/site-packages/pytest_alembic/tests/default.py:93:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.venv/lib/python3.9/site-packages/pytest_alembic/runner.py:124: in generate_revision
    result = self.command_executor.run_command(
.venv/lib/python3.9/site-packages/pytest_alembic/executor.py:52: in run_command
    executable_command(self.alembic_config, *args, **kwargs)
.venv/lib/python3.9/site-packages/alembic/command.py:236: in revision
    script_directory.run_env()
.venv/lib/python3.9/site-packages/alembic/script/base.py:583: in run_env
    util.load_python_file(self.dir, "env.py")
.venv/lib/python3.9/site-packages/alembic/util/pyfiles.py:95: in load_python_file
    module = load_module_py(module_id, path)
.venv/lib/python3.9/site-packages/alembic/util/pyfiles.py:113: in load_module_py
    spec.loader.exec_module(module)  # type: ignore
<frozen importlib._bootstrap_external>:850: in exec_module
    ???
<frozen importlib._bootstrap>:228: in _call_with_frames_removed
    ???
src/metadata_db_schema/env.py:122: in <module>
    run_migrations_online()
src/metadata_db_schema/env.py:116: in run_migrations_online
    context.run_migrations()
<string>:8: in run_migrations
    ???
.venv/lib/python3.9/site-packages/alembic/runtime/environment.py:948: in run_migrations
    self.get_context().run_migrations(**kw)
.venv/lib/python3.9/site-packages/alembic/runtime/migration.py:615: in run_migrations
    for step in self._migrations_fn(heads, self):
.venv/lib/python3.9/site-packages/alembic/command.py:212: in retrieve_migrations
    revision_context.run_autogenerate(rev, context)
.venv/lib/python3.9/site-packages/alembic/autogenerate/api.py:570: in run_autogenerate
    self._run_environment(rev, migration_context, True)
.venv/lib/python3.9/site-packages/alembic/autogenerate/api.py:617: in _run_environment
    compare._populate_migration_script(
.venv/lib/python3.9/site-packages/alembic/autogenerate/compare.py:65: in _populate_migration_script
    _produce_net_changes(autogen_context, upgrade_ops)
.venv/lib/python3.9/site-packages/alembic/autogenerate/compare.py:98: in _produce_net_changes
    comparators.dispatch("schema", [autogen_context.dialect.name](http://autogen_context.dialect.name/))(
.venv/lib/python3.9/site-packages/alembic/util/langhelpers.py:313: in go
    fn(*arg, **kw)
.venv/lib/python3.9/site-packages/sqlalchemy_declarative_extensions/alembic/view.py:20: in _compare_views
    result = compare_views(autogen_context.connection, views, metadata)
.venv/lib/python3.9/site-packages/sqlalchemy_declarative_extensions/view/compare.py:81: in compare_views
    normalized_view = view.normalize(
.venv/lib/python3.9/site-packages/sqlalchemy_declarative_extensions/view/base.py:252: in normalize
    definition=self.render_definition(conn, using_connection=using_connection),
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = View(name='high_foo', definition=<sqlalchemy.sql.selectable.Select object at 0x12ecf9f10>, schema=None, materialized=False, constraints=None)
conn = <sqlalchemy.engine.base.Connection object at 0x12f42e4f0>, using_connection = False

    def render_definition(self, conn: Connection, using_connection: bool = True):
        dialect = conn.engine.dialect
   
        compiled_definition = self.compile_definition(dialect)
   
        if using_connection and [dialect.name](http://dialect.name/) == "postgresql":
            from sqlalchemy_declarative_extensions.dialects import get_view
   
            with conn.begin_nested() as trans:
                try:
                    random_name = "v" + uuid.uuid4().hex
                    conn.execute(
                        text(f"CREATE VIEW {random_name} AS {compiled_definition}")
                    )
                    view = get_view(conn, random_name)
                    definition1 = view.definition
   
                    # Optimization, the view query **can** change if we re-run it,
                    # but if it's not changed from the first iteration, we assume it wont.
                    if definition1 == compiled_definition:
                        return escape_params(compiled_definition)
   
                    # Re-generate the view, it **can** not produce the same text twice.
                    random_name = "v" + uuid.uuid4().hex
                    conn.execute(text(f"CREATE VIEW {random_name} AS {definition1}"))
                    view = get_view(conn, random_name)
                    definition2 = view.definition
                    return escape_params(definition2)
                finally:
                    trans.rollback()
        else:
            # Fall back to library-based normalization, which cannot be perfect.
            try:
                import sqlglot
                from sqlglot.optimizer.normalize import normalize
            except ImportError:  # pragma: no cover
>               raise ImportError("View autogeneration requires the 'parse' extra.")
E               ImportError: View autogeneration requires the 'parse' extra.

[fernandoandreotti]

Ah yes, the dep is on the poetry lock. So probably something on my setup.

[DanCardin]

sqlglot is behind the parse extra, fwiw. Although i'm not sure how to specify extras on git installations. But shouldn't be an issue depending on it directly.

Also, fwiw, you can register alembic_utils views in the same way you do normal ones, as a compatibility layer: https://github.com/DanCardin/sqlalchemy-declarative-extensions/blob/main/tests/view/test_alembic_utils.py

I originally wrote this library for...a variety of reasons, but the fact that alembic_utils embeds its own imports into your migrations makes it quite difficult to remove after the fact, so it seemed prudent to include this ^, as a measure against all your views being force dropped on you.

Although with that said, I'm not sure I can guarantee all the same things through that layer, so ymmv.

[DanCardin]

Should be released in 0.7.2 momentarily!

[fernandoandreotti]

Awesome, many thanks for the PR and explanation :)
Also agree on the problematic imports on migration, will check on the registration a bit further tomorrow, see if I am missing something.