airflow.models.trigger¶
Module Contents¶
Classes¶
Triggers are a workload that run in an asynchronous event loop shared with |
- class airflow.models.trigger.Trigger(classpath, kwargs, created_date=None)[source]¶
Bases:
airflow.models.base.BaseTriggers are a workload that run in an asynchronous event loop shared with other Triggers, and fire off events that will unpause deferred Tasks, start linked DAGs, etc.
They are persisted into the database and then re-hydrated into a "triggerer" process, where many are run at once. We model it so that there is a many-to-one relationship between Task and Trigger, for future deduplication logic to use.
Rows will be evicted from the database when the triggerer detects no active Tasks/DAGs using them. Events are not stored in the database; when an Event is fired, the triggerer will directly push its data to the appropriate Task/DAG.
- classmethod from_object(cls, trigger)[source]¶
Alternative constructor that creates a trigger row based directly off of a Trigger object.
- classmethod bulk_fetch(cls, ids, session=None)[source]¶
Fetches all of the Triggers by ID and returns a dict mapping ID -> Trigger instance
- classmethod clean_unused(cls, session=None)[source]¶
Deletes all triggers that have no tasks/DAGs dependent on them (triggers have a one-to-many relationship to both)
- classmethod submit_event(cls, trigger_id, event, session=None)[source]¶
Takes an event from an instance of itself, and triggers all dependent tasks to resume.
- classmethod submit_failure(cls, trigger_id, exc=None, session=None)[source]¶
Called when a trigger has failed unexpectedly, and we need to mark everything that depended on it as failed. Notably, we have to actually run the failure code from a worker as it may have linked callbacks, so hilariously we have to re-schedule the task instances to a worker just so they can then fail.
We use a special __fail__ value for next_method to achieve this that the runtime code understands as immediate-fail, and pack the error into next_kwargs.
TODO: Once we have shifted callback (and email) handling to run on workers as first-class concepts, we can run the failure code here in-process, but we can't do that right now.