API Reference

alembic_utils.replaceable_entity.register_entities(entities, schemas=None, exclude_schemas=None, entity_types=None)

Register entities to be monitored for changes when alembic is invoked with revision --autogenerate.

Parameters:

  • entities - List[ReplaceableEntity]: A list of entities (PGFunction, PGView, etc) to monitor for revisions

Deprecated Parameters:

Configure schema and object inclusion/exclusion with include_name and include_object in env.py. For more information see https://alembic.sqlalchemy.org/en/latest/autogenerate.html#controlling-what-to-be-autogenerated

  • schemas - Optional[List[str]]: A list of SQL schema names to monitor. Note, schemas referenced in registered entities are automatically monitored.
  • exclude_schemas - Optional[List[str]]: A list of SQL schemas to ignore. Note, explicitly registered entities will still be monitored.
  • entity_types - Optional[List[Type[ReplaceableEntity]]]: A list of ReplaceableEntity classes to consider during migrations. Other entity types are ignored
# migrations/env.py

from alembic_utils.replaceable_entity import register_entities
from app.functions import my_function
from app.views import my_view

register_entities(entities=[my_function, my_view], exclude_schema=['audit'])
class alembic_utils.pg_function.PGFunction(schema, signature, definition)

A PostgreSQL Function compatible with alembic revision --autogenerate

Parameters:

  • schema - str: A SQL schema name
  • signature - str: A SQL function's call signature
  • definition - str: The remainig function body and identifiers
from alembic_utils.pg_function import PGFunction

to_lower = PGFunction(
    schema="public",
    signature="to_lower(some_text text)",
    definition="returns text as $$ lower(some_text) $$ language sql"
)
class alembic_utils.pg_view.PGView(schema, signature, definition)

A PostgreSQL View compatible with alembic revision --autogenerate

Parameters:

  • schema - str: A SQL schema name
  • signature - str: A SQL view's call signature
  • definition - str: The SQL select statement body of the view
from alembic_utils.pg_view import PGView

scifi_books = PGView(
    schema="public",
    signature="scifi_books",
    definition="select * from books where genre='scifi'"
)
class alembic_utils.pg_materialized_view.PGMaterializedView(schema, signature, definition, with_data=True)

A PostgreSQL Materialized View compatible with alembic revision --autogenerate

Limitations: Materialized views may not have other views or materialized views that depend on them.

Parameters:

  • schema - str: A SQL schema name
  • signature - str: A SQL view's call signature
  • definition - str: The SQL select statement body of the view
  • with_data - bool: Should create and replace statements populate data
from alembic_utils.pg_materialized_view import PGMaterializedView

scifi_books = PGMaterializedView(
    schema="public",
    signature="scifi_books",
    definition="select * from books where genre='scifi'",
    with_data=True
)
class alembic_utils.pg_trigger.PGTrigger(schema, signature, definition, on_entity, is_constraint=False)

A PostgreSQL Trigger compatible with alembic revision --autogenerate

Parameters:

  • schema - str: A SQL schema name
  • signature - str: A SQL trigger's call signature
  • definition - str: The remainig trigger body and identifiers
  • on_entity - str: fully qualifed entity that the policy applies
  • is_constraint - bool: Is the trigger a constraint trigger

Postgres Create Trigger Specification:

CREATE [ CONSTRAINT ] TRIGGER name { BEFORE | AFTER | INSTEAD OF } { event [ OR ... ] }
ON table
[ FROM referenced_table_name ]
[ NOT DEFERRABLE | [ DEFERRABLE ] { INITIALLY IMMEDIATE | INITIALLY DEFERRED } ]
[ FOR [ EACH ] { ROW | STATEMENT } ]
[ WHEN ( condition ) ]
EXECUTE PROCEDURE function_name ( arguments )

from alembic_utils.pg_trigger import PGTrigger

trigger = PGTrigger(
    schema="public",
    signature="lower_account_email",
    on_entity="public.account",
    definition="""
        BEFORE INSERT ON public.account
        FOR EACH ROW EXECUTE FUNCTION public.downcase_email()
    """,
)
class alembic_utils.pg_extension.PGExtension(schema, signature)

A PostgreSQL Extension compatible with alembic revision --autogenerate

Parameters:

  • schema - str: A SQL schema name
  • signature - str: A PostgreSQL extension's name
from alembic_utils.pg_extension import PGExtension

extension = PGExtension(
    schema="public",
    signature="uuid-ossp",
)
class alembic_utils.pg_policy.PGPolicy(schema, signature, definition, on_entity)

A PostgreSQL Policy compatible with alembic revision --autogenerate

Parameters:

  • schema - str: A SQL schema name
  • signature - str: A SQL policy name and tablename, separated by "."
  • definition - str: The definition of the policy, incl. permissive, for, to, using, with check
  • on_entity - str: fully qualifed entity that the policy applies
from alembic_utils.pg_policy import PGPolicy

policy = PGPolicy(
    schema="public",
    signature="allow_read",
    on_entity="public.account",
    definition="""
        AS PERMISSIVE
        FOR SELECT
        TO api_user
        USING (id = current_setting('api_current_user', true)::int)
    """,
)
class alembic_utils.pg_grant_table.PGGrantTable(schema, table, role, grant, columns=None, with_grant_option=False)

A PostgreSQL Grant Statement compatible with alembic revision --autogenerate

PGGrantTable requires the role name being used to generate migrations to match the role name that executes migrations.

If your system does not meet that requirement, disable them by excluding PGGrantTable in include_object https://alembic.sqlalchemy.org/en/latest/api/runtime.html#alembic.runtime.environment.EnvironmentContext.configure.params.include_object

Parameters:

  • schema - str: A SQL schema name
  • table - str: The table to grant access to
  • columns - List[str]: A list of column names on table to grant access to
  • role - str: The role to grant access to
  • grant - Union[Grant, str]: On of SELECT, INSERT, UPDATE, DELETE, TRUNCATE, REFERENCES, TRIGGER
  • with_grant_option - bool: Can the role grant access to other roles
from alembic_utils.pg_grant_table import PGGrantTable

grant = PGGrantTable(
    schema="public",
    table="account",
    columns=["id", "email"],
    role="anon_user",
    grant='SELECT',
    with_grant_option=False,
)