Triggers¶

from sqlalchemy.orm import declarative_base
from sqlalchemy_declarative_extensions import declarative_database, Function, Functions, Triggers
from sqlalchemy_declarative_extensions.dialects.postgresql import Trigger

_Base = declarative_base()


@declarative_database
class Base(_Base):
    __abstract__ = True

    functions = Functions().are(
        Function(
            "fancy_trigger",
            """
            BEGIN
            INSERT INTO foo (id) select NEW.id + 1;
            RETURN NULL;
            END
            """,
            language="plpgsql",
            returns="trigger",
        )
    )

    triggers = Triggers().are(
        Trigger.after("insert", on="foo", execute="fancy_trigger")
        .named("on_insert_foo")
        .when("pg_trigger_depth() < 1")
        .for_each_row(),
    )

Note

Triggers options are wildly different across dialects. As such, you should always use the dialect-specific sqlalchemy_declarative_extensions.trigger.base.Trigger subclasses.

Note

Trigger behavior not fully implemented at current time, although it should be functional for the options it does support. Any ability to instantiate an object which produces a syntax error should be considered a bug. Additionally, feature requests for supporting more trigger options are welcome!

class sqlalchemy_declarative_extensions.trigger.base.Trigger¶

Describes a generic trigger.

execute: str¶

property identity: Hashable¶

Return the dialect-specific identity used to compare triggers.

Return type:: Hashable

name: str¶

named(name)¶

Parameters:: name (str)

on: str¶

abstract to_sql_create()¶

to_sql_drop()¶

to_sql_update(connection=None)¶

Parameters:: connection (sqlalchemy.engine.Connection | None)

class sqlalchemy_declarative_extensions.trigger.base.Triggers¶

append(trigger)¶

Parameters:: trigger (Trigger)

are(*triggers)¶

Parameters:: triggers (Trigger)

classmethod coerce_from_unknown(unknown)¶

Parameters:: unknown (None | Iterable[Trigger] | Triggers)
Return type:: Triggers | None

classmethod extract(metadata)¶

Parameters:: metadata (sqlalchemy.MetaData | list[sqlalchemy.MetaData | None] | None)
Return type:: Self | None

ignore: list[str] = []¶

ignore_unspecified: bool = False¶

include: list[str] | None = None¶

triggers: list[Trigger] = []¶

sqlalchemy_declarative_extensions.trigger.base.register_trigger(base_or_metadata, trigger)¶

This can be used instead of the static registration through Triggers on a declarative base or MetaData, to imperitively register triggers.

Parameters:

base_or_metadata (sqlalchemy_declarative_extensions.sqlalchemy.HasMetaData | sqlalchemy.MetaData)
trigger (Trigger)