async/await lets a trigger wait for an external operation without blocking the workflow’s worker. The asynchronous method suspends during an await, allowing the triggerer to handle other deferred work. section 9.5.2
A useful implementation pattern is:
- Define the trigger’s run method with async def. section 9.5.2
- Use await asyncio.sleep() while waiting before retrying an external request. section 9.5.2
- Keep heavy logic out of the asynchronous flow so waiting does not disrupt triggerer operation. section 9.5.2
Polling for MovieLens ratings
A custom MovieLens trigger repeatedly checks for ratings within a date range. When no records exist, it awaits the configured interval; when records appear, it logs success and returns control to the sensor. section 9.5.2
Listing 9.33 Defining a custom trigger class
class MovielensTrigger(BaseTrigger):
def __init__(self,
conn_id,
start_date,
end_date,
sleep_interval,
):
super().__init__()
self._sleep_interval = sleep_interval
self._conn_id = conn_id
self._start_date = start_date
self._end_date = end_date
def serialize(self):
return ("custom.triggers.MovielensTrigger", { #1
"sleep_interval": self._sleep_interval,
"conn_id": self._conn_id,
"start_date": self._start_date,
"end_date": self._end_date,
}
)
async def run(self): #2
with MovielensHook(self._conn_id) as hook: #3
found_records = True
while not found_records: #4
try:
next(
hook.get_ratings( #5
start_date=self._start_date,
end_date=self._end_date,
batch_size=1
)
)
self.log.info(f"Found ratings for {self._start_date)")
found_records = True #6
except StopIteration: #7
self.log.info(
➥ f"Didn't find ratings for {self._start_date}")
self.log.info("waiting...")
await asyncio.sleep(self.check_interval) #8
yield TriggerEvent(str(uuid.uuid4())) #9
Inspect the custom trigger structure, including serialization and the async run method. section 9.5.2
