mafw.db.trigger

Module provides a Trigger class and related tools to create triggers in the database via the ORM.

It supports SQLite, MySQL and PostgreSQL with dialect-specific SQL generation.

Functions

`and_`(*conditions)	Concatenates conditions with logical AND.
`or_`(*conditions)	Concatenates conditions with logical OR.

Classes

`MySQLDialect`()	MySQL-specific trigger SQL generation.
`PostgreSQLDialect`()	PostgreSQL-specific trigger SQL generation.
`SQLiteDialect`()	SQLite-specific trigger SQL generation.
`Trigger`(trigger_name, trigger_type, source_table)	Trigger template wrapper for use with peewee ORM.
`TriggerAction`(value)	String enumerator for the trigger action (Delete, Insert, Update)
`TriggerDialect`()	Abstract base class for database-specific trigger SQL generation.
`TriggerWhen`(value)	String enumerator for the trigger execution time (Before, After or Instead Of)

class mafw.db.trigger.MySQLDialect[source]

Bases: TriggerDialect

MySQL-specific trigger SQL generation.

create_trigger_sql(trigger: Trigger) → str[source]: Generate MySQL trigger SQL.

drop_trigger_sql(trigger_name: str, safe: bool = True, table_name: str | None = None) → str[source]: Generate MySQL drop trigger SQL.

supports_safe_create() → bool[source]: MySQL supports IF NOT EXISTS for triggers.

supports_trigger_type(when: TriggerWhen, action: TriggerAction, on_view: bool = False) → bool[source]: MySQL doesn’t support INSTEAD OF triggers.

supports_update_of_columns() → bool[source]: MySQL doesn’t support column-specific UPDATE triggers.

supports_when_clause() → bool[source]: MySQL supports conditions but through WHERE instead of WHEN.

class mafw.db.trigger.PostgreSQLDialect[source]

Bases: TriggerDialect

PostgreSQL-specific trigger SQL generation.

_clean_sql(sql: str) → str[source]

Remove RETURNING clauses from SQL statements for PostgreSQL trigger functions.

Parameters:: sql – The SQL statement
Returns:: SQL statement without RETURNING clause

create_trigger_sql(trigger: Trigger) → str[source]: Generate PostgreSQL trigger SQL.

drop_trigger_sql(trigger_name: str, safe: bool = True, table_name: str | None = None) → str[source]: Generate PostgreSQL drop trigger SQL.

supports_safe_create() → bool[source]: PostgreSQL doesn’t support IF NOT EXISTS for triggers before v14, but we implement safety differently.

supports_trigger_type(when: TriggerWhen, action: TriggerAction, on_view: bool = False) → bool[source]: PostgreSQL supports INSTEAD OF only on views.

supports_update_of_columns() → bool[source]: PostgreSQL supports column-specific UPDATE triggers.

supports_when_clause() → bool[source]: PostgreSQL supports WHEN conditions.

class mafw.db.trigger.SQLiteDialect[source]

Bases: TriggerDialect

SQLite-specific trigger SQL generation.

create_trigger_sql(trigger: Trigger) → str[source]: Generate SQLite trigger SQL.

drop_trigger_sql(trigger_name: str, safe: bool = True, table_name: str | None = None) → str[source]: Generate SQLite drop trigger SQL.

supports_safe_create() → bool[source]: SQLite supports IF NOT EXISTS for triggers.

supports_trigger_type(when: TriggerWhen, action: TriggerAction, on_view: bool = False) → bool[source]: SQLite supports all trigger types except INSTEAD OF on tables (only on views).

supports_update_of_columns() → bool[source]: SQLite supports column-specific UPDATE triggers.

supports_when_clause() → bool[source]: SQLite supports WHEN conditions.

class mafw.db.trigger.Trigger(trigger_name: str, trigger_type: tuple[TriggerWhen, TriggerAction], source_table: type[Model] | Model | str, safe: bool = False, for_each_row: bool = False, update_columns: list[str] | None = None, on_view: bool = False)[source]

Bases: object

Trigger template wrapper for use with peewee ORM.

Constructor parameters:

Parameters:

trigger_name (str) – The name of this trigger. It needs to be unique!
trigger_type (tuple[TriggerWhen, TriggerAction]) – A tuple with TriggerWhen and TriggerAction to specify on which action the trigger should be invoked and if before, after or instead of.
source_table (type[Model] | Model | str) – The table originating the trigger. It can be a model class, instance, or also the name of the table.
safe (bool, Optional) – A boolean flag to define if in the trigger creation statement a ‘IF NOT EXISTS’ clause should be included. Defaults to False
for_each_row (bool, Optional) – A boolean flag to repeat the script content for each modified row in the table. Defaults to False.
update_columns (list[str], Optional) – A list of column names. When defining a trigger on a table update, it is possible to restrict the firing of the trigger to the cases when a subset of all columns have been updated. An column is updated also when the new value is equal to the old one. If you want to discriminate this case, use the add_when() method. Defaults to None.
on_view (bool, Optional) – A boolean flag to indicate if the target is a view. This affects the support for INSTEAD OF. Defaults to False.

_get_dialect() → TriggerDialect[source]

Get the appropriate dialect based on the database type.

Returns:: A dialect instance

_node_to_sql(node: Node) → str[source]

Convert a peewee Node (Expression, Query, etc.) to a SQL string with interpolated parameters.

This is based on peewee’s internal query_to_string function for debugging/logging purposes.

Added in version v2.0.0.

Parameters:: node – A peewee Node object
Returns:: SQL string with parameters interpolated

_value_to_sql(value: Any) → str[source]

Convert a Python value to its SQL representation.

Added in version v2.0.0.

Parameters:: value – A Python value (string, int, float, None, etc.)
Returns:: SQL string representation of the value

add_sql(sql: str | Query) → Self[source]

Add an SQL statement to be executed by the trigger.

The sql can be either a string containing the sql statement, or it can be any other peewee Query.

For example:

# assuming you have created a trigger ...

sql = AnotherTable.insert(
    field1=some_value, field2=another_value
)
trigger.add_sql(sql)

In this way the SQL code is generated with parametric placeholder if needed.

Parameters:: sql (str | peewee.Query) – The SQL statement.
Returns:: self for easy chaining
Return type:: Trigger

add_when(*conditions: str | Node) → Self[source]

Add conditions to the when statements.

Conditions are logically ANDed. To have mixed OR and AND logic, use the functions and_() and or_().

The conditions can be either strings containing SQL conditions, or peewee Node objects (such as Expression or Query objects).

For example:

# String condition
trigger.add_when("NEW.status = 'active'")

# Peewee expression
subq = TriggerStatus.select(TriggerStatus.status).where(
    TriggerStatus.trigger_type == 'DELETE_FILES'
)
trigger.add_when(Value(1) == subq)

Changed in version v2.0.0: The argument can also be a generic peewee Node.

Parameters:: conditions (str | peewee.Node) – Conditions to be added with logical AND. Can be strings or peewee Node objects.
Returns:: self for easy chaining
Return type:: Trigger

create() → str[source]

Generates the SQL create statement.

Returns:

The trigger creation statement.

Raises:

MissingSQLStatement – if no SQL statements are provided.
UnsupportedDatabaseError – if the trigger type is not supported by the database.

drop(safe: bool = True) → str[source]

Generates the SQL drop statement.

Parameters:: safe (bool, Optional) – If True, add an IF EXIST. Defaults to True.
Returns:: The drop statement
Return type:: str

set_database(database: Database | DatabaseProxy) → Self[source]

Set the database to use for this trigger.

Parameters:: database – The database instance
Returns:: self for easy chaining

class mafw.db.trigger.TriggerAction(value)[source]

Bases: StrEnum

String enumerator for the trigger action (Delete, Insert, Update)

_generate_next_value_(start, count, last_values): Return the lower-cased version of the member name.

class mafw.db.trigger.TriggerDialect[source]

Bases: ABC

Abstract base class for database-specific trigger SQL generation.

abstractmethod create_trigger_sql(trigger: Trigger) → str[source]

Generate the SQL to create a trigger for a specific database dialect.

Parameters:: trigger – The trigger object
Returns:: SQL string to create the trigger

abstractmethod drop_trigger_sql(trigger_name: str, safe: bool = True, table_name: str | None = None) → str[source]

Generate the SQL to drop a trigger for a specific database dialect.

Parameters:

trigger_name (str) – The name of the trigger to drop
safe (bool, Optional) – If True, add an IF EXISTS clause. Defaults to True.
table_name (str, Optional) – The name of the target table for the trigger. Defaults to None.

Returns:

SQL string to drop the trigger

Return type:

str

abstractmethod supports_safe_create() → bool[source]

Check if the database supports IF NOT EXISTS for triggers.

Returns:: True if supported, False otherwise

abstractmethod supports_trigger_type(when: TriggerWhen, action: TriggerAction, on_view: bool = False) → bool[source]

Check if the database supports the specified trigger type.

Parameters:

when – When the trigger should fire (BEFORE, AFTER, INSTEAD OF)
action – The action that triggers the trigger (INSERT, UPDATE, DELETE)
on_view – Whether the trigger is on a view

Returns:

True if supported, False otherwise

abstractmethod supports_update_of_columns() → bool[source]

Check if the database supports column-specific UPDATE triggers.

Returns:: True if supported, False otherwise

abstractmethod supports_when_clause() → bool[source]

Check if the database supports WHEN conditions.

Returns:: True if supported, False otherwise

class mafw.db.trigger.TriggerWhen(value)[source]

Bases: StrEnum

String enumerator for the trigger execution time (Before, After or Instead Of)

_generate_next_value_(start, count, last_values): Return the lower-cased version of the member name.

mafw.db.trigger.and_(*conditions: str) → str[source]

Concatenates conditions with logical AND.

Parameters:: conditions (str) – The condition to join.
Returns:: The and-concatenated string of conditions
Return type:: str

mafw.db.trigger.or_(*conditions: str) → str[source]

Concatenates conditions with logical OR.

Parameters:: conditions (str) – The condition to join.
Returns:: The or-concatenated string of conditions.
Return type:: str