jroachgolf84 opened a new pull request, #66595:
URL: https://github.com/apache/airflow/pull/66595
## Description
With the foundation laid by AIP-103, the path forward for AIP-93 is a bit
more clear; if the `BaseEventTrigger` can be Asset-aware, the `Asset.get(...)`
and `Asset.set(...)` functionality can be leveraged.
#65103 drafted an approach that "flipped" the definition of the `Asset` and
`AssetWatcher`. However, for good reason, the approach was challenged. After
conversation, the goal became to "pass" the `Asset` through to the
`BaseEventTrigger` with some runtime magic.
## Testing
No unit tests have been written for this logic yet. However, testing has
been performed E2E locally with `breeze`. The trigger authored below is what
was used for testing. When this ran, the Asset `name` and `uri` were output in
the Triggerer logs.
```python
class GenericEventTrigger(BaseEventTrigger):
def __init__(
self,
random_number,
waiter_delay,
**kwargs
):
super().__init__(**kwargs)
self.random_number = random_number
self.waiter_delay = waiter_delay
def serialize(self) -> tuple[str, dict[str, Any]]:
"""Serialize the Trigger, including the func, params, and
waiter_delay."""
return (
self.__class__.__module__ + "." + self.__class__.__qualname__,
{
"random_number": self.random_number,
},
)
async def run(self) -> AsyncIterator[TriggerEvent]:
"""Logic that fires a TriggerEvent."""
logging.info(f"***** watched_asset: {self.watched_asset}")
logging.info(f"***** watched_asset.name: {self.watched_asset.name}")
logging.info(f"***** watched_asset.uri: {self.watched_asset.uri}")
while True:
result = random.randint(0, 5)
logging.info(f"result: {result}")
if result == self.random_number:
logging.info("yield'ing TriggerEvent")
yield TriggerEvent({"status": "success", "result": result})
break
logging.info(f"Sleeping for {self.waiter_delay} seconds")
await asyncio.sleep(self.waiter_delay)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
