Skip to content

# 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`

    165
    166
    167
    168
    169

    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`

    77
    78
    79
    80
    81
    82
    83
    84

    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`

    512
    513

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

#### `` pgtrigger.Func.render

    render(model: 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`

    515
    516
    517
    518
    519
    520
    521
    522
    523
    524
    525
    526
    527

    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 = None)

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

Source code in `pgtrigger/core.py`

    196
    197
    198
    199
    200

    def __init__(self, sql: str | None = 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`

    414
    415
    416
    417
    418
    419
    420
    421
    422
    423
    424
    425
    426
    427
    428
    429
    430
    431

    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`

    437
    438
    439
    440
    441
    442
    443
    444
    445
    446
    447
    448
    449
    450
    451
    452
    453
    454

    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`

    460
    461
    462
    463
    464
    465
    466
    467
    468
    469
    470
    471
    472
    473
    474
    475
    476
    477

    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`

    483
    484
    485
    486
    487
    488
    489
    490
    491
    492
    493
    494
    495
    496
    497
    498
    499
    500

    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 = 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`

    196
    197
    198
    199
    200

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

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

### `` pgtrigger.F

    F(*args: Any, **kwargs: Any)

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`

    253
    254
    255
    256
    257
    258
    259
    260
    261
    262
    263

    def __init__(self, *args: Any, **kwargs: Any) -> None:
        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 = None, level: Level | None = None, when: When | None = None, operation: Operation | None = None, condition: Condition | None = None, referencing: Referencing | None = None, func: Func | str | None = None, declare: List[Tuple[str, str]] | None = None, timing: Timing | None = None)

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

Source code in `pgtrigger/core.py`

    562
    563
    564
    565
    566
    567
    568
    569
    570
    571
    572
    573
    574
    575
    576
    577
    578
    579
    580
    581
    582
    583
    584
    585
    586
    587
    588
    589
    590
    591
    592
    593
    594
    595
    596
    597
    598
    599
    600
    601
    602
    603
    604
    605
    606
    607
    608
    609

    def __init__(
        self,
        *,
        name: str | None = None,
        level: Level | None = None,
        when: When | None = None,
        operation: Operation | None = None,
        condition: Condition | None = None,
        referencing: Referencing | None = None,
        func: Func | str | None = None,
        declare: List[Tuple[str, str]] | None = None,
        timing: Timing | None = 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: 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`

    801
    802
    803
    804
    805
    806
    807
    808
    809
    810
    811
    812
    813
    814
    815
    816

    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: Model) -> 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`

    772
    773
    774
    775
    776
    777
    778
    779
    780
    781
    782
    783
    784
    785
    786
    787
    788
    789
    790
    791
    792
    793
    794
    795
    796
    797
    798
    799

    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: 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`

    962
    963
    964
    965
    966
    967
    968
    969
    970
    971

    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: 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`

    949
    950
    951
    952
    953
    954
    955
    956
    957
    958
    959
    960

    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: 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`

    829
    830
    831
    832
    833
    834
    835
    836
    837
    838
    839
    840
    841
    842
    843
    844
    845
    846
    847
    848

    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`

    818
    819
    820
    821
    822
    823
    824
    825
    826
    827

    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: 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`

    652
    653
    654
    655
    656
    657
    658
    659
    660
    661

    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: 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`

    663
    664
    665
    666
    667
    668
    669
    670
    671
    672
    673
    674
    675

    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: 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`

    677
    678
    679
    680
    681
    682
    683
    684
    685
    686
    687
    688
    689
    690

    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: 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`

    850
    851
    852
    853
    854
    855
    856
    857
    858
    859
    860
    861
    862
    863
    864
    865
    866
    867
    868
    869
    870
    871
    872
    873
    874
    875
    876
    877
    878
    879
    880
    881
    882
    883
    884
    885
    886
    887
    888
    889
    890
    891
    892
    893
    894
    895
    896
    897
    898
    899
    900

    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: 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`

    629
    630
    631
    632
    633
    634
    635
    636
    637
    638
    639
    640
    641
    642
    643
    644
    645
    646
    647
    648
    649
    650

    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: 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 "<app>.<model>:<trigger>"

Source code in `pgtrigger/core.py`

    692
    693
    694
    695
    696
    697
    698
    699
    700
    701
    702

    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: 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`

    924
    925
    926
    927
    928
    929
    930
    931
    932
    933
    934

    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: 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`

    902
    903
    904
    905
    906
    907
    908
    909
    910
    911

    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: 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`

    704
    705
    706
    707
    708
    709
    710
    711
    712
    713
    714
    715
    716
    717
    718
    719
    720
    721

    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: 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`

    723
    724
    725
    726
    727
    728
    729
    730
    731
    732
    733
    734
    735
    736
    737
    738
    739
    740

    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: 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`

    742
    743
    744
    745
    746
    747
    748
    749
    750
    751
    752
    753

    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: 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`

    755
    756
    757
    758
    759
    760
    761
    762
    763
    764
    765
    766
    767
    768
    769
    770

    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: 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`

    936
    937
    938
    939
    940
    941
    942
    943
    944
    945
    946
    947

    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: 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`

    913
    914
    915
    916
    917
    918
    919
    920
    921
    922

    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`

    614
    615
    616
    617
    618
    619
    620
    621
    622
    623
    624
    625
    626
    627

    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 = None, level: Level | None = None, when: When | None = None, operation: Operation | None = None, condition: Condition | None = None, referencing: Referencing | None = None, func: Func | str | None = None, declare: List[Tuple[str, str]] | None = None, timing: Timing | None = None)

Bases: `Trigger`

A trigger that raises an exception.

Source code in `pgtrigger/core.py`

    562
    563
    564
    565
    566
    567
    568
    569
    570
    571
    572
    573
    574
    575
    576
    577
    578
    579
    580
    581
    582
    583
    584
    585
    586
    587
    588
    589
    590
    591
    592
    593
    594
    595
    596
    597
    598
    599
    600
    601
    602
    603
    604
    605
    606
    607
    608
    609

    def __init__(
        self,
        *,
        name: str | None = None,
        level: Level | None = None,
        when: When | None = None,
        operation: Operation | None = None,
        condition: Condition | None = None,
        referencing: Referencing | None = None,
        func: Func | str | None = None,
        declare: List[Tuple[str, str]] | None = None,
        timing: Timing | None = 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`

    65
    66
    67
    68
    69
    70
    71
    72
    73
    74
    75
    76
    77
    78

    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[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`

    200
    201
    202
    203
    204
    205
    206
    207
    208
    209
    210
    211
    212
    213
    214

    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[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`

    118
    119
    120
    121
    122
    123
    124
    125
    126
    127
    128
    129
    130
    131
    132
    133
    134
    135
    136
    137
    138
    139
    140
    141
    142
    143
    144
    145
    146
    147
    148
    149
    150
    151
    152
    153
    154

    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`

    268
    269
    270
    271
    272
    273
    274
    275
    276
    277
    278
    279
    280
    281
    282
    283
    284
    285
    286
    287
    288
    289

    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`

    305
    306
    307
    308
    309
    310
    311
    312
    313
    314
    315
    316
    317
    318
    319
    320
    321
    322
    323
    324
    325
    326
    327
    328
    329
    330
    331
    332
    333
    334
    335
    336
    337

    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) -> Generator[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`

    174
    175
    176
    177
    178
    179
    180
    181
    182
    183
    184
    185
    186
    187
    188
    189
    190
    191
    192
    193
    194
    195
    196
    197
    198
    199
    200
    201
    202
    203
    204
    205
    206
    207

    @contextlib.contextmanager
    def ignore(*uris: str, databases: Union[List[str], None] = None) -> Generator[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) -> Generator[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`

    281
    282
    283
    284
    285
    286
    287
    288
    289
    290
    291
    292
    293
    294
    295
    296
    297
    298
    299

    @contextlib.contextmanager
    def schema(*schemas: str, databases: Union[List[str], None] = None) -> Generator[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`

    131
    132
    133
    134
    135
    136
    137
    138
    139
    140
    141
    142
    143
    144
    145
    146
    147
    148
    149
    150
    151
    152
    153
    154
    155
    156
    157
    158
    159
    160
    161
    162

    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`

    115
    116
    117
    118
    119
    120
    121
    122
    123
    124
    125
    126
    127
    128

    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`

    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35

    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`

    121
    122
    123
    124
    125
    126
    127
    128
    129
    130
    131
    132
    133
    134
    135
    136
    137
    138
    139
    140

    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`

    102
    103
    104
    105
    106
    107
    108
    109
    110
    111
    112
    113
    114
    115
    116
    117
    118

    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`

    143
    144
    145
    146
    147
    148
    149
    150
    151
    152
    153
    154
    155
    156
    157
    158
    159

    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`

    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    67
    68
    69
    70
    71
    72
    73
    74
    75
    76

    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`

    79
    80
    81
    82
    83
    84
    85
    86
    87
    88
    89
    90
    91
    92
    93
    94
    95
    96
    97
    98
    99

    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)