cc @andrewgodwin

If we really want to allow user to do stuff post-yield, and still accept only one event, it's possible to do:

async def do_loop_with_stuff_after_yield():
    inner_loops = 0
    while True:
        inner_loops += 1
        print(f"{inner_loops}: before ")
        yield "hi"
        print(f"{inner_loops}: after")
        return

async def main():
    outer_loops = 0
    async for event in do_loop_with_stuff_after_yield():
        outer_loops += 1
        if outer_loops > 1:
            raise RuntimeError("Yielding more than one event is not supported.")
        print(outer_loops, ',', event)

asyncio.run(main())

But were we to do this, I think it would still be possible (though perhaps not super likely) that the trigger in question is canceled before it comes around to resume the iteration and thus the code between yield and return would not have been run. As such, it does not seem like something anyone should ever do.

Though if we really desired this, we could actually guarantee that the code between yield and return is always run. And the way to do this would be to delay adding the event to the events deque until after the generator exits. But then the event emission would be delayed until the trigger returns, which begs the question, why not just delay yielding the event in the first place.

So, there doesn't seem to be much value in running code post yield. At the same time, it is a little weird to just not fully consume the generator that the user wrote, and moreover to do so silently. Curious if @uranusjr might have a thought.

I don’t have particular thoughts toward the analyses, but this change actually raises a big question in the interface design. If we always just break after the first yielded event, why do events need to be yielded in the first place anyway? Why is run on a trigger not just returning an event instead? That would resolve the entire post-yield work discussion since it’s not possible to do things after a return in the first place. It seems to me if we do this, we should start re-design the trigger interface for 2.8.

I don’t have particular thoughts toward the analyses, but this change actually raises a big question in the interface design. If we always just break after the first yielded event, why do events need to be yielded in the first place anyway? Why is run on a trigger not just returning an event instead? That would resolve the entire post-yield work discussion since it’s not possible to do things after a return in the first place. It seems to me if we do this, we should start re-design the trigger interface for 2.8.

Very good point. I did not realise why it has bothered me, but this sums it up.

Why is run on a trigger not just returning an event instead? That would resolve the entire post-yield work discussion since it’s not possible to do things after a return in the first place. It seems to me if we do this, we should start re-design the trigger interface for 2.8.

I had the same question, but I found the answer in the deferring doc :

Currently Triggers are only used up to their first event, as they are only used for resuming deferred tasks (which happens on the first event fired). However, we plan to allow DAGs to be launched from triggers in future, which is where multi-event triggers will be more useful.

I don't know if this plan is still valid or not, but switching from yield event then return to just return event is a good idea. I already tested it when I added a return statement after each yield, but I tried to avoid changing the design.

Yup I had the same thought, why yield at all if we're just going to accept one event. It's possible that we could support both. Immediately after we call run, nothing happens at that point and we can inspect whether it's a generator or just coroutine and handle it appropriately. Thus I think we could incrementally expand the interface (by putting it in user's hands, that they can use simple return) while maintaining backcompat by supporting yielding events as well. Or deprecate.

Example code:

async def not_gen():
    return "hi"
async def is_gen():
    yield "yo"
    return

async def main():
    f = not_gen()
    g = is_gen()
    assert isinstance(f, Coroutine)
    assert isinstance(g, AsyncGenerator)
    print("getting ready to run things")
    print(await f)
    async for thing in g:
        print(thing)

asyncio.run(main())

Currently Triggers are only used up to their first event, as they are only used for resuming deferred tasks (which happens on the first event fired). However, we plan to allow DAGs to be launched from triggers in future, which is where multi-event triggers will be more useful.

Looks like classic YAGNI

We can definitely support both pretty trivially. The simplest way would be to just call run and check if the return value is a TriggerEvent (if not, run is a generator, not a plain function). We can change the docs so most people use the plain function variant.

Not important, but I don't think we can do exactly this. If we call run, the return value is either coroutine or asynciterator. If it's coroutine we can await it and obtain the return. If it is asynciterator, we cannot await it but must iterate it. So I think we could must either try-await-except-iterate, or inspect the object type and handle appropriately.

Either way, I am OK with evolving the interface to support return and update the docs as you suggested.

But I guess it remains for us to decide whether to deprecate iterator, and if given iterator, whether to consume up to the return, or just stop at first event, or consume all events....

But I guess it remains for us to decide whether to deprecate iterator, and if given iterator, whether to consume up to the return, or just stop at first event, or consume all events....

@dstandish We also need to think about triggers in the different providers and how to make them b/c; should we add an ugly Airflow version check in each trigger to know if we should yield an event or return it, or we only change the core triggers (airflow/triggers/*) and wait until bumping the minimum version of Airflow in providers to 2.7.x to make the change.

@dstandish We also need to think about triggers in the different providers and how to make them b/c; should we add an ugly Airflow version check in each trigger to know if we should yield an event or return it, or we only change the core triggers (airflow/triggers/*) and wait until bumping the minimum version of Airflow in providers to 2.7.x to make the change

Yes, valid concern, future compat is something we must deal with pretty regularly though, nothing terribly unusal here. I.e. it's common that we add some new interface or feature in core and we have to wait for min airflow version of provider to catch up before using it. Related note, I actually have a PR on the shelf to add some amount of automation / alerting to help us with this kind of thing... i.e. reminding us to remove deprecations, or to remove future compat checks, or to update code at X min-airflow-version. I had to drop it to focus on AIP-52 but hope to pick it up again soon.

If it's coroutine we can await it and obtain the return. If it is asynciterator, we cannot await it but must iterate it.

An isawaitable check would cover this pretty easily.

An isawaitable check would cover this pretty easily.

nice

I experiemented this a bit. It seems like the most difficult part is actually to convince Mypy run() can either be implemented as an async generator function or a coroutine function. I want to believe there’s a way but have not been able to find it.

I experiemented this a bit. It seems like the most difficult part is actually to convince Mypy run() can either be implemented as an async generator function or a coroutine function. I want to believe there’s a way but have not been able to find it.

@uranusjr could you check my last commit? I tested it with both types, and it appears to work, and it passes the Mypy pre-commit check

The static checks fail.

Even the generic type didn't fix it:

T = TypeVar("T")

class BaseTrigger(abc.ABC, LoggingMixin, Generic[T]):
    async def run(self) -> T:
        pass

class Trigger1(BaseTrigger[TriggerEvent]):
    async def run(self) -> TriggerEvent:
        return event

class Trigger2(BaseTrigger[AsyncIterator[TriggerEvent]]):
    async def run(self) -> AsyncIterator[TriggerEvent]:
        yield event

This is the same problem I had. For some reason Mypy really don’t like a subclass only implement one of the unioned return types. Probably a covariance issue, but I’m not sure what the right fix is.