Plugins, Listeners, and Notifiers
The extension points for UI additions, lifecycle hooks, and provider-backed notifications in modern Airflow.
Three requests land on the bookshop platform team in the same week, and they sound similar enough to be confused for one. The first: analysts want a “View in warehouse” button on the task page so they can jump straight from a failed load to the table it writes. The second: the data-governance group wants an audit trail — every task outcome, from every DAG, shipped to their lineage system, and they very much do not want to open a code review on forty DAGs to get it. The third: on-call wants a Slack ping when the nightly bookshop_daily run fails, formatted, with a link, not a wall of log text.
Those are three different jobs, and Airflow has three different tools for them — which is exactly why they get muddled. A plugin extends the platform itself: it adds things to Airflow’s UI and template surface. A listener observes the platform: it fires on lifecycle events across every DAG without touching any of them. A notifier reacts to an outcome: it’s a reusable, parameterized alert you attach where you want it. Get the mapping wrong and you end up hand-rolling Slack calls in a listener, or copy-pasting a callback into forty DAGs, or reaching for a plugin when a three-line notifier would do. This chapter is about keeping them straight.
A plugin extends Airflow, not your pipeline
An Airflow plugin is a class that subclasses AirflowPlugin and declares what it contributes. You drop it in the project’s plugins/ folder, Airflow discovers it at startup, and its contributions become part of the platform. The mental model that keeps you out of trouble: a plugin adds capabilities to Airflow, never business logic to a DAG. If the code you’re about to write does work when a DAG runs, it belongs in an operator or a @task. A plugin is for the machinery around the work — the buttons, the macros, the registrations.
Here’s the shape. The class body is almost entirely a manifest — a list of named attributes Airflow reads to wire up each contribution type:
from airflow.plugins_manager import AirflowPlugin
from airflow.sdk import BaseOperatorLink
class BookshopWarehouseLink(BaseOperatorLink):
name = "View in warehouse"
def get_link(self, operator, *, ti_key):
table = getattr(operator, "table", "unknown")
return f"https://warehouse.bookshop.dev/tables/{table}"
def days_since_release(release_date):
from datetime import date, datetime
d = datetime.fromisoformat(release_date).date()
return (date.today() - d).days
class BookshopPlugin(AirflowPlugin):
name = "bookshop"
operator_extra_links = [BookshopWarehouseLink()]
macros = [days_since_release]Two contribution types are shown, and they’re the two that survive cleanly into Airflow 3.x. Operator extra links add a button to the task-instance page — the “View in warehouse” the analysts asked for. get_link receives the operator instance and a ti_key identifying the exact run, so the URL can deep-link to that run’s table rather than a generic dashboard. (You can attach an extra link directly on an operator via operator_extra_links, as the custom-operators chapter did; registering it through a plugin instead lets you bolt a link onto operators you don’t own — a built-in SQLExecuteQueryOperator, say — by matching on operators = [SQLExecuteQueryOperator] in the link class.)
Macros register callables into the Jinja render context, so any templated field in any DAG can call {{ macros.bookshop.days_since_release(params.release) }}. This is the right home for a formatting or lookup helper you want available everywhere — a fiscal-quarter calculator, a partition-path builder — without importing it into each DAG file. It’s genuinely global, which is the point and also the caution: a macro is shared surface, so keep it pure and cheap.
Beyond those two, the plugin manifest can also register timetables (custom schedule classes, so a DAG can write schedule=BookshopTradingCalendar()), listeners (the subject of the next section — a plugin is how a listener module gets loaded), and a handful of others. Airflow discovers plugins two ways: modules dropped in plugins/, and packages that advertise an airflow.plugins entry point. For a single Astro project, plugins/ is the whole story — astro dev start bakes the folder into the image and the plugin is live after a restart. You can confirm what got loaded at Admin → Plugins in the UI, which lists every registered contribution and is the first place to look when your extra link stubbornly refuses to appear.
The honest part: UI plugins changed in Airflow 3.x
If you’ve written Airflow plugins before, you’re waiting for the part where I show you flask_blueprints, appbuilder_views, and appbuilder_menu_items — the attributes that let a 2.x plugin mount a whole custom page, a Flask view, or a top-nav menu entry into the web UI. Here’s the thing you need to know before you port anything forward: Airflow 3 replaced the Flask webserver with a React UI served by a new FastAPI api-server. The classic Flask-blueprint UI-plugin path is a casualty of that change.
Concretely: a 2.x plugin that registered flask_blueprints or appbuilder_views to render a custom HTML page does not light up in the 3.x React UI the way it used to. The AirflowPlugin attributes still exist — Airflow won’t reject the class — but the Flask app that used to host those blueprints is gone, so a blueprint has no webserver to attach to. Menu items and full custom pages that were trivial in 2.x are the affected surface. If your platform leaned on a bespoke admin page mounted as a plugin, that’s the piece to re-plan, not just re-import.
What still works, and works well, is everything that isn’t a Flask view: operator extra links, macros, timetables, and listeners. Those are wired through Airflow’s own machinery, not through the webserver, so they carry forward unchanged — the extra link renders as a button in the React UI, the macro lands in the render context, the listener fires. The forward-looking replacement for “a whole custom page in the UI” is the plugin external_views / React-based app-plugin surface the 3.x UI exposes (you point the UI at your own app rather than injecting server-rendered HTML), but if all you need is a button and a macro — which is most of what plugins are actually used for — you don’t touch any of that. The takeaway for a migration: audit your plugins, sort them into “extra-link-or-macro” (keep) versus “renders a page” (rework), and don’t assume a green import means the page still shows up.
A listener observes every run, without touching a DAG
The governance team’s request — every task outcome shipped to their lineage system, no per-DAG changes — is the textbook case for the listener API. A listener is a module of functions that Airflow calls on lifecycle events: a task instance starts, succeeds, or fails; a DAG run succeeds or fails. Crucially, it’s ambient. You register it once, and it fires for every eligible event in the whole deployment. No DAG author has to know it exists, and no DAG file changes. That’s precisely the property you want for cross-cutting concerns — audit, metrics, lineage — where the alternative is a callback duplicated into every DAG that then rots the moment someone forgets to add it.
Listeners are built on pluggy, the same hook framework pytest uses. You write plain functions, decorate them with @hookimpl, and give them the exact names and signatures Airflow’s hook specification defines. The names are the contract — Airflow calls on_task_instance_success because that’s what the spec names it, so a typo means your hook silently never fires. The signatures Airflow 3.x hands you are built around the Task SDK’s runtime objects:
# plugins/bookshop_listener/audit.py
from airflow.listeners import hookimpl
def _emit(event_type, payload):
"""Ship one audit event. Cheap and non-blocking — see the note below."""
import json, urllib.request
req = urllib.request.Request(
"https://lineage.bookshop.dev/events",
data=json.dumps({"event": event_type, **payload}).encode(),
headers={"Content-Type": "application/json"},
method="POST",
)
urllib.request.urlopen(req, timeout=2)
@hookimpl
def on_task_instance_running(previous_state, task_instance, session=None):
_emit("task_running", {
"dag_id": task_instance.dag_id,
"task_id": task_instance.task_id,
"run_id": task_instance.run_id,
"try_number": task_instance.try_number,
"operator": task_instance.operator,
})
@hookimpl
def on_task_instance_success(previous_state, task_instance, session=None):
_emit("task_success", {
"dag_id": task_instance.dag_id,
"task_id": task_instance.task_id,
"run_id": task_instance.run_id,
"duration": task_instance.duration,
})
@hookimpl
def on_task_instance_failed(previous_state, task_instance, error=None, session=None):
_emit("task_failed", {
"dag_id": task_instance.dag_id,
"task_id": task_instance.task_id,
"run_id": task_instance.run_id,
"try_number": task_instance.try_number,
"error": str(error),
})Three task-level hooks, and their DAG-run siblings work the same way:
@hookimpl
def on_dag_run_success(dag_run, msg=None):
_emit("dag_success", {
"dag_id": dag_run.dag_id,
"run_id": dag_run.run_id,
"start": str(dag_run.start_date),
"end": str(dag_run.end_date),
})
@hookimpl
def on_dag_run_failed(dag_run, msg=None):
_emit("dag_failed", {"dag_id": dag_run.dag_id, "run_id": dag_run.run_id})A listener module isn’t discovered on its own — you register it through a plugin, using the listeners attribute you met earlier. The value is the module, not a function; Airflow scans it for every @hookimpl it can find:
# plugins/bookshop_listener/__init__.py
from airflow.plugins_manager import AirflowPlugin
from bookshop_listener import audit
class BookshopAuditPlugin(AirflowPlugin):
name = "bookshop_audit"
listeners = [audit]That’s the entire wiring. Restart the project, run any DAG, and audit events start flowing for tasks whose author has never heard of your listener. The on_task_instance_failed hook receives the actual exception as error, which makes it a natural spot to compute failure metrics or route a structured incident record — richer than parsing logs after the fact.
Now the sharp edge, because listeners have a nasty one. A listener runs synchronously, in the process that raised the event — task-instance hooks execute in the worker as part of the task’s own lifecycle, and DAG-run hooks fire in scheduler-adjacent context. That means a slow listener slows down the thing it’s observing, and a listener that raises can disrupt it. The urllib.urlopen(..., timeout=2) above is deliberate: a short, bounded timeout so a flaky lineage endpoint costs every task two seconds at most instead of hanging it. In production you go further — the body of _emit should hand the event to something that delivers it out-of-band (a local queue, a StatsD line, a fire-and-forget append) rather than making a blocking HTTP call inline. A listener is the right place to decide an event happened; it is the wrong place to do slow work in response. The framing to carry: a listener is on the critical path of every run, so it must stay cheap the way a hot loop must stay cheap.
The full hook set, and the lifecycle pair worth knowing
The five hooks above — three task, two DAG-run — are the ones you reach for daily, but they aren’t the whole specification. The hook set Airflow 3.x exposes rounds out the lifecycle: on_task_instance_running/_success/_failed on the task side, on_dag_run_running/_success/_failed on the run side (note the running variants, which give you a fired-at-start event on both levels), and a process-lifecycle pair that fires not on pipeline events but on the component hosting the listener: on_starting and before_stopping.
Those two are the ones people miss, and they’re the answer to “where do I open and close the thing my hooks talk to.” on_starting(component) fires once as the hosting component (scheduler, or a worker’s task-running process) boots and before any task or DAG hook runs; before_stopping(component) fires once as it shuts down. That’s the natural bracket for a client you don’t want to reconstruct on every event — a StatsD socket, a batched HTTP session, a lineage-SDK handle. Open it in on_starting, drain and close it in before_stopping, and the hot task hooks just enqueue against something already warm:
# plugins/bookshop_listener/audit.py (lifecycle additions)
from airflow.listeners import hookimpl
_client = {}
@hookimpl
def on_starting(component):
"""Fires once as the hosting process boots, before any task hook."""
import queue, threading
q = _client["queue"] = queue.Queue(maxsize=10_000)
_client["stop"] = threading.Event()
def _drain():
while not (_client["stop"].is_set() and q.empty()):
try:
_emit(*q.get(timeout=0.5))
except Exception:
pass # never let the background sender die
t = _client["thread"] = threading.Thread(target=_drain, daemon=True)
t.start()
@hookimpl
def before_stopping(component):
"""Fires once on shutdown — flush what's still queued."""
_client["stop"].set()
_client["thread"].join(timeout=5)Now the task hooks can call _client["queue"].put(("task_success", payload)) instead of a blocking urlopen, and the two-second timeout stops being every task’s tax — the emit happens off the critical path on the drain thread, and before_stopping guarantees the tail gets flushed rather than dropped when the worker recycles. component is passed so a single listener module can behave differently depending on which process loaded it (a scheduler-side listener that watches DAG-run events versus a worker-side one that watches task instances). It’s the same registration as before — the plugin’s listeners = [audit] picks up every @hookimpl in the module, lifecycle hooks included, with no extra wiring.
One more property worth internalizing: a listener sees events across DAGs but doesn’t belong to any of them, which is its whole value and also why it’s the wrong tool for “alert me when this one pipeline fails.” For that you want the outcome to be attached to the DAG, declaratively — which is a notifier.
The asset hooks nobody uses
The hooks above cover task instances and DAG runs — the two things everyone reaches for. But Airflow 3 ships four more that are exactly what a lineage or governance listener wants, and they’re almost never mentioned:
from airflow.listeners import hookimpl
class AssetAudit:
@hookimpl
def on_asset_created(self, asset):
audit.record("asset.created", uri=asset.uri, name=asset.name)
@hookimpl
def on_asset_changed(self, asset):
audit.record("asset.changed", uri=asset.uri)
@hookimpl
def on_asset_event_emitted(self, asset_event):
# fires whenever a task marks an asset updated — the moment a
# downstream DAG becomes eligible to run
audit.record("asset.updated", uri=asset_event.asset.uri,
extra=asset_event.extra)(There’s an on_asset_alias_created too, plus on_existing_dag_import_error / on_new_dag_import_error for the parse-failure side.)
Why these matter more than they look: chapter 04 made assets the unit of data-aware scheduling, which means an asset event is the closest thing Airflow has to a semantic statement — “this dataset is now current.” A task-instance listener tells you a Python function succeeded; an asset listener tells you a table became fresh. If you’re pushing lineage to a catalog, feeding a data-observability tool, or answering “when did orders last actually update?”, these are the hooks that carry the meaning. The task-level hooks carry only the mechanics.
A notifier is a reusable, testable alert
On-call’s Slack request could be satisfied with a raw callback — on_failure_callback has always accepted any callable, and you could write a function that builds a Slack payload and posts it. People do this, and it’s how notifications rot: the function lives in one DAG, someone copies it to a second with the channel hardcoded, the token handling drifts, and now there’s no single place to fix the message format. A notifier is Airflow’s answer to that decay. It’s a small class — a subclass of BaseNotifier — that packages “how to send this kind of alert” into a reusable, parameterized, testable object you attach wherever you want it.
The contract is one method, notify, plus whatever constructor arguments parameterize the message. BaseNotifier gives you two things a raw callback lacks: it templates your fields (list them in template_fields and they Jinja-render against the run context, exactly like an operator), and it standardizes the call signature so the same notifier plugs into a callback slot or the declarative notifiers list.
from airflow.sdk import BaseNotifier
from airflow.providers.slack.hooks.slack_webhook import SlackWebhookHook
class BookshopSlackNotifier(BaseNotifier):
template_fields = ("channel", "dag_id", "run_id")
def __init__(self, slack_conn_id, channel, dag_id="{{ dag.dag_id }}",
run_id="{{ run_id }}"):
self.slack_conn_id = slack_conn_id
self.channel = channel
self.dag_id = dag_id
self.run_id = run_id
def notify(self, context):
ti = context["task_instance"]
text = (
f":rotating_light: *{self.dag_id}* failed\n"
f"task `{ti.task_id}` · run `{self.run_id}`\n"
f"<https://airflow.bookshop.dev/dags/{self.dag_id}/runs/{self.run_id}|open run>"
)
SlackWebhookHook(slack_webhook_conn_id=self.slack_conn_id).send(
text=text, channel=self.channel
)Because channel and the id fields are in template_fields, a caller can pass templated values and they resolve per-run; because notify takes the standard context, the object drops straight into any callback slot. You attach it declaratively:
from airflow.sdk import DAG
with DAG(
"bookshop_daily",
on_failure_callback=[
BookshopSlackNotifier(slack_conn_id="slack_default", channel="#data-alerts"),
],
) as dag:
...Note the failure notifier is set on the DAG, so it fires once for a failed DAG run — the right granularity for “the nightly pipeline broke.” Set a notifier on a task’s on_failure_callback instead and it fires per failing task, which you want for a noisy-but-isolated task and not for a whole-pipeline alert. Both slots accept a list, so you can stack a Slack ping and a PagerDuty escalation on the same event without either knowing about the other. There’s also on_success_callback for the “the backfill finally finished, tell the analysts” case, and the same notifier objects work there.
You usually don’t write the notifier — the provider did
The custom class above is worth understanding, but for the common targets you don’t write one at all: the provider packages ship notifiers, and you configure rather than code. Slack and SMTP are the two you’ll reach for most:
from airflow.providers.slack.notifications.slack import SlackNotifier
from airflow.providers.smtp.notifications.smtp import SmtpNotifier
with DAG(
"bookshop_daily",
on_failure_callback=[
SlackNotifier(
slack_conn_id="slack_default",
text="bookshop_daily failed — run {{ run_id }}",
channel="#data-alerts",
),
SmtpNotifier(
from_email="[email protected]",
to="[email protected]",
subject="[Airflow] {{ dag.dag_id }} failed",
html_content="Run {{ run_id }} failed. See the UI for the failed task.",
),
],
) as dag:
...SlackNotifier reads its token from a Slack connection (slack_conn_id) so no secret sits in the DAG. SmtpNotifier reads the mail server from an SMTP connection — by default the connection id smtp_default, which you define once (host, port, login, password, and TLS flags in extra) either in the UI or as an AIRFLOW_CONN_SMTP_DEFAULT URI. This is the Airflow-3 replacement for the old [smtp] section in airflow.cfg and the airflow.email module: email delivery is a provider now, configured through a connection, and SmtpNotifier is the idiomatic way to send it. Both notifiers template their message fields, so {{ run_id }} and {{ dag.dag_id }} resolve against the failing run.
The reason to prefer a notifier over a hand-written callback comes down to four properties that a bare function doesn’t give you for free. It’s reusable — one object, attached to many DAGs, one place to change the format. It’s parameterized — channel, recipient, and subject are constructor arguments, not hardcoded strings, so the same class serves #data-alerts and #ml-alerts. It’s templated — message fields resolve against the run context through the same Jinja machinery as operators. And it’s testable — a notifier is a plain object, so you can construct one, call notify with a fake context, and assert on the payload it would have sent, with no scheduler and no live Slack workspace:
def test_slack_notifier_builds_link(mocker):
sent = mocker.patch(
"airflow.providers.slack.hooks.slack_webhook.SlackWebhookHook.send"
)
n = BookshopSlackNotifier(slack_conn_id="slack_default", channel="#data-alerts")
n.dag_id, n.run_id = "bookshop_daily", "manual__2026-06-29"
n.notify({"task_instance": mocker.Mock(task_id="load_orders")})
assert "bookshop_daily" in sent.call_args.kwargs["text"]
assert "manual__2026-06-29" in sent.call_args.kwargs["text"]That test is the whole argument in code: the alert format is now something you can pin down in CI instead of discovering is broken the night it’s supposed to fire.
When to reach for which
The three tools blur together because all three can, technically, “do something when a task finishes.” The distinction that actually decides is scope and intent:
- Plugin — when you’re extending the platform. A button on the task page, a macro every DAG can call, a custom timetable, a connection type. The test: does this add a capability to Airflow itself, independent of any one pipeline’s logic? If yes, it’s a plugin. (And in 3.x, if that capability is a full custom page, budget for the React-UI rework — the Flask-blueprint path is gone.)
- Listener — when you need to observe every run without editing DAGs. Audit, lineage, deployment-wide metrics — anything that must fire for pipelines whose authors shouldn’t have to opt in. Ambient by design; keep it cheap because it’s on the critical path.
- Notifier — when you want a reusable alert on a specific outcome. Slack on failure, email on success, PagerDuty on a critical DAG. It’s attached where you want it, parameterized per use, and testable. Prefer it over a raw callback the moment the same alert shows up in a second DAG.
- Raw callback — the honest fourth option. When the reaction is genuinely one-off and lives in exactly one DAG, a plain
on_failure_callbackfunction is fine. The instant it wants to be reused, promote it to a notifier.
Examples in this chapter are docs-checked against the series’ stated versions, but they were not executed in this repository unless a companion project explicitly says so.
Final thoughts
The reason these three get conflated is that they answer the same-sounding question — “run some code around a task” — from three different altitudes, and picking the wrong altitude is where the pain comes from. A notifier crammed into a listener means every task in the deployment pays for one pipeline’s Slack call. A per-DAG callback copy-pasted for a governance requirement means the one DAG that skipped it is the one the auditors find. A Flask-blueprint UI plugin ported straight into 3.x means a page that silently never renders. The fix is boring and reliable: match the tool to the scope. Extend the platform with a plugin, observe the platform with a listener, alert on an outcome with a notifier, and reach for a raw callback only when the reaction truly lives in one place. Do that and each of the three earns its keep — the button saves on-call three clicks, the listener feeds governance without a single DAG review, and the alert is something you can test in CI instead of hoping it fires. Keep the listener and notifier paths cheap, respect that Airflow 3’s React UI moved the goalposts for UI plugins, and these extension points stop being a source of surprise and become the quiet infrastructure they’re supposed to be.
Comments