1# Licensed to the Apache Software Foundation (ASF) under one
2# or more contributor license agreements. See the NOTICE file
3# distributed with this work for additional information
4# regarding copyright ownership. The ASF licenses this file
5# to you under the Apache License, Version 2.0 (the
6# "License"); you may not use this file except in compliance
7# with the License. You may obtain a copy of the License at
8#
9# http://www.apache.org/licenses/LICENSE-2.0
10#
11# Unless required by applicable law or agreed to in writing,
12# software distributed under the License is distributed on an
13# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14# KIND, either express or implied. See the License for the
15# specific language governing permissions and limitations
16# under the License.
17from __future__ import annotations
18
19import abc
20from typing import Any, AsyncIterator
21
22from airflow.utils.log.logging_mixin import LoggingMixin
23
24
25class BaseTrigger(abc.ABC, LoggingMixin):
26 """
27 Base class for all triggers.
28
29 A trigger has two contexts it can exist in:
30
31 - Inside an Operator, when it's passed to TaskDeferred
32 - Actively running in a trigger worker
33
34 We use the same class for both situations, and rely on all Trigger classes
35 to be able to return the arguments (possible to encode with Airflow-JSON) that will
36 let them be re-instantiated elsewhere.
37 """
38
39 def __init__(self, **kwargs):
40 # these values are set by triggerer when preparing to run the instance
41 # when run, they are injected into logger record.
42 self.task_instance = None
43 self.trigger_id = None
44
45 def _set_context(self, context):
46 """Part of LoggingMixin and used mainly for configuration of task logging; not used for triggers."""
47 raise NotImplementedError
48
49 @abc.abstractmethod
50 def serialize(self) -> tuple[str, dict[str, Any]]:
51 """
52 Return the information needed to reconstruct this Trigger.
53
54 :return: Tuple of (class path, keyword arguments needed to re-instantiate).
55 """
56 raise NotImplementedError("Triggers must implement serialize()")
57
58 @abc.abstractmethod
59 async def run(self) -> AsyncIterator[TriggerEvent]:
60 """
61 Run the trigger in an asynchronous context.
62
63 The trigger should yield an Event whenever it wants to fire off
64 an event, and return None if it is finished. Single-event triggers
65 should thus yield and then immediately return.
66
67 If it yields, it is likely that it will be resumed very quickly,
68 but it may not be (e.g. if the workload is being moved to another
69 triggerer process, or a multi-event trigger was being used for a
70 single-event task defer).
71
72 In either case, Trigger classes should assume they will be persisted,
73 and then rely on cleanup() being called when they are no longer needed.
74 """
75 raise NotImplementedError("Triggers must implement run()")
76 yield # To convince Mypy this is an async iterator.
77
78 async def cleanup(self) -> None:
79 """
80 Cleanup the trigger.
81
82 Called when the trigger is no longer needed, and it's being removed
83 from the active triggerer process.
84
85 This method follows the async/await pattern to allow to run the cleanup
86 in triggerer main event loop. Exceptions raised by the cleanup method
87 are ignored, so if you would like to be able to debug them and be notified
88 that cleanup method failed, you should wrap your code with try/except block
89 and handle it appropriately (in async-compatible way).
90 """
91
92 def __repr__(self) -> str:
93 classpath, kwargs = self.serialize()
94 kwargs_str = ", ".join(f"{k}={v}" for k, v in kwargs.items())
95 return f"<{classpath} {kwargs_str}>"
96
97
98class TriggerEvent:
99 """
100 Something that a trigger can fire when its conditions are met.
101
102 Events must have a uniquely identifying value that would be the same
103 wherever the trigger is run; this is to ensure that if the same trigger
104 is being run in two locations (for HA reasons) that we can deduplicate its
105 events.
106 """
107
108 def __init__(self, payload: Any):
109 self.payload = payload
110
111 def __repr__(self) -> str:
112 return f"TriggerEvent<{self.payload!r}>"
113
114 def __eq__(self, other):
115 if isinstance(other, TriggerEvent):
116 return other.payload == self.payload
117 return False