Handling PostgreSQL Enum Updates with View Dependencies in Alembic Migrations

When working with PostgreSQL, updating an ENUM type can be challenging, especially when dependent views rely on the column using the enum. A common error that arises in such cases is:

sqlalchemy.exc.NotSupportedError: (psycopg2.errors.FeatureNotSupported) cannot alter type of a column used by a view or rule
DETAIL:  rule _RETURN on view custom_table_view depends on column "custom_enum_type"

Meaning that in PostgreSQL, a column’s type cannot be modified if it is used by a view. The simplest solution is to drop and recreate the view after modifying the enum. However, manually copying and pasting a complex view definition is not an elegant approach.

A solution I came up with uses PostgreSQL build-in functions instead of manually handling view recreation. The approach involves automating the process using SQLAlchemy and Alembic migrations. Below is an optimized migration script that:

1. Identifies all views that depend on the ENUM.
2. Retrieves their definitions before dropping them.
3. Updates the ENUM type.
4. Restores the views using the stored definitions.

def get_tables_with_enum(conn, typename):
    """Fetches all views that depend on a given enum type."""
    result = conn.execute(
        sa.text(
            """
            SELECT DISTINCT c.table_name
            FROM information_schema.columns c
            JOIN pg_type t ON c.udt_name = t.typname
            JOIN information_schema.views v ON c.table_name = v.table_name
            WHERE t.typname = :typename;
            """
        ),
        {"typename": typename},
    )
    return [row[0] for row in result]

def upgrade() -> None:
    rules = {}
    conn = op.get_bind()

    # Get all views that depend on the enum type
    table_names = get_tables_with_enum(conn, "custom_enum_type")
    for view_name in table_names:
        rules[view_name] = conn.execute(sa.text(f"SELECT pg_get_viewdef('{view_name}', true)")).scalar()

        if not rules[view_name]:
            raise RuntimeError(f"Failed to retrieve definition for view: {view_name}")

        # Drop the view before modifying the enum
        op.execute(f"DROP VIEW {view_name}")

    # Update the enum type
    op.sync_enum_values(
        enum_schema="public",
        enum_name="custom_enum_type",
        new_values=["NEW", "VALUE", "ADDED"],
        affected_columns=[
            TableReference(table_schema="public", table_name="another_custom_table", column_name="custom_enum_type"),
            TableReference(table_schema="public", table_name="custom_table", column_name="custom_enum_type"),
        ],
        enum_values_to_rename=[],
    )

    # Recreate the dropped views
    for view, query in rules.items():
        op.execute(sa.text(f"CREATE OR REPLACE VIEW {view} AS {query}"))

This uses pg_get_viewdef to fetch the current definition of views before dropping them. Next I use alembic-postgresql-enum to update the enum safely. But in theory you can also use plain SQL here. Finally it recreates the views using their original definitions, ensuring no manual intervention is required.

This approach ensures a seamless migration process, even for complex views, eliminating the need for manual copy-pasting and reducing the risk of errors. 🚀