alembic-utils

import sys from alembic_utils.pg_view import PGView from alembic_utils.pg_function import PGFunction # Mock op object for demonstration purposes to show API usage. # In a real Alembic migration, `op` is provided by Alembic. class MockOp: def execute(self, statement): print(f"Executing DDL: {statement}") def create_entity(self, entity): print(f"Creating entity: {entity.entity_name}") print(f" Definition: {entity.to_sql_statement()}") def drop_entity(self, entity): print(f"Dropping entity: {entity.entity_name}") print(f" Definition: {entity.to_sql_drop_statement()}") def sync_alembic_utils(self): print("Synchronizing Alembic-Utils entities (mocked).") op = MockOp() # Define a PostgreSQL view my_view = PGView( schema="public", signature="my_first_view", definition="SELECT id, name FROM users WHERE active = TRUE;" ) # Define a PostgreSQL function my_function = PGFunction( schema="public", signature="my_first_function(a integer, b integer)", definition="RETURNS integer LANGUAGE plpgsql IMMUTABLE AS $$ BEGIN RETURN a + b; END; $$", returns="integer" # Required since 0.6.0 ) # Simulate usage in an Alembic migration's upgrade() function print("--- Simulating 'upgrade' phase ---") op.create_entity(my_view) op.create_entity(my_function) # For Alembic >= 1.10.0 and Alembic-Utils >= 0.8.0, use this in env.py # to automatically discover and manage entities defined in your model. # If you define entities directly in migration scripts, this might not be needed. op.sync_alembic_utils() print("\n--- Simulating 'downgrade' phase ---") # Entities should be dropped in reverse order of creation if dependencies exist op.drop_entity(my_function) op.drop_entity(my_view)