Description

Currently, operations toward a non-declared outlet to outlet_events is simply ignored:

@task
def emit_alias_event():
    context = get_current_context()
    context["outlet_events"][AssetAlias("a")].add(Asset("new"))
    # Line above does nothing since AssetAlias("a") is not an outlet.

The same also applies to yield Metadata.

This is a bit confusing to the user since they did something, and reasonably expect it’s an Airflow bug if nothing seems to happen.

I can think of a couple of ways to resolve this:

Automatically add outlets when you do this

If you access something, it automatically becomes an outlet. However, Airflow would not be able to detect such a dynamic outlet in many other places, such as the dag dependency graph (which is generated statically). Would this only shift the confusion elsewhere?

Raise a KeyError

Or a more detailed error with a hint. More explicit, but would it be awkward for certain workflows?

Use case/motivation

No response

Related issues

No response

Are you willing to submit a PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct