Module

Below are the core classes and functions of the pgtrigger module.

Level clause

pgtrigger.Row module-attribute

Row = Level('ROW')

For specifying row-level triggers (the default)

pgtrigger.Statement module-attribute

Statement = Level('STATEMENT')

For specifying statement-level triggers

When clause

pgtrigger.After module-attribute

After = When('AFTER')

For specifying AFTER in the when clause of a trigger.

pgtrigger.Before module-attribute

Before = When('BEFORE')

For specifying BEFORE in the when clause of a trigger.

pgtrigger.InsteadOf module-attribute

InsteadOf = When('INSTEAD OF')

For specifying INSTEAD OF in the when clause of a trigger.

Operation clause

pgtrigger.Insert module-attribute

Insert = Operation('INSERT')

For specifying INSERT as the trigger operation.

pgtrigger.Update module-attribute

Update = Operation('UPDATE')

For specifying UPDATE as the trigger operation.

pgtrigger.Delete module-attribute

Delete = Operation('DELETE')

For specifying DELETE as the trigger operation.

pgtrigger.Truncate module-attribute

Truncate = Operation('TRUNCATE')

For specifying TRUNCATE as the trigger operation.

pgtrigger.UpdateOf

UpdateOf(*columns)

Bases: Operation

For specifying UPDATE OF as the trigger operation.

Source code in pgtrigger/core.py

def __init__(self, *columns):
    if not columns:
        raise ValueError("Must provide at least one column")

    self.columns = columns

Referencing clause

pgtrigger.Referencing

Referencing(*, old=None, new=None)

For specifying the REFERENCING clause of a statement-level trigger

Source code in pgtrigger/core.py

def __init__(self, *, old=None, new=None):
    if not old and not new:
        raise ValueError(
            'Must provide either "old" and/or "new" to the referencing'
            " construct of a trigger"
        )

    self.old = old
    self.new = new

Timing clause

pgtrigger.Immediate module-attribute

Immediate = Timing('IMMEDIATE')

For deferrable triggers that run immediately by default

pgtrigger.Deferred module-attribute

Deferred = Timing('DEFERRED')

For deferrable triggers that run at the end of the transaction by default

Func clause

pgtrigger.Func

Func(func)

Allows for rendering a function with access to the "meta", "fields", and "columns" variables of the current model.

For example, func=Func("SELECT {columns.id} FROM {meta.db_table};") makes it possible to do inline SQL in the Meta of a model and reference its properties.

Source code in pgtrigger/core.py

def __init__(self, func):
    self.func = func

pgtrigger.Func.render

render(model: models.Model) -> str

Render the SQL of the function.

Parameters:

Name	Type	Description	Default
model	Model	The model.	required

Returns:

Type	Description
str	The rendered SQL.

Source code in pgtrigger/core.py

def render(self, model: models.Model) -> str:
    """
    Render the SQL of the function.

    Args:
        model: The model.

    Returns:
        The rendered SQL.
    """
    fields = utils.AttrDict({field.name: field for field in model._meta.fields})
    columns = utils.AttrDict({field.name: field.column for field in model._meta.fields})
    return self.func.format(meta=model._meta, fields=fields, columns=columns)

Conditions

pgtrigger.Condition

Condition(sql: str = None)

For specifying free-form SQL in the condition of a trigger.

Source code in pgtrigger/core.py

def __init__(self, sql: str = None):
    self.sql = sql or self.sql

    if not self.sql:
        raise ValueError("Must provide SQL to condition")

pgtrigger.AnyChange

AnyChange(
    *fields: str,
    exclude: Union[List[str], None] = None,
    exclude_auto: Union[bool, None] = None
)

Bases: _Change

If any supplied fields change, trigger the condition.

Parameters:

Name	Type	Description	Default
*fields	str	If any supplied fields change, trigger the condition. If no fields are supplied, defaults to all fields on the model.	()
exclude	Union[List[str], None]	Fields to exclude.	None
exclude_auto	Union[bool, None]	Exclude all auto_now and auto_now_add fields automatically.	None

Source code in pgtrigger/core.py

def __init__(
    self,
    *fields: str,
    exclude: Union[List[str], None] = None,
    exclude_auto: Union[bool, None] = None,
):
    """
    If any supplied fields change, trigger the condition.

    Args:
        *fields: If any supplied fields change, trigger the condition.
            If no fields are supplied, defaults to all fields on the model.
        exclude: Fields to exclude.
        exclude_auto: Exclude all `auto_now` and `auto_now_add` fields automatically.
    """
    super().__init__(
        *fields, exclude=exclude, exclude_auto=exclude_auto, all=False, comparison="df"
    )

pgtrigger.AnyDontChange

AnyDontChange(
    *fields: str,
    exclude: Union[List[str], None] = None,
    exclude_auto: Union[bool, None] = None
)

Bases: _Change

If any supplied fields don't change, trigger the condition.

Parameters:

Name	Type	Description	Default
*fields	str	If any supplied fields don't change, trigger the condition. If no fields are supplied, defaults to all fields on the model.	()
exclude	Union[List[str], None]	Fields to exclude.	None
exclude_auto	Union[bool, None]	Exclude all auto_now and auto_now_add fields automatically.	None

Source code in pgtrigger/core.py

def __init__(
    self,
    *fields: str,
    exclude: Union[List[str], None] = None,
    exclude_auto: Union[bool, None] = None,
):
    """
    If any supplied fields don't change, trigger the condition.

    Args:
        *fields: If any supplied fields don't change, trigger the condition.
            If no fields are supplied, defaults to all fields on the model.
        exclude: Fields to exclude.
        exclude_auto: Exclude all `auto_now` and `auto_now_add` fields automatically.
    """
    super().__init__(
        *fields, exclude=exclude, exclude_auto=exclude_auto, all=False, comparison="ndf"
    )

pgtrigger.AllChange

AllChange(
    *fields: str,
    exclude: Union[List[str], None] = None,
    exclude_auto: Union[bool, None] = None
)

Bases: _Change

If all supplied fields change, trigger the condition.

Parameters:

Name	Type	Description	Default
*fields	str	If all supplied fields change, trigger the condition. If no fields are supplied, defaults to all fields on the model.	()
exclude	Union[List[str], None]	Fields to exclude.	None
exclude_auto	Union[bool, None]	Exclude all auto_now and auto_now_add fields automatically.	None

Source code in pgtrigger/core.py

def __init__(
    self,
    *fields: str,
    exclude: Union[List[str], None] = None,
    exclude_auto: Union[bool, None] = None,
):
    """
    If all supplied fields change, trigger the condition.

    Args:
        *fields: If all supplied fields change, trigger the condition.
            If no fields are supplied, defaults to all fields on the model.
        exclude: Fields to exclude.
        exclude_auto: Exclude all `auto_now` and `auto_now_add` fields automatically.
    """
    super().__init__(
        *fields, exclude=exclude, exclude_auto=exclude_auto, all=True, comparison="df"
    )

pgtrigger.AllDontChange

AllDontChange(
    *fields: str,
    exclude: Union[List[str], None] = None,
    exclude_auto: Union[bool, None] = None
)

Bases: _Change

If all supplied don't fields change, trigger the condition.

Parameters:

Name	Type	Description	Default
*fields	str	If all supplied fields don't change, trigger the condition. If no fields are supplied, defaults to all fields on the model.	()
exclude	Union[List[str], None]	Fields to exclude.	None
exclude_auto	Union[bool, None]	Exclude all auto_now and auto_now_add fields automatically.	None

Source code in pgtrigger/core.py

def __init__(
    self,
    *fields: str,
    exclude: Union[List[str], None] = None,
    exclude_auto: Union[bool, None] = None,
):
    """
    If all supplied fields don't change, trigger the condition.

    Args:
        *fields: If all supplied fields don't change, trigger the condition.
            If no fields are supplied, defaults to all fields on the model.
        exclude: Fields to exclude.
        exclude_auto: Exclude all `auto_now` and `auto_now_add` fields automatically.
    """
    super().__init__(
        *fields, exclude=exclude, exclude_auto=exclude_auto, all=True, comparison="ndf"
    )

pgtrigger.Q

Q(sql: str = None)

Bases: Q, Condition

Similar to Django's Q object, allows building filter clauses based on the old and new rows in a trigger condition.

Source code in pgtrigger/core.py

def __init__(self, sql: str = None):
    self.sql = sql or self.sql

    if not self.sql:
        raise ValueError("Must provide SQL to condition")

pgtrigger.F

F(*args, **kwargs)

Bases: F

Similar to Django's F object, allows referencing the old and new rows in a trigger condition.

Source code in pgtrigger/core.py

def __init__(self, *args, **kwargs):
    super().__init__(*args, **kwargs)

    if self.name.startswith("old__"):
        self.row_alias = "OLD"
    elif self.name.startswith("new__"):
        self.row_alias = "NEW"
    else:
        raise ValueError("F() values must reference old__ or new__")

    self.col_name = self.name[5:]

pgtrigger.IsDistinctFrom

Bases: Lookup

A custom IS DISTINCT FROM field lookup for common trigger conditions. For example, pgtrigger.Q(old__field__df=pgtrigger.F("new__field")).

pgtrigger.IsNotDistinctFrom

Bases: Lookup

A custom IS NOT DISTINCT FROM field lookup for common trigger conditions. For example, pgtrigger.Q(old__field__ndf=pgtrigger.F("new__field")).

Triggers

pgtrigger.Trigger

Trigger(
    *,
    name: str = None,
    level: Level = None,
    when: When = None,
    operation: Operation = None,
    condition: Union[Condition, None] = None,
    referencing: Union[Referencing, None] = None,
    func: Union[Func, str] = None,
    declare: Union[List[Tuple[str, str]], None] = None,
    timing: Union[Timing, None] = None
)

For specifying a free-form PL/pgSQL trigger function or for creating derived trigger classes.

Source code in pgtrigger/core.py

def __init__(
    self,
    *,
    name: str = None,
    level: Level = None,
    when: When = None,
    operation: Operation = None,
    condition: Union[Condition, None] = None,
    referencing: Union[Referencing, None] = None,
    func: Union[Func, str] = None,
    declare: Union[List[Tuple[str, str]], None] = None,
    timing: Union[Timing, None] = None,
):
    self.name = name or self.name
    self.level = level or self.level
    self.when = when or self.when
    self.operation = operation or self.operation
    self.condition = condition or self.condition
    self.referencing = referencing or self.referencing
    self.func = func or self.func
    self.declare = declare or self.declare
    self.timing = timing or self.timing

    if not self.level or not isinstance(self.level, Level):
        raise ValueError(f'Invalid "level" attribute: {self.level}')

    if not self.when or not isinstance(self.when, When):
        raise ValueError(f'Invalid "when" attribute: {self.when}')

    if not self.operation or not isinstance(self.operation, Operation):
        raise ValueError(f'Invalid "operation" attribute: {self.operation}')

    if self.timing and not isinstance(self.timing, Timing):
        raise ValueError(f'Invalid "timing" attribute: {self.timing}')

    if self.level == Row and self.referencing:
        raise ValueError('Row-level triggers cannot have a "referencing" attribute')

    if self.timing and self.level != Row:
        raise ValueError('Deferrable triggers must have "level" attribute as "pgtrigger.Row"')

    if self.timing and self.when != After:
        raise ValueError('Deferrable triggers must have "when" attribute as "pgtrigger.After"')

    if not self.name:
        raise ValueError('Trigger must have "name" attribute')

    self.validate_name()

pgtrigger.Trigger.allow_migrate

allow_migrate(model: models.Model, database: Union[str, None] = None) -> bool

True if the trigger for this model can be migrated.

Defaults to using the router's allow_migrate.

Parameters:

Name	Type	Description	Default
model	Model	The model.	required
database	Union[str, None]	The name of the database configuration.	None

Returns:

Type	Description
bool	True if the trigger for the model can be migrated.

Source code in pgtrigger/core.py

def allow_migrate(self, model: models.Model, database: Union[str, None] = None) -> bool:
    """True if the trigger for this model can be migrated.

    Defaults to using the router's allow_migrate.

    Args:
        model: The model.
        database: The name of the database configuration.

    Returns:
        `True` if the trigger for the model can be migrated.
    """
    model = model._meta.concrete_model
    return utils.is_postgres(database) and router.allow_migrate(
        database or DEFAULT_DB_ALIAS, model._meta.app_label, model_name=model._meta.model_name
    )

pgtrigger.Trigger.compile

compile(model: models.Model) -> compiler.Trigger

Create a compiled representation of the trigger. useful for migrations.

Parameters:

Name	Type	Description	Default
model	Model	The model	required

Returns:

Type	Description
Trigger	The compiled trigger object.

Source code in pgtrigger/core.py

def compile(self, model: models.Model) -> compiler.Trigger:
    """
    Create a compiled representation of the trigger. useful for migrations.

    Args:
        model: The model

    Returns:
        The compiled trigger object.
    """
    return compiler.Trigger(
        name=self.name,
        sql=compiler.UpsertTriggerSql(
            ignore_func_name=_ignore_func_name(),
            pgid=self.get_pgid(model),
            declare=self.render_declare(model),
            func=self.render_func(model),
            table=model._meta.db_table,
            constraint="CONSTRAINT" if self.timing else "",
            when=self.when,
            operation=self.operation,
            timing=f"DEFERRABLE INITIALLY {self.timing}" if self.timing else "",
            referencing=self.referencing or "",
            level=self.level,
            condition=self.render_condition(model),
            execute=self.render_execute(model),
        ),
    )

pgtrigger.Trigger.disable

disable(model: models.Model, database: Union[str, None] = None)

Disables the trigger for a model.

Parameters:

Name	Type	Description	Default
model	Model	The model.	required
database	Union[str, None]	The name of the database configuration.	None

Source code in pgtrigger/core.py

def disable(self, model: models.Model, database: Union[str, None] = None):
    """Disables the trigger for a model.

    Args:
        model: The model.
        database: The name of the database configuration.
    """
    disable_sql = self.compile(model).disable_sql
    self.exec_sql(disable_sql, model, database=database)
    return _cleanup_on_exit(lambda: self.enable(model, database=database))  # pragma: no branch

pgtrigger.Trigger.enable

enable(model: models.Model, database: Union[str, None] = None)

Enables the trigger for a model.

Parameters:

Name	Type	Description	Default
model	Model	The model.	required
database	Union[str, None]	The name of the database configuration.	None

Source code in pgtrigger/core.py

def enable(self, model: models.Model, database: Union[str, None] = None):
    """Enables the trigger for a model.

    Args:
        model: The model.
        database: The name of the database configuration.
    """
    enable_sql = self.compile(model).enable_sql
    self.exec_sql(enable_sql, model, database=database)
    return _cleanup_on_exit(  # pragma: no branch
        lambda: self.disable(model, database=database)
    )

pgtrigger.Trigger.exec_sql

exec_sql(
    sql: str,
    model: models.Model,
    database: Union[str, None] = None,
    fetchall: bool = False,
) -> Any

Conditionally execute SQL if migrations are allowed.

Parameters:

Name	Type	Description	Default
sql	str	The SQL string.	required
model	Model	The model.	required
database	Union[str, None]	The name of the database configuration.	None
fetchall	bool	True if all results should be fetched	False

Returns:

Type	Description
Any	A psycopg cursor result

Source code in pgtrigger/core.py

def exec_sql(
    self,
    sql: str,
    model: models.Model,
    database: Union[str, None] = None,
    fetchall: bool = False,
) -> Any:
    """Conditionally execute SQL if migrations are allowed.

    Args:
        sql: The SQL string.
        model: The model.
        database: The name of the database configuration.
        fetchall: True if all results should be fetched

    Returns:
        A psycopg cursor result
    """
    if self.allow_migrate(model, database=database):
        return utils.exec_sql(str(sql), database=database, fetchall=fetchall)

pgtrigger.Trigger.format_sql

format_sql(sql: str) -> str

Returns SQL as one line that has trailing whitespace removed from each line.

Parameters:

Name	Type	Description	Default
sql	str	The unformatted SQL	required

Returns:

Type	Description
str	The formatted SQL

Source code in pgtrigger/core.py

def format_sql(self, sql: str) -> str:
    """Returns SQL as one line that has trailing whitespace removed from each line.

    Args:
        sql: The unformatted SQL

    Returns:
        The formatted SQL
    """
    return " ".join(line.strip() for line in sql.split("\n") if line.strip()).strip()

pgtrigger.Trigger.get_condition

get_condition(model: models.Model) -> Condition

Get the condition of the trigger.

Parameters:

Name	Type	Description	Default
model	Model	The model.	required

Returns:

Type	Description
Condition	The condition.

Source code in pgtrigger/core.py

def get_condition(self, model: models.Model) -> Condition:
    """Get the condition of the trigger.

    Args:
        model: The model.

    Returns:
        The condition.
    """
    return self.condition

pgtrigger.Trigger.get_declare

get_declare(model: models.Model) -> List[Tuple[str, str]]

Gets the DECLARE part of the trigger function if any variables are used.

Parameters:

Name	Type	Description	Default
model	Model	The model	required

Returns:

Type	Description
List[Tuple[str, str]]	A list of variable name / type tuples that will
List[Tuple[str, str]]	be shown in the DECLARE. For example [('row_data', 'JSONB')]

Source code in pgtrigger/core.py

def get_declare(self, model: models.Model) -> List[Tuple[str, str]]:
    """
    Gets the DECLARE part of the trigger function if any variables
    are used.

    Args:
        model: The model

    Returns:
        A list of variable name / type tuples that will
        be shown in the DECLARE. For example [('row_data', 'JSONB')]
    """
    return self.declare or []

pgtrigger.Trigger.get_func

get_func(model: models.Model) -> Union[str, Func]

Returns the trigger function that comes between the BEGIN and END clause.

Parameters:

Name	Type	Description	Default
model	Model	The model	required

Returns:

Type	Description
Union[str, Func]	The trigger function as a SQL string or pgtrigger.Func object.

Source code in pgtrigger/core.py

def get_func(self, model: models.Model) -> Union[str, Func]:
    """
    Returns the trigger function that comes between the BEGIN and END
    clause.

    Args:
        model: The model

    Returns:
        The trigger function as a SQL string or [pgtrigger.Func][] object.
    """
    if not self.func:
        raise ValueError("Must define func attribute or implement get_func")
    return self.func

pgtrigger.Trigger.get_installation_status

get_installation_status(
    model: models.Model, database: Union[str, None] = None
) -> Tuple[str, Union[bool, None]]

Returns the installation status of a trigger.

The return type is (status, enabled), where status is one of:

1. INSTALLED: If the trigger is installed
2. UNINSTALLED: If the trigger is not installed
3. OUTDATED: If the trigger is installed but has been modified
4. IGNORED: If migrations are not allowed

"enabled" is True if the trigger is installed and enabled or false if installed and disabled (or uninstalled).

Parameters:

Name	Type	Description	Default
model	Model	The model.	required
database	Union[str, None]	The name of the database configuration.	None

Returns:

Type	Description
Tuple[str, Union[bool, None]]	A tuple with the installation and enablement status.

Source code in pgtrigger/core.py

def get_installation_status(
    self, model: models.Model, database: Union[str, None] = None
) -> Tuple[str, Union[bool, None]]:
    """Returns the installation status of a trigger.

    The return type is (status, enabled), where status is one of:

    1. `INSTALLED`: If the trigger is installed
    2. `UNINSTALLED`: If the trigger is not installed
    3. `OUTDATED`: If the trigger is installed but has been modified
    4. `IGNORED`: If migrations are not allowed

    "enabled" is True if the trigger is installed and enabled or false
    if installed and disabled (or uninstalled).

    Args:
        model: The model.
        database: The name of the database configuration.

    Returns:
        A tuple with the installation and enablement status.
    """
    if not self.allow_migrate(model, database=database):
        return (UNALLOWED, None)

    trigger_exists_sql = f"""
        SELECT oid, obj_description(oid) AS hash, tgenabled AS enabled
        FROM pg_trigger
        WHERE tgname='{self.get_pgid(model)}'
            AND tgrelid='{utils.quote(model._meta.db_table)}'::regclass;
    """
    try:
        with transaction.atomic(using=database):
            results = self.exec_sql(
                trigger_exists_sql, model, database=database, fetchall=True
            )
    except ProgrammingError:  # pragma: no cover
        # When the table doesn't exist yet, possibly because migrations
        # haven't been executed, a ProgrammingError will happen because
        # of an invalid regclass cast. Return 'UNINSTALLED' for this
        # case
        return (UNINSTALLED, None)

    if not results:
        return (UNINSTALLED, None)
    else:
        hash = self.compile(model).hash
        if hash != results[0][1]:
            return (OUTDATED, results[0][2] == "O")
        else:
            return (INSTALLED, results[0][2] == "O")

pgtrigger.Trigger.get_pgid

get_pgid(model: models.Model) -> str

The ID of the trigger and function object in postgres

All objects are prefixed with "pgtrigger_" in order to be discovered/managed by django-pgtrigger.

Parameters:

Name	Type	Description	Default
model	Model	The model.	required

Returns:

Type	Description
str	The Postgres ID.

Source code in pgtrigger/core.py

def get_pgid(self, model: models.Model) -> str:
    """The ID of the trigger and function object in postgres

    All objects are prefixed with "pgtrigger_" in order to be
    discovered/managed by django-pgtrigger.

    Args:
        model: The model.

    Returns:
        The Postgres ID.
    """
    model_hash = hashlib.sha1(self.get_uri(model).encode()).hexdigest()[:5]
    pgid = f"pgtrigger_{self.name}_{model_hash}"

    if len(pgid) > 63:
        raise ValueError(f'Trigger identifier "{pgid}" is greater than 63 chars')

    # NOTE - Postgres always stores names in lowercase. Ensure that all
    # generated IDs are lowercase so that we can properly do installation
    # and pruning tasks.
    return pgid.lower()

pgtrigger.Trigger.get_uri

get_uri(model: models.Model) -> str

The URI for the trigger.

Parameters:

Name	Type	Description	Default
model	Model	The model	required

Returns:

Type	Description
str	The URI in the format of the ".:"

Source code in pgtrigger/core.py

def get_uri(self, model: models.Model) -> str:
    """The URI for the trigger.

    Args:
        model: The model

    Returns:
        The URI in the format of the "<app>.<model>:<trigger>"
    """

    return f"{model._meta.app_label}.{model._meta.object_name}:{self.name}"

pgtrigger.Trigger.install

install(model: models.Model, database: Union[str, None] = None)

Installs the trigger for a model.

Parameters:

Name	Type	Description	Default
model	Model	The model.	required
database	Union[str, None]	The name of the database configuration.	None

Source code in pgtrigger/core.py

def install(self, model: models.Model, database: Union[str, None] = None):
    """Installs the trigger for a model.

    Args:
        model: The model.
        database: The name of the database configuration.
    """
    install_sql = self.compile(model).install_sql
    with transaction.atomic(using=database):
        self.exec_sql(install_sql, model, database=database)
    return _cleanup_on_exit(lambda: self.uninstall(model, database=database))

pgtrigger.Trigger.register

register(*models: models.Model)

Register model classes with the trigger

Parameters:

Name	Type	Description	Default
*models	Model	Models to register to this trigger.	()

Source code in pgtrigger/core.py

def register(self, *models: models.Model):
    """Register model classes with the trigger

    Args:
        *models: Models to register to this trigger.
    """
    for model in models:
        registry.set(self.get_uri(model), model=model, trigger=self)

    return _cleanup_on_exit(lambda: self.unregister(*models))

pgtrigger.Trigger.render_condition

render_condition(model: models.Model) -> str

Renders the condition SQL in the trigger declaration.

Parameters:

Name	Type	Description	Default
model	Model	The model.	required

Returns:

Type	Description
str	The rendered condition SQL

Source code in pgtrigger/core.py

def render_condition(self, model: models.Model) -> str:
    """Renders the condition SQL in the trigger declaration.

    Args:
        model: The model.

    Returns:
        The rendered condition SQL
    """
    condition = self.get_condition(model)
    resolved = condition.resolve(model).strip() if condition else ""

    if resolved:
        if not resolved.startswith("("):
            resolved = f"({resolved})"
        resolved = f"WHEN {resolved}"

    return resolved

pgtrigger.Trigger.render_declare

render_declare(model: models.Model) -> str

Renders the DECLARE of the trigger function, if any.

Parameters:

Name	Type	Description	Default
model	Model	The model.	required

Returns:

Type	Description
str	The rendered declare SQL.

Source code in pgtrigger/core.py

def render_declare(self, model: models.Model) -> str:
    """Renders the DECLARE of the trigger function, if any.

    Args:
        model: The model.

    Returns:
        The rendered declare SQL.
    """
    declare = self.get_declare(model)
    if declare:
        rendered_declare = "DECLARE " + " ".join(
            f"{var_name} {var_type};" for var_name, var_type in declare
        )
    else:
        rendered_declare = ""

    return rendered_declare

pgtrigger.Trigger.render_execute

render_execute(model: models.Model) -> str

Renders what should be executed by the trigger. This defaults to the trigger function.

Parameters:

Name	Type	Description	Default
model	Model	The model	required

Returns:

Type	Description
str	The SQL for the execution of the trigger function.

Source code in pgtrigger/core.py

def render_execute(self, model: models.Model) -> str:
    """
    Renders what should be executed by the trigger. This defaults
    to the trigger function.

    Args:
        model: The model

    Returns:
        The SQL for the execution of the trigger function.
    """
    return f"{self.get_pgid(model)}()"

pgtrigger.Trigger.render_func

render_func(model: models.Model) -> str

Renders the func.

Parameters:

Name	Type	Description	Default
model	Model	The model	required

Returns:

Type	Description
str	The rendered SQL of the trigger function

Source code in pgtrigger/core.py

def render_func(self, model: models.Model) -> str:
    """
    Renders the func.

    Args:
        model: The model

    Returns:
        The rendered SQL of the trigger function
    """
    func = self.get_func(model)

    if isinstance(func, Func):
        return func.render(model)
    else:
        return func

pgtrigger.Trigger.uninstall

uninstall(model: models.Model, database: Union[str, None] = None)

Uninstalls the trigger for a model.

Parameters:

Name	Type	Description	Default
model	Model	The model.	required
database	Union[str, None]	The name of the database configuration.	None

Source code in pgtrigger/core.py

def uninstall(self, model: models.Model, database: Union[str, None] = None):
    """Uninstalls the trigger for a model.

    Args:
        model: The model.
        database: The name of the database configuration.
    """
    uninstall_sql = self.compile(model).uninstall_sql
    self.exec_sql(uninstall_sql, model, database=database)
    return _cleanup_on_exit(  # pragma: no branch
        lambda: self.install(model, database=database)
    )

pgtrigger.Trigger.unregister

unregister(*models: models.Model)

Unregister model classes with the trigger.

Parameters:

Name	Type	Description	Default
*models	Model	Models to unregister to this trigger.	()

Source code in pgtrigger/core.py

def unregister(self, *models: models.Model):
    """Unregister model classes with the trigger.

    Args:
        *models: Models to unregister to this trigger.
    """
    for model in models:
        registry.delete(self.get_uri(model))

    return _cleanup_on_exit(lambda: self.register(*models))

pgtrigger.Trigger.validate_name

validate_name() -> None

Verifies the name is under the maximum length and has valid characters.

Raises:

Type	Description
ValueError	If the name is invalid

Source code in pgtrigger/core.py

def validate_name(self) -> None:
    """Verifies the name is under the maximum length and has valid characters.

    Raises:
        ValueError: If the name is invalid
    """
    if len(self.name) > MAX_NAME_LENGTH:
        raise ValueError(f'Trigger name "{self.name}" > {MAX_NAME_LENGTH} characters.')

    if not re.match(r"^[a-zA-Z0-9-_]+$", self.name):
        raise ValueError(
            f'Trigger name "{self.name}" has invalid characters.'
            " Only alphanumeric characters, hyphens, and underscores are allowed."
        )

pgtrigger.Protect

Protect(
    *,
    name: str = None,
    level: Level = None,
    when: When = None,
    operation: Operation = None,
    condition: Union[Condition, None] = None,
    referencing: Union[Referencing, None] = None,
    func: Union[Func, str] = None,
    declare: Union[List[Tuple[str, str]], None] = None,
    timing: Union[Timing, None] = None
)

Bases: Trigger

A trigger that raises an exception.

Source code in pgtrigger/core.py

def __init__(
    self,
    *,
    name: str = None,
    level: Level = None,
    when: When = None,
    operation: Operation = None,
    condition: Union[Condition, None] = None,
    referencing: Union[Referencing, None] = None,
    func: Union[Func, str] = None,
    declare: Union[List[Tuple[str, str]], None] = None,
    timing: Union[Timing, None] = None,
):
    self.name = name or self.name
    self.level = level or self.level
    self.when = when or self.when
    self.operation = operation or self.operation
    self.condition = condition or self.condition
    self.referencing = referencing or self.referencing
    self.func = func or self.func
    self.declare = declare or self.declare
    self.timing = timing or self.timing

    if not self.level or not isinstance(self.level, Level):
        raise ValueError(f'Invalid "level" attribute: {self.level}')

    if not self.when or not isinstance(self.when, When):
        raise ValueError(f'Invalid "when" attribute: {self.when}')

    if not self.operation or not isinstance(self.operation, Operation):
        raise ValueError(f'Invalid "operation" attribute: {self.operation}')

    if self.timing and not isinstance(self.timing, Timing):
        raise ValueError(f'Invalid "timing" attribute: {self.timing}')

    if self.level == Row and self.referencing:
        raise ValueError('Row-level triggers cannot have a "referencing" attribute')

    if self.timing and self.level != Row:
        raise ValueError('Deferrable triggers must have "level" attribute as "pgtrigger.Row"')

    if self.timing and self.when != After:
        raise ValueError('Deferrable triggers must have "when" attribute as "pgtrigger.After"')

    if not self.name:
        raise ValueError('Trigger must have "name" attribute')

    self.validate_name()

pgtrigger.ReadOnly

ReadOnly(
    *,
    fields: Union[List[str], None] = None,
    exclude: Union[List[str], None] = None,
    **kwargs: Any
)

Bases: Protect

A trigger that prevents edits to fields.

If fields are provided, will protect edits to only those fields. If exclude is provided, will protect all fields except the ones excluded. If none of these arguments are provided, all fields cannot be edited.

Source code in pgtrigger/contrib.py

def __init__(
    self,
    *,
    fields: Union[List[str], None] = None,
    exclude: Union[List[str], None] = None,
    **kwargs: Any,
):
    self.fields = fields or self.fields
    self.exclude = exclude or self.exclude

    if self.fields and self.exclude:
        raise ValueError('Must provide only one of "fields" or "exclude" to ReadOnly trigger')

    super().__init__(**kwargs)

pgtrigger.SoftDelete

SoftDelete(
    *,
    name: str = None,
    condition: Union[core.Condition, None] = None,
    field: str = None,
    value: Union[bool, str, int, None] = _unset
)

Bases: Trigger

Sets a field to a value when a delete happens.

Supply the trigger with the "field" that will be set upon deletion and the "value" to which it should be set. The "value" defaults to False.

Note

This trigger currently only supports nullable BooleanField, CharField, and IntField fields.

Source code in pgtrigger/contrib.py

def __init__(
    self,
    *,
    name: str = None,
    condition: Union[core.Condition, None] = None,
    field: str = None,
    value: Union[bool, str, int, None] = _unset,
):
    self.field = field or self.field
    self.value = value if value is not _unset else self.value

    if not self.field:  # pragma: no cover
        raise ValueError('Must provide "field" for soft delete')

    super().__init__(name=name, condition=condition)

pgtrigger.FSM

FSM(
    *,
    name: str = None,
    condition: Union[core.Condition, None] = None,
    field: str = None,
    transitions: List[Tuple[str, str]] = None,
    separator: str = None
)

Bases: Trigger

Enforces a finite state machine on a field.

Supply the trigger with the field that transitions and then a list of tuples of valid transitions to the transitions argument.

Note

Only non-null CharField fields without quotes are currently supported. If your strings have a colon symbol in them, you must override the "separator" argument to be a value other than a colon.

Source code in pgtrigger/contrib.py

def __init__(
    self,
    *,
    name: str = None,
    condition: Union[core.Condition, None] = None,
    field: str = None,
    transitions: List[Tuple[str, str]] = None,
    separator: str = None,
):
    self.field = field or self.field
    self.transitions = transitions or self.transitions
    self.separator = separator or self.separator

    if not self.field:  # pragma: no cover
        raise ValueError('Must provide "field" for FSM')

    if not self.transitions:  # pragma: no cover
        raise ValueError('Must provide "transitions" for FSM')

    # This trigger doesn't accept quoted values or values that
    # contain the configured separator
    for value in itertools.chain(*self.transitions):
        if "'" in value or '"' in value:
            raise ValueError(f'FSM transition value "{value}" contains quotes')
        elif self.separator in value:
            raise ValueError(
                f'FSM value "{value}" contains separator "{self.separator}".'
                ' Configure your trigger with a different "separator" attribute'
            )

    # The separator must be a single character that isn't a quote
    if len(self.separator) != 1:
        raise ValueError(f'Separator "{self.separator}" must be a single character')
    elif self.separator in ('"', "'"):
        raise ValueError("Separator must not have quotes")

    super().__init__(name=name, condition=condition)

pgtrigger.UpdateSearchVector

UpdateSearchVector(
    *,
    name: str = None,
    vector_field: str = None,
    document_fields: List[str] = None,
    config_name: str = None
)

Bases: Trigger

Updates a django.contrib.postgres.search.SearchVectorField from document fields.

Supply the trigger with the vector_field that will be updated with changes to the document_fields. Optionally provide a config_name, which defaults to pg_catalog.english.

This trigger uses tsvector_update_trigger to update the vector field. See the Postgres docs for more information.

Note

UpdateSearchVector triggers are not compatible with pgtrigger.ignore since it references a built-in trigger. Trying to ignore this trigger results in a RuntimeError.

Source code in pgtrigger/contrib.py

def __init__(
    self,
    *,
    name: str = None,
    vector_field: str = None,
    document_fields: List[str] = None,
    config_name: str = None,
):
    self.vector_field = vector_field or self.vector_field
    self.document_fields = document_fields or self.document_fields
    self.config_name = config_name or self.config_name

    if not self.vector_field:
        raise ValueError('Must provide "vector_field" to update search vector')

    if not self.document_fields:
        raise ValueError('Must provide "document_fields" to update search vector')

    if not self.config_name:  # pragma: no cover
        raise ValueError('Must provide "config_name" to update search vector')

    super().__init__(name=name, operation=core.Insert | core.UpdateOf(*document_fields))

Runtime execution

pgtrigger.constraints

constraints(
    timing: Timing, *uris: str, databases: Union[List[str], None] = None
) -> None

Set deferrable constraint timing for the given triggers, which will persist until overridden or until end of transaction. Must be in a transaction to run this.

Parameters:

Name	Type	Description	Default
timing	Timing	The timing value that overrides the default trigger timing.	required
*uris	str	Trigger URIs over which to set constraint timing. If none are provided, all trigger constraint timing will be set. All triggers must be deferrable.	()
databases	Union[List[str], None]	The databases on which to set constraints. If none, all postgres databases will be used.	None

Raises:

Type	Description
RuntimeError	If the database of any triggers is not in a transaction.
ValueError	If any triggers are not deferrable.

Source code in pgtrigger/runtime.py

def constraints(timing: "Timing", *uris: str, databases: Union[List[str], None] = None) -> None:
    """
    Set deferrable constraint timing for the given triggers, which
    will persist until overridden or until end of transaction.
    Must be in a transaction to run this.

    Args:
        timing: The timing value that overrides the default trigger timing.
        *uris: Trigger URIs over which to set constraint timing.
            If none are provided, all trigger constraint timing will
            be set. All triggers must be deferrable.
        databases: The databases on which to set constraints. If none, all
            postgres databases will be used.

    Raises:
        RuntimeError: If the database of any triggers is not in a transaction.
        ValueError: If any triggers are not deferrable.
    """

    for model, trigger in registry.registered(*uris):
        if not trigger.timing:
            raise ValueError(
                f"Trigger {trigger.name} on model {model._meta.label_lower} is not deferrable."
            )

    for database in utils.postgres_databases(databases):
        if not connections[database].in_atomic_block:
            raise RuntimeError(f'Database "{database}" is not in a transaction.')

        names = ", ".join(trigger.get_pgid(model) for model, trigger in registry.registered(*uris))

        with connections[database].cursor() as cursor:
            cursor.execute(f"SET CONSTRAINTS {names} {timing}")

pgtrigger.ignore

ignore(*uris: str, databases: Union[List[str], None] = None)

Dynamically ignore registered triggers matching URIs from executing in an individual thread. If no URIs are provided, ignore all pgtriggers from executing in an individual thread.

Parameters:

Name	Type	Description	Default
*uris	str	Trigger URIs to ignore. If none are provided, all triggers will be ignored.	()
databases	Union[List[str], None]	The databases to use. If none, all postgres databases will be used.	None

Example

Ingore triggers in a context manager:

with pgtrigger.ignore("my_app.Model:trigger_name"):
    # Do stuff while ignoring trigger

Example

Ignore multiple triggers as a decorator:

@pgtrigger.ignore("my_app.Model:trigger_name", "my_app.Model:other_trigger")
def my_func():
    # Do stuff while ignoring trigger

Source code in pgtrigger/runtime.py

@contextlib.contextmanager
def ignore(*uris: str, databases: Union[List[str], None] = None):
    """
    Dynamically ignore registered triggers matching URIs from executing in
    an individual thread.
    If no URIs are provided, ignore all pgtriggers from executing in an
    individual thread.

    Args:
        *uris: Trigger URIs to ignore. If none are provided, all
            triggers will be ignored.
        databases: The databases to use. If none, all postgres databases
            will be used.

    Example:
        Ingore triggers in a context manager:

            with pgtrigger.ignore("my_app.Model:trigger_name"):
                # Do stuff while ignoring trigger

    Example:
        Ignore multiple triggers as a decorator:

            @pgtrigger.ignore("my_app.Model:trigger_name", "my_app.Model:other_trigger")
            def my_func():
                # Do stuff while ignoring trigger
    """
    with contextlib.ExitStack() as stack:
        stack.enter_context(_ignore_session(databases=databases))

        for model, trigger in registry.registered(*uris):
            stack.enter_context(_set_ignore_state(model, trigger))

        yield

pgtrigger.schema

schema(*schemas: str, databases: Union[List[str], None] = None)

Sets the search path to the provided schemas.

If nested, appends the schemas to the search path if not already in it.

Parameters:

Name	Type	Description	Default
*schemas	str	Schemas that should be appended to the search path. Schemas already in the search path from nested calls will not be appended.	()
databases	Union[List[str], None]	The databases to set the search path. If none, all postgres databases will be used.	None

Source code in pgtrigger/runtime.py

@contextlib.contextmanager
def schema(*schemas: str, databases: Union[List[str], None] = None):
    """
    Sets the search path to the provided schemas.

    If nested, appends the schemas to the search path if not already in it.

    Args:
        *schemas: Schemas that should be appended to the search path.
            Schemas already in the search path from nested calls will not be
            appended.
        databases: The databases to set the search path. If none, all postgres
            databases will be used.
    """
    with contextlib.ExitStack() as stack:
        stack.enter_context(_schema_session(databases=databases))
        stack.enter_context(_set_schema_state(*schemas))

        yield

Registry

pgtrigger.register

register(*triggers: Trigger) -> Callable

Register the given triggers with wrapped Model class.

Parameters:

Name	Type	Description	Default
*triggers	Trigger	Trigger classes to register.	()

Example

Register by decorating a model:

@pgtrigger.register(
    pgtrigger.Protect(
        name="append_only",
        operation=(pgtrigger.Update | pgtrigger.Delete)
    )
)
class MyModel(models.Model):
    pass

Example

Register by calling functionally:

pgtrigger.register(trigger_object)(MyModel)

Source code in pgtrigger/registry.py

def register(*triggers: "Trigger") -> Callable:
    """
    Register the given triggers with wrapped Model class.

    Args:
        *triggers: Trigger classes to register.

    Example:
        Register by decorating a model:

            @pgtrigger.register(
                pgtrigger.Protect(
                    name="append_only",
                    operation=(pgtrigger.Update | pgtrigger.Delete)
                )
            )
            class MyModel(models.Model):
                pass

    Example:
        Register by calling functionally:

            pgtrigger.register(trigger_object)(MyModel)
    """

    def _model_wrapper(model_class):
        for trigger in triggers:
            trigger.register(model_class)

        return model_class

    return _model_wrapper

pgtrigger.registered

registered(*uris: str) -> List[Tuple[Model, Trigger]]

Get registered trigger objects.

Parameters:

Name	Type	Description	Default
*uris	str	URIs of triggers to get. If none are provided, all triggers are returned. URIs are in the format of {app_label}.{model_name}:{trigger_name}.	()

Returns:

Type	Description
List[Tuple[Model, Trigger]]	Matching trigger objects.

Source code in pgtrigger/registry.py

def registered(*uris: str) -> List[Tuple["Model", "Trigger"]]:
    """
    Get registered trigger objects.

    Args:
        *uris: URIs of triggers to get. If none are provided,
            all triggers are returned. URIs are in the format of
            `{app_label}.{model_name}:{trigger_name}`.

    Returns:
        Matching trigger objects.
    """
    uris = uris or _registry.keys()
    return [_registry[uri] for uri in uris]

Installation

pgtrigger.install

install(*uris: str, database: Union[str, None] = None) -> None

Install triggers.

Parameters:

Name	Type	Description	Default
*uris	str	URIs of triggers to install. If none are provided, all triggers are installed and orphaned triggers are pruned.	()
database	Union[str, None]	The database. Defaults to the "default" database.	None

Source code in pgtrigger/installation.py

def install(*uris: str, database: Union[str, None] = None) -> None:
    """
    Install triggers.

    Args:
        *uris: URIs of triggers to install. If none are provided,
            all triggers are installed and orphaned triggers are pruned.
        database: The database. Defaults to the "default" database.
    """
    for model, trigger in registry.registered(*uris):
        LOGGER.info(
            "pgtrigger: Installing %s trigger for %s table on %s database.",
            trigger,
            model._meta.db_table,
            database or DEFAULT_DB_ALIAS,
        )
        trigger.install(model, database=database)

    if not uris and features.prune_on_install():  # pragma: no branch
        prune(database=database)

pgtrigger.uninstall

uninstall(*uris: str, database: Union[str, None] = None) -> None

Uninstalls triggers.

Parameters:

Name	Type	Description	Default
*uris	str	URIs of triggers to uninstall. If none are provided, all triggers are uninstalled and orphaned triggers are pruned.	()
database	Union[str, None]	The database. Defaults to the "default" database.	None

Source code in pgtrigger/installation.py

def uninstall(*uris: str, database: Union[str, None] = None) -> None:
    """
    Uninstalls triggers.

    Args:
        *uris: URIs of triggers to uninstall. If none are provided,
            all triggers are uninstalled and orphaned triggers are pruned.
        database: The database. Defaults to the "default" database.
    """
    for model, trigger in registry.registered(*uris):
        LOGGER.info(
            "pgtrigger: Uninstalling %s trigger for %s table on %s database.",
            trigger,
            model._meta.db_table,
            database or DEFAULT_DB_ALIAS,
        )
        trigger.uninstall(model, database=database)

    if not uris and features.prune_on_install():
        prune(database=database)

pgtrigger.enable

enable(*uris: str, database: Union[str, None] = None) -> None

Enables registered triggers.

Parameters:

Name	Type	Description	Default
*uris	str	URIs of triggers to enable. If none are provided, all triggers are enabled.	()
database	Union[str, None]	The database. Defaults to the "default" database.	None

Source code in pgtrigger/installation.py

def enable(*uris: str, database: Union[str, None] = None) -> None:
    """
    Enables registered triggers.

    Args:
        *uris: URIs of triggers to enable. If none are provided,
            all triggers are enabled.
        database: The database. Defaults to the "default" database.
    """
    for model, trigger in registry.registered(*uris):
        LOGGER.info(
            "pgtrigger: Enabling %s trigger for %s table on %s database.",
            trigger,
            model._meta.db_table,
            database or DEFAULT_DB_ALIAS,
        )
        trigger.enable(model, database=database)

pgtrigger.disable

disable(*uris: str, database: Union[str, None] = None) -> None

Disables triggers.

Parameters:

Name	Type	Description	Default
*uris	str	URIs of triggers to disable. If none are provided, all triggers are disabled.	()
database	Union[str, None]	The database. Defaults to the "default" database.	None

Source code in pgtrigger/installation.py

def disable(*uris: str, database: Union[str, None] = None) -> None:
    """
    Disables triggers.

    Args:
        *uris: URIs of triggers to disable. If none are provided,
            all triggers are disabled.
        database: The database. Defaults to the "default" database.
    """
    for model, trigger in registry.registered(*uris):
        LOGGER.info(
            "pgtrigger: Disabling %s trigger for %s table on %s database.",
            trigger,
            model._meta.db_table,
            database or DEFAULT_DB_ALIAS,
        )
        trigger.disable(model, database=database)

pgtrigger.prunable

prunable(database: Union[str, None] = None) -> List[Tuple[str, str, bool, str]]

Return triggers that are candidates for pruning

Parameters:

Name	Type	Description	Default
database	Union[str, None]	The database. Defaults to the "default" database.	None

Returns:

Type	Description
List[Tuple[str, str, bool, str]]	A list of tuples consisting of the table, trigger ID, enablement, and database

Source code in pgtrigger/installation.py

def prunable(database: Union[str, None] = None) -> List[Tuple[str, str, bool, str]]:
    """Return triggers that are candidates for pruning

    Args:
        database: The database. Defaults to the "default" database.

    Returns:
        A list of tuples consisting of the table, trigger ID, enablement, and database
    """
    if not utils.is_postgres(database):
        return []

    registered = {
        (utils.quote(model._meta.db_table), trigger.get_pgid(model))
        for model, trigger in registry.registered()
    }

    with utils.connection(database).cursor() as cursor:
        parent_trigger_clause = "tgparentid = 0 AND" if utils.pg_maj_version(cursor) >= 13 else ""

        # Only select triggers that are in the current search path. We accomplish
        # this by parsing the tgrelid and only selecting triggers that don't have
        # a schema name in their path
        cursor.execute(
            f"""
            SELECT tgrelid::regclass, tgname, tgenabled
                FROM pg_trigger
                WHERE tgname LIKE 'pgtrigger_%%' AND
                      {parent_trigger_clause}
                      array_length(parse_ident(tgrelid::regclass::varchar), 1) = 1
            """
        )
        triggers = set(cursor.fetchall())

    return [
        (trigger[0], trigger[1], trigger[2] == "O", database or DEFAULT_DB_ALIAS)
        for trigger in triggers
        if (utils.quote(trigger[0]), trigger[1]) not in registered
    ]

pgtrigger.prune

prune(database: Union[str, None] = None) -> None

Remove any pgtrigger triggers in the database that are not used by models. I.e. if a model or trigger definition is deleted from a model, ensure it is removed from the database

Parameters:

Name	Type	Description	Default
database	Union[str, None]	The database. Defaults to the "default" database.	None

Source code in pgtrigger/installation.py

def prune(database: Union[str, None] = None) -> None:
    """
    Remove any pgtrigger triggers in the database that are not used by models.
    I.e. if a model or trigger definition is deleted from a model, ensure
    it is removed from the database

    Args:
        database: The database. Defaults to the "default" database.
    """
    for trigger in prunable(database=database):
        LOGGER.info(
            "pgtrigger: Pruning trigger %s for table %s on %s database.",
            trigger[1],
            trigger[0],
            trigger[3],
        )

        connection = connections[trigger[3]]
        uninstall_sql = utils.render_uninstall(trigger[0], trigger[1])
        with connection.cursor() as cursor:
            cursor.execute(uninstall_sql)