[QUESTION] compatibility with sqlalchemy_utils.functions.create_database about alembic_utils HOT 2 CLOSED

from alembic_utils.experimental import collect instances
from alembic_utils.pg_function import PGFunction

def create_database(connection, metadata) -> None:
    """Create database from sqlalchemy metadata + alembic_utils PGFunctions"""
    sqlalchemy_utils.functions.create_database(...)

    for function in collect_instances(your_app, PGFunction):
        connection.execute(function.to_sql_statement_create())

def pgfunction_to_sqlalchemy_text_create_or_replace(function: PGFunction):
    statement = f"CREATE OR REPLACE FUNCTION {function.literal_schema}.{function.literal_signature} {function.definition}"
    return text(statement)


def create_all_pg_triggers(connection: Connection, pg_triggers: Iterable[PGTrigger]):
    for pg_trigger in pg_triggers:
        for generated_text in pg_trigger.to_sql_statement_create_or_replace():
            connection.execute(generated_text.text)


def create_all_pg_functions(connection: Connection, pg_functions: Iterable[PGFunction]):
    for pg_function in pg_functions:
        connection.execute(pgfunction_to_sqlalchemy_text_create_or_replace(pg_function))


@event.listens_for(BaseModel.metadata, 'after_create', named=True)
def create_category_pg_functions_and_triggers_on_db(connection: Connection, **kwargs):
    create_all_pg_functions(connection=connection, pg_functions=PG_FUNCTION_LIST)
    create_all_pg_triggers(connection=connection, pg_triggers=PG_TRIGGER_LIST)