[QUESTION] compatibility with sqlalchemy_utils.functions.create_database

[mhaselsteiner]

When defining a Postgres function using PGFunction the function is only created when migrations are applied. When creating a new database using sqlalchemy_utils.functions.create_database and the sqlalchemy schema definition the function is not there, unless I define the same function as sqlalchemy DDL and associate it with an event. Is there a way to also create a DDL with an event in sqlalchemy for each PGFunction through alembic_utils more elegantly/less redundant?

[olirice]

alembic and, by extension, alembic_utils provide tooling for people who want to produce procedural migration scripts for altering their DB's schema. Supporting point-in-time reflection is not a goal of the project

But, if your use case is simple enough, you may be able to do something like

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())

[mhaselsteiner]

Thanks for the quick reply! I followed a similar approach using event listening

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)