alembic_utils create_entity if not exists

Similar to alembic create_table, check if table exists However this question is for creating entities like extensions. The solution for tables using reflection does not appear like it would work for extensions as I don’t see a way to get the list of existing installed extensions through that https://docs.sqlalchemy.org/en/14/core/reflection.html

Going to see if I can run a query to get that list on the connection. Any other ideas? Would be great if the create_entity had an if_not_exists option to only do it if it doesn’t already exist.

My use case here involves creating a new schema via alembic in a database with schemas created through Liquibase. We are moving from that tool to this one. Eventually dropping the old schemas altogether etc. For existing databases the uuid-ossp extension will already be there. For new databases where we just have the new alembic based schema it will not be there. Needs to work in both cases.

Example:

def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    public_uuid_ossp = PGExtension(
        schema="public",
        signature="uuid-ossp"
    )
    op.create_entity(public_uuid_ossp)

Solution

PostgrSQL does support an IF EXISTS qualifier: https://www.postgresql.org/docs/current/sql-createextension.html

CREATE EXTENSION [ IF NOT EXISTS ] extension_name
    [ WITH ] [ SCHEMA schema_name ]
             [ VERSION version ]
             [ CASCADE ]

However this isn't available via alembic_utils presently: https://github.com/olirice/alembic_utils/blob/d5d69519159fdabb2cac7cc10f748415f9fe6537/src/alembic_utils/pg_extension.py#L31

def to_sql_statement_create(self) -> TextClause:
    """Generates a SQL "create extension" statement"""
    return sql_text(f'CREATE EXTENSION "{self.signature}" WITH SCHEMA {self.literal_schema};')

So for now, I've gone with this. I'll also consider submitting a PR to that project to add support for if exists perhaps.

def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    public_uuid_ossp = PGExtension(
        schema="public",
        signature="uuid-ossp"
    )
    conn = op.get_bind()
    query = "SELECT COUNT(*) FROM pg_available_extensions WHERE name = 'uuid-ossp' AND installed_version IS NOT NULL"
    exists = [row for row in conn.execute(query)][0][0] > 0
    if not exists:
        op.create_entity(public_uuid_ossp)