# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
"""
Per-scope emission policy resolution for the OpenLineage provider.
The ``emission_policy`` Airflow configuration option (``[openlineage]`` section)
accepts a JSON array of rule objects. Each rule has the shape::
{
"scope": {<scope keys>}, # required, may be empty for "global"
"match_mode": "exact" | "regex", # optional, default "exact"
"controls": {<control flags>}, # required, must be non-empty
"locked": true | false # optional, default false
}
Top-level keys outside the four above are rejected with a WARNING and the rule
is skipped.
Scope keys (all optional inside ``scope``):
- ``operator`` — fully-qualified operator class name; matches task events for that
operator type.
- ``dag_id`` — matches task and/or dag run events for all tasks / dag runs belonging
to the named DAG. Use ``emit_task_events`` / ``emit_dag_events`` to target one
event type selectively.
- ``task_id`` — only valid alongside ``dag_id``; targets a specific task.
- ``hook``: fully-qualified class name of the object that reported a hook-collected
dataset (usually a hook). Can be added to any of the scopes above, and narrows the
rule to datasets reported by that hook. Only ``exclude_datasets`` and ``hook_lineage``
are valid controls for a hook-scoped rule.
- *(empty ``scope: {}``)* — global default override; applies to every task and DAG event.
``operator`` cannot be combined with ``dag_id`` or ``task_id`` (scope is one of:
global, operator-only, dag-only, dag+task, each optionally with ``hook``). ``task_id``
without ``dag_id`` and ``operator`` combined with ``emit_dag_events`` (or ``task_id``
combined with ``emit_dag_events``) are also rejected with a WARNING.
``match_mode`` lives at the top level: ``"exact"`` (default) or ``"regex"``.
When set to ``"regex"``, every value inside ``scope`` (``dag_id``, ``task_id``,
``operator``, ``hook``) is treated as a ``re.fullmatch`` pattern.
Control flag keys (all optional inside ``controls``; the dict must be non-empty):
- ``emit`` — shorthand: disable ALL OpenLineage events in scope (both task and dag
run events). Default: ``true``.
- ``emit_task_events`` — disable task-level events only; takes precedence over
``emit`` for task event decisions. Default: ``true``.
- ``emit_dag_events`` — disable dag-run-level events only; takes precedence over
``emit`` for dag event decisions. Default: ``true``.
- ``extract_operator_metadata`` — whether to run operator-specific extractor-based metadata
collection. When ``true``, the extractor manager calls the operator's OpenLineage extractor
(if registered), which may produce dataset inputs/outputs, job facets, run facets, and
other operator-specific metadata. When ``false``, the entire extraction pipeline is skipped
and a minimal event is emitted. Only meaningful for task events. Default: ``true``.
- ``include_source_code`` — whether to include operator source code in the
``SourceCodeJobFacet`` for Python and Bash operators. Only meaningful for task events
when ``extract_operator_metadata`` is also ``true``. Default: ``true``.
- ``hook_lineage`` — whether to use ``HookLineageCollector`` as a fallback when the
extractor finds no inputs/outputs. **Has no effect when ``extract_operator_metadata``
is ``false``** — the entire extraction pipeline (including hook lineage) is skipped.
Only meaningful for task events. Default: ``true``. In a hook-scoped rule it instead
keeps (``true``) or drops (``false``) every dataset reported by the matching hook, and it
cannot be combined with ``exclude_datasets`` in the same rule.
- ``include_full_task_info`` — whether to include the full serialized operator state
in the ``AirflowRunFacet``. When ``false`` (default), only a curated subset of
task attributes is sent. Only meaningful for task events. Default: ``false``.
- ``exclude_datasets``: list of ``re.fullmatch`` patterns (independent of ``match_mode``)
matched against ``"<namespace>/<name>"`` of each OpenLineage dataset (without a doubled ``/``
when the name starts with one); matching datasets
are dropped from task events and from the SQL hook lineage events of the task. Applies
to extractor, hook-collected and manually declared lineage. Default: ``[]``.
Unlike the boolean controls, ``exclude_datasets`` is resolved per dataset, and the most
specific matching rule's list **replaces** broader lists instead of merging with them, so a
narrower rule can shorten or clear (``[]``) a broader exclusion.
``locked: true`` is an admin floor lock that prevents per-Dag / per-task authoring
overrides from changing the field(s) carried by this rule's ``controls`` dict.
Flag hierarchy
~~~~~~~~~~~~~~
Flags are not fully independent — some only take effect when a higher-level flag is
enabled:
- ``extract_operator_metadata: false`` skips the **entire** operator extraction pipeline.
When this is ``false``, both ``include_source_code`` and ``hook_lineage`` have **no
effect** regardless of their values, because the code paths they control are never
reached.
- ``include_source_code`` only applies inside Python and Bash operator extractors; setting
it to ``false`` on other operator types is a no-op.
Priority for **task events** (most specific tier wins; within a tier, last matching
rule wins):
1. ``dag_id`` + ``task_id``
2. ``dag_id``
3. ``operator``
4. Global (no match keys)
5. Built-in defaults
For ``emit`` at the task level: ``emit_task_events`` beats ``emit`` within the same
rule.
For ``exclude_datasets`` on a dataset reported by a hook, each tier above is preceded by
its hook-scoped variant: ``dag_id`` + ``task_id`` + ``hook``, ``dag_id`` + ``task_id``,
``dag_id`` + ``hook``, ``dag_id``, ``operator`` + ``hook``, ``operator``, ``hook``,
global. Datasets from other sources only see the tiers without ``hook``.
Priority for **DAG run events**:
1. ``dag_id`` rule (using ``emit_dag_events`` or ``emit``)
2. Global (no match keys)
3. Built-in defaults
For ``emit`` at the dag level: ``emit_dag_events`` beats ``emit`` within the same
rule.
Legacy config translation
-------------------------
Any active legacy config option (``disabled_for_operators``, ``disable_source_code``,
``include_full_task_info``, ``selective_enable``) is **always** translated into equivalent
``emission_policy`` rules, regardless of whether ``emission_policy`` itself is set. The
translated rules are prepended before any user-provided rules, so user rules win within
each priority tier (last-wins). A ``DeprecationWarning`` is issued listing every translated
option — silence it by migrating those options into ``emission_policy`` exclusively.
When no legacy option is active and ``emission_policy`` is empty, the resolver returns
the built-in defaults with no warnings.
Audit logging
-------------
Every resolved field that is non-default — whether from a rule or from a legacy
translation — is logged at INFO level, identifying the field, the event context,
and the exact rule that caused the change.
"""
from __future__ import annotations
import logging
import re
import warnings
from collections.abc import Callable, Sequence
from dataclasses import dataclass, field as dataclass_field, replace
from typing import TYPE_CHECKING, TypeVar
from airflow.exceptions import AirflowProviderDeprecationWarning
from airflow.providers.common.compat.sdk import Param
from airflow.providers.openlineage import conf as _ol_conf
if TYPE_CHECKING:
from openlineage.client.event_v2 import Dataset
from airflow.providers.openlineage.utils.utils import AnyOperator
[docs]
DatasetT = TypeVar("DatasetT", bound=Dataset)
[docs]
log = logging.getLogger(__name__)
# A validated ``controls`` value: bool for the flags, list of patterns for ``exclude_datasets``.
[docs]
ControlValue = bool | list[str]
# Control flag names — the keys that may appear inside a rule's ``controls`` dict.
[docs]
EMIT_TASK_EVENTS = "emit_task_events"
[docs]
EMIT_DAG_EVENTS = "emit_dag_events"
[docs]
INCLUDE_SOURCE_CODE = "include_source_code"
[docs]
HOOK_LINEAGE = "hook_lineage"
[docs]
INCLUDE_FULL_TASK_INFO = "include_full_task_info"
[docs]
EXCLUDE_DATASETS = "exclude_datasets"
# Scope keys — the keys that may appear inside a rule's ``scope`` dict.
[docs]
SCOPE_DAG_ID = "dag_id"
[docs]
SCOPE_TASK_ID = "task_id"
[docs]
SCOPE_OPERATOR = "operator"
# Top-level rule keys.
[docs]
RULE_CONTROLS = "controls"
[docs]
RULE_MATCH_MODE = "match_mode"
# Param names used by the authoring API to store flow-control flags on operator / DAG
# objects. Kept in this module (rather than the api module) so the resolver can read
# them without a circular import.
[docs]
OL_EMISSION_POLICY_PARAM = "_openlineage_emission_policy"
# Schema enforcement: only these keys may appear at each level. Unknown keys cause
# the rule to be skipped with a WARNING (catches typos like {"scope": {"dgg_id": "x"}}).
_ALLOWED_TOP_LEVEL_KEYS: frozenset[str] = frozenset({RULE_SCOPE, RULE_MATCH_MODE, RULE_CONTROLS, RULE_LOCKED})
_ALLOWED_SCOPE_KEYS: frozenset[str] = frozenset({SCOPE_DAG_ID, SCOPE_TASK_ID, SCOPE_OPERATOR, SCOPE_HOOK})
_ALLOWED_CONTROL_KEYS: frozenset[str] = frozenset(
{
EMIT,
EMIT_TASK_EVENTS,
EMIT_DAG_EVENTS,
EXTRACT_OPERATOR_METADATA,
INCLUDE_SOURCE_CODE,
HOOK_LINEAGE,
INCLUDE_FULL_TASK_INFO,
EXCLUDE_DATASETS,
}
)
# The hook is only known per dataset, so a hook-scoped rule may only carry controls that
# act on individual datasets.
_HOOK_SCOPE_CONTROL_KEYS: frozenset[str] = frozenset({EXCLUDE_DATASETS, HOOK_LINEAGE})
# Fields scanned for ``locked: true`` on task-event resolution.
# ``emit`` is intentionally absent: emit-locking is driven by either ``emit`` or
# ``emit_task_events`` appearing in a locked rule (handled separately in
# :func:`_compute_locked_task_fields`), so listing it here would double-count.
_LOCKABLE_TASK_FIELDS: tuple[str, ...] = (
EXTRACT_OPERATOR_METADATA,
INCLUDE_SOURCE_CODE,
HOOK_LINEAGE,
INCLUDE_FULL_TASK_INFO,
)
# Authoring flag classification: which keys are relevant for task-level vs DAG-run
# resolution. Used by the authoring API to split a single call into the two stored
# param dicts.
_TASK_FLAG_KEYS: frozenset[str] = frozenset(
{
EMIT,
EMIT_TASK_EVENTS,
EXTRACT_OPERATOR_METADATA,
INCLUDE_SOURCE_CODE,
HOOK_LINEAGE,
INCLUDE_FULL_TASK_INFO,
EXCLUDE_DATASETS,
}
)
_DAG_FLAG_KEYS: frozenset[str] = frozenset({EMIT, EMIT_DAG_EVENTS})
@dataclass(frozen=True)
[docs]
class EmissionPolicy:
"""Resolved emission policy for a specific event context."""
[docs]
include_source_code: bool
[docs]
include_full_task_info: bool
# Rules hold dicts, so hashing the filter would fail; equal policies still hash equally without it.
[docs]
dataset_filter: DatasetFilter = dataclass_field(default_factory=lambda: DatasetFilter(), hash=False)
@classmethod
[docs]
def defaults(cls) -> EmissionPolicy:
"""Return the default policy (all controls at their built-in defaults)."""
return cls(
emit=True,
extract_operator_metadata=True,
include_source_code=True,
hook_lineage=True,
include_full_task_info=False,
)
@dataclass(frozen=True)
[docs]
class Rule:
"""
A single, fully-validated ``emission_policy`` rule.
Instances are produced exclusively by :func:`_parse_rule`, so holding a ``Rule``
is itself the "this passed schema validation" guarantee — downstream resolution
never re-validates. ``scope`` / ``controls`` are the raw (validated) sub-dicts.
"""
[docs]
controls: dict[str, ControlValue]
[docs]
match_mode: str = "exact"
@property
[docs]
def has_dag(self) -> bool:
return SCOPE_DAG_ID in self.scope
@property
[docs]
def has_task(self) -> bool:
return SCOPE_TASK_ID in self.scope
@property
[docs]
def has_op(self) -> bool:
return SCOPE_OPERATOR in self.scope
@property
[docs]
def has_hook(self) -> bool:
return SCOPE_HOOK in self.scope
def _matches(pattern: str, value: str, match_mode: str) -> bool:
"""Return ``True`` if *value* matches *pattern* according to *match_mode*."""
if match_mode == "regex":
return bool(re.fullmatch(pattern, value))
return pattern == value
[docs]
def get_hook_class_name(hook: object | None) -> str | None:
"""Return the fully-qualified class name that the ``hook`` scope key matches, if any."""
if hook is None:
return None
hook_type = type(hook)
return f"{hook_type.__module__}.{hook_type.__name__}"
def _get_dataset_identity(dataset: Dataset) -> str:
"""Return the ``"<namespace>/<name>"`` string that ``exclude_datasets`` patterns match."""
# File and other path-based datasets have names starting with "/"; joining them with another
# "/" would make ``file://host/path`` match as ``file://host//path``.
separator = "" if dataset.name.startswith("/") else "/"
return f"{dataset.namespace}{separator}{dataset.name}"
def _get_rule_exclude_patterns(rule: Rule) -> tuple[str, ...] | None:
"""Return the ``exclude_datasets`` patterns *rule* sets, or ``None`` if it sets none."""
patterns = rule.controls.get(EXCLUDE_DATASETS)
if isinstance(patterns, list):
return tuple(patterns)
if rule.has_hook and HOOK_LINEAGE in rule.controls:
return () if rule.controls[HOOK_LINEAGE] else (".*",)
return None
@dataclass(frozen=True)
[docs]
class DatasetFilter:
"""
``exclude_datasets`` resolution for the datasets of one task.
Hook-scoped rules can only be matched once the reporting hook is known, so the winning
patterns are resolved lazily per hook (``None`` for datasets not reported by a hook)
and cached. *tiers* holds only rules that set patterns, most specific first:
dag+task+hook, dag+task, dag+hook, dag, operator+hook, operator, hook, global.
"""
[docs]
tiers: tuple[tuple[Rule, ...], ...] = ()
[docs]
authoring_patterns: tuple[str, ...] | None = None
[docs]
context: str = dataclass_field(default="", compare=False)
_cache: dict[str | None, tuple[tuple[str, ...], tuple[re.Pattern[str], ...]]] = dataclass_field(
default_factory=dict, init=False, compare=False, repr=False
)
def _get_resolved(self, hook: str | None) -> tuple[tuple[str, ...], tuple[re.Pattern[str], ...]]:
if hook not in self._cache:
patterns = self._resolve_patterns(hook)
self._cache[hook] = (patterns, tuple(re.compile(p) for p in patterns))
return self._cache[hook]
[docs]
def get_patterns(self, hook: str | None = None) -> tuple[str, ...]:
"""Return the exclusion patterns that apply to datasets reported by *hook*."""
return self._get_resolved(hook)[0]
[docs]
def is_excluded(self, dataset: Dataset, hook: str | None = None) -> bool:
"""Return ``True`` if *dataset*, reported by *hook* (if any), matches an exclusion pattern."""
compiled = self._get_resolved(hook)[1]
if not compiled:
return False
identity = _get_dataset_identity(dataset)
if any(p.fullmatch(identity) for p in compiled):
log.debug("OpenLineage emission policy: excluding dataset %r for %s.", identity, self.context)
return True
return False
[docs]
def exclude(self, datasets: Sequence[DatasetT], hook: str | None = None) -> list[DatasetT]:
"""Return *datasets* without the ones excluded for *hook*."""
return [d for d in datasets if not self.is_excluded(d, hook)]
def _resolve_patterns(self, hook: str | None) -> tuple[str, ...]:
if not self.tiers and self.authoring_patterns is None:
return ()
tiers = [
[
rule
for rule in tier
if not rule.has_hook
or (hook is not None and _matches(rule.scope[SCOPE_HOOK], hook, rule.match_mode))
]
for tier in self.tiers
]
context = self.context if hook is None else f"datasets reported by '{hook}' in {self.context}"
no_patterns: tuple[str, ...] = ()
patterns, winning_rule = _walk_tiers(tiers, _get_rule_exclude_patterns, EXCLUDE_DATASETS, no_patterns)
if winning_rule is not None:
winning_tier = next(i for i, tier in enumerate(tiers) if winning_rule in tier)
replaced = [rule for tier in tiers[winning_tier + 1 :] for rule in tier]
_audit_log_conf_patterns(patterns, context, winning_rule, replaced)
if self.authoring_patterns is None or self.authoring_patterns == patterns:
return patterns
if any(rule.locked for tier in tiers for rule in tier):
log.warning(
"OpenLineage emission_policy: extend_global_openlineage_emission_policy call for '%s' on %s"
" has no effect; locked by conf rule at value %r",
EXCLUDE_DATASETS,
context,
list(patterns),
)
return patterns
log.info(
"OpenLineage emission policy: '%s' set to %r for %s "
"by manual `extend_global_openlineage_emission_policy` call.",
EXCLUDE_DATASETS,
list(self.authoring_patterns),
context,
)
return self.authoring_patterns
[docs]
def find_invalid_pattern(field_name: str, value: object) -> str | None:
"""
Return an error message if *value* is not a list of valid regex patterns, else ``None``.
Shared by the conf parser (which skips the offending rule) and the authoring API (which
raises), so every pattern that reaches dataset matching has compiled at least once.
"""
if not isinstance(value, list):
return f"'{field_name}' must be a list of strings, got {value!r}"
for pattern in value:
if not isinstance(pattern, str):
return f"'{field_name}' must contain only strings, got {pattern!r}"
try:
re.compile(pattern)
except re.error as exc:
return f"'{field_name}' pattern {pattern!r} is not a valid regex ({exc})"
return None
def _read_param(obj: object, param_name: str) -> dict[str, ControlValue]:
"""Read the flags dict stored at *param_name* on *obj*, or ``{}`` if absent."""
val = getattr(obj, "params", {}).get(param_name)
if val is None:
return {}
if isinstance(val, Param):
val = val.value
return val if isinstance(val, dict) else {}
def _merge_param(obj: object, param_name: str, new_flags: dict[str, ControlValue]) -> None:
"""
Merge *new_flags* into the ``param_name`` param on *obj*, creating it if absent.
The public entry point
(:func:`~airflow.providers.openlineage.api.emission_policy.extend_global_openlineage_emission_policy`)
validates that *obj* is a DAG or task-like object before reaching here, so a missing
``params`` attribute indicates a programmer error (e.g. an incorrectly-mocked object
in tests) — we raise loud rather than silently dropping flags.
"""
params = getattr(obj, "params", None)
if params is None:
raise TypeError(
f"extend_global_openlineage_emission_policy: {type(obj).__name__} instance has no "
"'params' attribute — cannot store flow-control flags. "
"Pass a DAG, an Operator, or an XComArg."
)
existing = params.get(param_name)
if existing is not None:
existing_val = existing.value if isinstance(existing, Param) else existing
if isinstance(existing_val, dict):
new_flags = {**existing_val, **new_flags}
params[param_name] = Param(new_flags)
def _audit_log_conf_field(field: str, value: bool, context: str, source: Rule) -> None:
"""Log the winning conf value for a field whenever any rule touched it."""
log.info(
"OpenLineage emission policy: '%s' %s for %s by %r",
field,
"enabled" if value else "disabled",
context,
source,
)
def _audit_log_conf_patterns(
patterns: tuple[str, ...], context: str, source: Rule, replaced: Sequence[Rule]
) -> None:
"""Log the winning ``exclude_datasets`` list, and the broader rules it replaced, if any."""
if replaced:
log.info(
"OpenLineage emission policy: '%s' set to %r for %s by %r, replacing broader rule(s) %r",
EXCLUDE_DATASETS,
list(patterns),
context,
source,
list(replaced),
)
else:
log.info(
"OpenLineage emission policy: '%s' set to %r for %s by %r",
EXCLUDE_DATASETS,
list(patterns),
context,
source,
)
def _audit_log_authoring_updates(
field_changes: dict[str, bool],
context: str,
) -> None:
"""Audit-log authoring overrides that actually change the resolved policy value."""
for field, new_value in field_changes.items():
log.info(
"OpenLineage emission policy: '%s' %s for %s "
"by manual `extend_global_openlineage_emission_policy` call.",
field,
"enabled" if new_value else "disabled",
context,
)
def _parse_rule(rule: object) -> Rule | None:
"""
Parse and validate a raw rule dict into a :class:`Rule`; warn and return ``None`` otherwise.
A valid rule has exactly these top-level keys: ``scope`` (dict, possibly empty),
``controls`` (non-empty dict), and the optionals ``match_mode`` / ``locked``.
Unknown keys at any level cause the rule to be skipped — that catches user
typos like ``{"scope": {"dgg_id": "x"}}``.
This is the single validation gate: it runs once at config-read time (see
:func:`_parse_user_rules`), so the resolution hot path never re-validates.
"""
if not isinstance(rule, dict):
log.warning("OpenLineage emission_policy entry is not a dict: %r; ignoring.", rule)
return None
unknown_top = set(rule) - _ALLOWED_TOP_LEVEL_KEYS
if unknown_top:
log.warning(
"OpenLineage emission_policy rule has unknown top-level key(s) %s (allowed: %s); ignoring: %r",
sorted(unknown_top),
sorted(_ALLOWED_TOP_LEVEL_KEYS),
rule,
)
return None
if RULE_SCOPE not in rule:
log.warning(
"OpenLineage emission_policy rule is missing required 'scope' key (use 'scope': {} for global); ignoring: %r",
rule,
)
return None
scope = rule[RULE_SCOPE]
if not isinstance(scope, dict):
log.warning(
"OpenLineage emission_policy rule 'scope' must be a dict (use {} for global), got %r; ignoring: %r",
type(scope).__name__,
rule,
)
return None
unknown_scope = set(scope) - _ALLOWED_SCOPE_KEYS
if unknown_scope:
log.warning(
"OpenLineage emission_policy rule 'scope' has unknown key(s) %s (allowed: %s); ignoring: %r",
sorted(unknown_scope),
sorted(_ALLOWED_SCOPE_KEYS),
rule,
)
return None
has_dag = SCOPE_DAG_ID in scope
has_task = SCOPE_TASK_ID in scope
has_op = SCOPE_OPERATOR in scope
if has_task and not has_dag:
log.warning(
"OpenLineage emission_policy rule scope has 'task_id' without 'dag_id'; ignoring: %r",
rule,
)
return None
if has_op and (has_dag or has_task):
log.warning(
"OpenLineage emission_policy rule scope combines 'operator' with 'dag_id'/'task_id' "
"(must be exactly one of: global, operator-only, dag-only, dag+task); ignoring: %r",
rule,
)
return None
if RULE_CONTROLS not in rule:
log.warning(
"OpenLineage emission_policy rule is missing required 'controls' dict; ignoring: %r",
rule,
)
return None
controls = rule[RULE_CONTROLS]
if not isinstance(controls, dict):
log.warning(
"OpenLineage emission_policy rule 'controls' must be a dict, got %r; ignoring: %r",
type(controls).__name__,
rule,
)
return None
if not controls:
log.warning(
"OpenLineage emission_policy rule has empty 'controls' dict "
"(at least one control flag is required); ignoring: %r",
rule,
)
return None
unknown_controls = set(controls) - _ALLOWED_CONTROL_KEYS
if unknown_controls:
log.warning(
"OpenLineage emission_policy rule 'controls' has unknown key(s) %s (allowed: %s); ignoring: %r",
sorted(unknown_controls),
sorted(_ALLOWED_CONTROL_KEYS),
rule,
)
return None
for k, v in controls.items():
if k == EXCLUDE_DATASETS:
invalid = find_invalid_pattern(k, v)
if invalid:
log.warning("OpenLineage emission_policy rule controls.%s; ignoring: %r", invalid, rule)
return None
elif not isinstance(v, bool):
log.warning(
"OpenLineage emission_policy rule 'controls.%s' must be bool, got %r; ignoring: %r",
k,
v,
rule,
)
return None
if has_op and EMIT_DAG_EVENTS in controls:
log.warning(
"OpenLineage emission_policy rule has scope 'operator' with controls 'emit_dag_events' which is meaningless "
"(operators are not associated with DAG run events); ignoring: %r",
rule,
)
return None
if has_task and EMIT_DAG_EVENTS in controls:
log.warning(
"OpenLineage emission_policy rule has scope 'task_id' with controls 'emit_dag_events' which is meaningless "
"(task-specific rules do not target DAG run events); ignoring: %r",
rule,
)
return None
if SCOPE_HOOK in scope:
unsupported = set(controls) - _HOOK_SCOPE_CONTROL_KEYS
if unsupported:
log.warning(
"OpenLineage emission_policy rule has scope 'hook' with controls %s, but hook-scoped rules "
"only support %s; ignoring: %r",
sorted(unsupported),
sorted(_HOOK_SCOPE_CONTROL_KEYS),
rule,
)
return None
if EXCLUDE_DATASETS in controls and HOOK_LINEAGE in controls:
log.warning(
"OpenLineage emission_policy rule has scope 'hook' with both 'exclude_datasets' and "
"'hook_lineage', which both set the datasets dropped for that hook; ignoring: %r",
rule,
)
return None
match_mode = rule.get(RULE_MATCH_MODE, "exact")
if match_mode not in ("exact", "regex"):
log.warning(
"OpenLineage emission_policy rule has invalid 'match_mode' %r (must be 'exact' or 'regex'); ignoring: %r",
match_mode,
rule,
)
return None
if match_mode == "regex":
for key in (SCOPE_DAG_ID, SCOPE_TASK_ID, SCOPE_OPERATOR, SCOPE_HOOK):
if key in scope:
try:
re.compile(scope[key])
except re.error as exc:
log.warning(
"OpenLineage emission_policy rule scope.'%s' pattern %r is not a valid regex (%s); ignoring: %r",
key,
scope[key],
exc,
rule,
)
return None
if RULE_LOCKED in rule and not isinstance(rule[RULE_LOCKED], bool):
log.warning(
"OpenLineage emission_policy rule 'locked' must be bool, got %r; ignoring: %r",
rule[RULE_LOCKED],
rule,
)
return None
# Under the nested schema every allowed control key is lockable (validation already
# requires non-empty controls with keys from the allowed set), so there is no
# "lock with no lockable field" warning case to raise.
return Rule(
scope=rule[RULE_SCOPE],
controls=rule[RULE_CONTROLS],
match_mode=match_mode,
locked=rule.get(RULE_LOCKED, False),
)
@_ol_conf.cache
def _parse_user_rules() -> tuple[Rule, ...]:
"""
Parse and cache the user-configured ``emission_policy`` rules.
Validation runs here exactly once per config value (``conf.emission_policy()`` is
itself cached), so task / dag resolution operates on pre-validated :class:`Rule`
objects without re-validating on every event.
"""
return tuple(parsed for raw in _ol_conf.emission_policy() if (parsed := _parse_rule(raw)) is not None)
def _classify_task_rules(
rules: Sequence[Rule],
fqcn: str,
dag_id: str,
task_id: str,
*,
hook_scoped: bool = False,
) -> tuple[list[Rule], list[Rule], list[Rule], list[Rule]]:
"""
Classify pre-validated rules into the four priority tiers for task resolution.
Only rules whose scope has a ``hook`` key (``hook_scoped=True``) or has none
(``hook_scoped=False``) are considered; the ``hook`` key itself is left for
:class:`DatasetFilter` to match per dataset.
Returns ``(task_rules, dag_rules, operator_rules, global_rules)``.
"""
task_rules: list[Rule] = []
dag_rules: list[Rule] = []
operator_rules: list[Rule] = []
global_rules: list[Rule] = []
for rule in rules:
if rule.has_hook != hook_scoped:
continue
scope = rule.scope
match_mode = rule.match_mode
if rule.has_dag and rule.has_task:
if _matches(scope[SCOPE_DAG_ID], dag_id, match_mode) and _matches(
scope[SCOPE_TASK_ID], task_id, match_mode
):
task_rules.append(rule)
elif rule.has_dag:
if _matches(scope[SCOPE_DAG_ID], dag_id, match_mode):
dag_rules.append(rule)
elif rule.has_op:
if _matches(scope[SCOPE_OPERATOR], fqcn, match_mode):
operator_rules.append(rule)
else:
global_rules.append(rule)
return task_rules, dag_rules, operator_rules, global_rules
def _classify_dag_rules(
rules: Sequence[Rule],
dag_id: str,
) -> tuple[list[Rule], list[Rule]]:
"""
Classify pre-validated rules into the two priority tiers for dag event resolution.
Only rules with empty ``scope`` (global) and rules with exactly ``{dag_id}``
contribute to dag-event resolution. Operator-scoped and task-scoped rules
never affect dag-run events.
Returns ``(dag_rules, global_rules)``.
"""
dag_rules: list[Rule] = []
global_rules: list[Rule] = []
for rule in rules:
if rule.has_dag and not rule.has_task and not rule.has_op:
if _matches(rule.scope[SCOPE_DAG_ID], dag_id, rule.match_mode):
dag_rules.append(rule)
elif not (rule.has_dag or rule.has_task or rule.has_op): # global = empty scope dict
global_rules.append(rule)
return dag_rules, global_rules
def _walk_tiers(
tiers: Sequence[Sequence[Rule]],
extract_value: Callable[[Rule], V | None],
field_label: str,
default: V,
) -> tuple[V, Rule | None]:
"""
Walk priority tiers from most-specific to least-specific, last-wins within each tier.
*extract_value* maps a rule to its value for the field; ``None`` means
"this rule does not cover this field". A contradiction WARNING is emitted when two
rules in the same tier produce different non-``None`` values. Returns
``(default, None)`` when no rule in any tier matches.
"""
for tier_rules in tiers:
value: V | None = None
winning_rule: Rule | None = None
for rule in tier_rules:
new_v = extract_value(rule)
if new_v is not None:
if winning_rule is not None and value != new_v:
log.warning(
"OpenLineage emission_policy: field '%s' has contradictory values in the same "
"priority tier (using last value %r); conflicting rules:"
"\n - %r"
"\n - %r",
field_label,
new_v,
winning_rule,
rule,
)
value = new_v
winning_rule = rule
if value is not None:
return value, winning_rule
return default, None
def _resolve_field_with_source(
tiers: list[list[Rule]], field: str, default: bool
) -> tuple[bool, Rule | None]:
"""Walk tiers for a single boolean *field* in ``controls``."""
def _extract(rule: Rule) -> bool | None:
value = rule.controls.get(field)
return value if isinstance(value, bool) else None
return _walk_tiers(tiers, _extract, field, default)
def _resolve_emit_with_source(
tiers: list[list[Rule]],
scope: str,
default: bool,
) -> tuple[bool, Rule | None]:
"""
Resolve the effective ``emit`` decision for the given *scope*.
``scope`` must be ``"task"`` or ``"dag"``. Within each rule, the
scope-specific key (``emit_task_events`` / ``emit_dag_events``) takes
precedence over the generic ``emit`` shorthand.
"""
specific_key = EMIT_TASK_EVENTS if scope == "task" else EMIT_DAG_EVENTS
def _extract(rule: Rule) -> bool | None:
value = rule.controls.get(specific_key, rule.controls.get(EMIT))
return value if isinstance(value, bool) else None
return _walk_tiers(tiers, _extract, f"emit ({scope})", default)
def _synthesize_legacy_task_rules(
operator: AnyOperator,
dag_id: str,
task_id: str,
) -> list[Rule]:
"""
Translate legacy config options into equivalent ``emission_policy`` rules for task resolution.
The returned rules are intended to be **prepended** to the user-provided rules so that
user rules (appended later) win within each tier (last-wins within a tier). They are
constructed directly (trusted, no validation path).
"""
rules: list[Rule] = []
# disabled_for_operators -> operator-scope emit:false rules
for fqcn in _ol_conf.disabled_operators():
rules.append(Rule(scope={SCOPE_OPERATOR: fqcn}, controls={EMIT: False}))
# disable_source_code -> global include_source_code:false
if not _ol_conf.is_source_enabled():
rules.append(Rule(scope={}, controls={INCLUDE_SOURCE_CODE: False}))
# include_full_task_info -> global include_full_task_info:true (only when non-default True)
if _ol_conf.include_full_task_info():
rules.append(Rule(scope={}, controls={INCLUDE_FULL_TASK_INFO: True}))
# selective_enable -> global emit:false baseline + per-task opt-in rule
if _ol_conf.selective_enable():
rules.append(Rule(scope={}, controls={EMIT: False})) # global baseline
try:
from airflow.providers.openlineage.utils.selective_enable import is_task_lineage_enabled
if is_task_lineage_enabled(operator):
# Task is explicitly opted in -> task-tier rule that overrides the global baseline.
rules.append(
Rule(scope={SCOPE_DAG_ID: dag_id, SCOPE_TASK_ID: task_id}, controls={EMIT: True})
)
except Exception:
# Non-standard operator types may not expose params correctly. Surface at debug
# level so the failure is discoverable without spamming production logs.
log.debug(
"OpenLineage selective_enable translation: is_task_lineage_enabled() failed for task '%s'",
task_id,
exc_info=True,
)
return rules
def _synthesize_legacy_dag_rules(
dag_id: str,
dag: object | None,
) -> list[Rule]:
"""Translate legacy config options into equivalent ``emission_policy`` rules for dag event resolution."""
rules: list[Rule] = []
if _ol_conf.selective_enable() and dag is not None:
rules.append(Rule(scope={}, controls={EMIT: False})) # global baseline
try:
from airflow.providers.openlineage.utils.selective_enable import is_dag_lineage_enabled
if is_dag_lineage_enabled(dag): # type: ignore[arg-type]
rules.append(Rule(scope={SCOPE_DAG_ID: dag_id}, controls={EMIT: True}))
except Exception:
log.debug(
"OpenLineage selective_enable translation: is_dag_lineage_enabled() failed for dag '%s'",
dag_id,
exc_info=True,
)
return rules
def _warn_legacy_with_emission_policy(legacy_rules: list[Rule], scope: str) -> None:
"""
Issue a DeprecationWarning listing legacy options that were translated to rules.
*scope* is ``"task"`` or ``"dag"``: a task-scope warning enumerates every legacy
option (``disabled_for_operators``, ``disable_source_code``,
``include_full_task_info``, ``selective_enable``); a dag-scope warning only
mentions ``selective_enable`` because the other three are task-only.
"""
if not legacy_rules:
return
parts: list[str] = []
if scope == "task":
if _ol_conf.disabled_operators():
translated = [
{RULE_SCOPE: {SCOPE_OPERATOR: f}, RULE_CONTROLS: {EMIT: False}}
for f in _ol_conf.disabled_operators()
]
parts.append(f" - disabled_for_operators -> {translated}")
if not _ol_conf.is_source_enabled():
parts.append(
' - disable_source_code -> [{"scope": {}, "controls": {"include_source_code": false}}]'
)
if _ol_conf.include_full_task_info():
parts.append(
' - include_full_task_info -> [{"scope": {}, "controls": {"include_full_task_info": true}}]'
)
if _ol_conf.selective_enable():
parts.append(
" - selective_enable -> "
'[{"scope": {}, "controls": {"emit": false}}] (global baseline) '
"+ per-task opt-in rules injected at runtime from enable_lineage()/disable_lineage()"
)
else: # dag scope: only selective_enable contributes legacy dag-event rules
if not _ol_conf.selective_enable():
return
parts.append(
" - selective_enable -> "
'[{"scope": {}, "controls": {"emit": false}}] (global baseline) '
"+ dag opt-in rules injected at runtime from enable_lineage()/disable_lineage()"
)
if not parts:
return
warnings.warn(
"[openlineage] One or more legacy config options are set and have been translated "
"into equivalent 'emission_policy' rules:\n" + "\n".join(parts) + "\n"
"Migrate to 'emission_policy' exclusively to silence this warning. "
"These legacy options will be removed in a future version.",
AirflowProviderDeprecationWarning,
stacklevel=4,
)
def _compute_locked_fields(
rules: list[Rule],
emit_key: str,
extra_lockable: Sequence[str] = (),
) -> frozenset[str]:
"""
Return the set of fields locked by *any* rule in *rules* that carries ``locked: true``.
**Floor-lock semantics**: a field is locked if *any* matching rule marks it locked,
regardless of whether that rule wins the value race.
*emit_key* is the scope-specific emit control (``emit_task_events`` or
``emit_dag_events``); both it and the generic ``emit`` shorthand lock the ``"emit"``
policy field. *extra_lockable* lists additional fields to scan (task-event only).
"""
locked: set[str] = set()
for rule in rules:
if not rule.locked:
continue
controls = rule.controls
if emit_key in controls or EMIT in controls:
locked.add(EMIT)
for field in extra_lockable:
if field in controls:
locked.add(field)
return frozenset(locked)
def _compute_locked_task_fields(
task_rules: list[Rule],
dag_rules: list[Rule],
operator_rules: list[Rule],
global_rules: list[Rule],
) -> frozenset[str]:
"""Compute the set of task-event fields locked by any matching conf rule."""
return _compute_locked_fields(
task_rules + dag_rules + operator_rules + global_rules,
EMIT_TASK_EVENTS,
_LOCKABLE_TASK_FIELDS,
)
def _apply_authoring_overrides(
config: EmissionPolicy,
locked_fields: frozenset[str],
flags: dict[str, bool],
emit_key: str,
context: str,
extra_fields: tuple[str, ...] = (),
) -> EmissionPolicy:
"""
Core of the authoring-override merge: apply *flags* onto *config*, skipping locked fields.
*emit_key* is the scope-specific emit control (``emit_task_events`` for tasks,
``emit_dag_events`` for DAG events). *extra_fields* lists additional lockable
fields that the task path carries but the DAG path does not.
"""
updates: dict[str, bool] = {}
effective_emit: bool | None = flags.get(emit_key, flags.get(EMIT))
if effective_emit is not None:
if EMIT in locked_fields:
log.warning(
"OpenLineage emission_policy: extend_global_openlineage_emission_policy call for 'emit' on %s"
" has no effect — locked by conf rule at value %r",
context,
config.emit,
)
else:
updates[EMIT] = effective_emit
for field in extra_fields:
if field not in flags:
continue
if field in locked_fields:
log.warning(
"OpenLineage emission_policy: extend_global_openlineage_emission_policy call for '%s' on %s"
" has no effect — locked by conf rule at value %r",
field,
context,
getattr(config, field),
)
else:
updates[field] = flags[field]
changed = {k: v for k, v in updates.items() if getattr(config, k) != v}
if not changed:
return config
_audit_log_authoring_updates(changed, context)
# ``changed`` only holds the boolean fields, never ``dataset_filter``.
return replace(config, **changed) # type: ignore[arg-type]
def _extend_policy_with_task_authoring(
config: EmissionPolicy,
locked_fields: frozenset[str],
operator: AnyOperator,
context: str,
) -> EmissionPolicy:
"""
Apply per-task authoring flags on top of *config*.
Authoring flags are stored on operator objects by
:func:`~airflow.providers.openlineage.api.emission_policy.extend_global_openlineage_emission_policy`.
Fields listed in *locked_fields* are protected — attempts to override them are
logged at WARNING level and silently ignored. ``emit_task_events`` in the authoring
flags takes precedence over ``emit`` for the task ``emit`` field, mirroring the
same precedence used in conf rule resolution.
"""
flags = _read_param(operator, OL_EMISSION_POLICY_PARAM)
if not flags:
return config
config = _apply_authoring_overrides(
config,
locked_fields,
{k: v for k, v in flags.items() if isinstance(v, bool)},
EMIT_TASK_EVENTS,
context,
extra_fields=(EXTRACT_OPERATOR_METADATA, INCLUDE_SOURCE_CODE, HOOK_LINEAGE, INCLUDE_FULL_TASK_INFO),
)
# Locks on exclude_datasets depend on the hook reporting each dataset, so the filter
# checks them when it resolves patterns.
patterns = flags.get(EXCLUDE_DATASETS)
if patterns is None:
return config
# Params written without the authoring API skip its validation, and a bad pattern would
# otherwise raise during extraction and drop the task event.
invalid = find_invalid_pattern(EXCLUDE_DATASETS, patterns)
if invalid or not isinstance(patterns, list):
log.warning("OpenLineage emission_policy: ignoring authored %s for %s", invalid, context)
return config
return replace(config, dataset_filter=replace(config.dataset_filter, authoring_patterns=tuple(patterns)))
def _extend_policy_with_dag_authoring(
config: EmissionPolicy,
locked_fields: frozenset[str],
dag: object,
context: str,
) -> EmissionPolicy:
"""
Apply per-DAG authoring flags on top of *config* for dag-run events.
``emit_dag_events`` in the authoring flags takes precedence over ``emit``,
mirroring conf rule resolution.
"""
flags = _read_param(dag, OL_EMISSION_POLICY_PARAM)
if not flags:
return config
return _apply_authoring_overrides(
config,
locked_fields,
{k: v for k, v in flags.items() if isinstance(v, bool)},
EMIT_DAG_EVENTS,
context,
)
def _resolve_task_policy_from_conf_only(
rules: Sequence[Rule],
fqcn: str,
dag_id: str,
task_id: str,
) -> tuple[EmissionPolicy, frozenset[str]]:
"""
Pure, deterministic conf resolution for task events.
Takes the already-parsed, pre-validated *rules* and a pre-computed *fqcn* string.
Rules are validated once at config-read time (:func:`_parse_user_rules`), so this
path performs no re-validation.
"""
task_rules, dag_rules, operator_rules, global_rules = _classify_task_rules(rules, fqcn, dag_id, task_id)
defaults = EmissionPolicy.defaults()
tiers = [task_rules, dag_rules, operator_rules, global_rules]
context = f"task '{task_id}' in dag '{dag_id}'"
emit, emit_rule = _resolve_emit_with_source(tiers, "task", defaults.emit)
extract_operator_metadata, em_rule = _resolve_field_with_source(
tiers, EXTRACT_OPERATOR_METADATA, defaults.extract_operator_metadata
)
include_source_code, isc_rule = _resolve_field_with_source(
tiers, INCLUDE_SOURCE_CODE, defaults.include_source_code
)
hook_lineage, hl_rule = _resolve_field_with_source(tiers, HOOK_LINEAGE, defaults.hook_lineage)
include_full_task_info, ift_rule = _resolve_field_with_source(
tiers, INCLUDE_FULL_TASK_INFO, defaults.include_full_task_info
)
if emit_rule is not None:
_audit_log_conf_field(EMIT, emit, context, emit_rule)
if em_rule is not None:
_audit_log_conf_field(EXTRACT_OPERATOR_METADATA, extract_operator_metadata, context, em_rule)
if isc_rule is not None:
_audit_log_conf_field(INCLUDE_SOURCE_CODE, include_source_code, context, isc_rule)
if hl_rule is not None:
_audit_log_conf_field(HOOK_LINEAGE, hook_lineage, context, hl_rule)
if ift_rule is not None:
_audit_log_conf_field(INCLUDE_FULL_TASK_INFO, include_full_task_info, context, ift_rule)
locked_fields = _compute_locked_task_fields(task_rules, dag_rules, operator_rules, global_rules)
hook_tiers = _classify_task_rules(rules, fqcn, dag_id, task_id, hook_scoped=True)
dataset_filter_tiers = tuple(
tuple(rule for rule in tier if _get_rule_exclude_patterns(rule) is not None)
for pair in zip(hook_tiers, tiers)
for tier in pair
)
if not any(dataset_filter_tiers):
dataset_filter_tiers = ()
return (
EmissionPolicy(
emit=emit,
extract_operator_metadata=extract_operator_metadata,
include_source_code=include_source_code,
hook_lineage=hook_lineage,
include_full_task_info=include_full_task_info,
dataset_filter=DatasetFilter(tiers=dataset_filter_tiers, context=context),
),
locked_fields,
)
def _resolve_dag_policy_from_conf_only(
rules: Sequence[Rule],
dag_id: str,
) -> tuple[EmissionPolicy, frozenset[str]]:
"""
Pure, deterministic conf resolution for dag-run events.
Returns ``(config, locked_fields)``. Only ``emit`` is meaningful for dag events;
all other fields in :class:`EmissionPolicy` carry their built-in defaults.
**Floor-lock semantics** (see :func:`_compute_locked_task_fields`): a field
is locked if *any* matching conf rule carries ``locked: true`` for that field.
``emission_policy()`` is cached at the conf layer, so calling this directly
is cheap without any additional memoisation.
"""
defaults = EmissionPolicy.defaults()
dag_rules, global_rules = _classify_dag_rules(rules, dag_id)
tiers = [dag_rules, global_rules]
context = f"dag event '{dag_id}'"
emit, emit_rule = _resolve_emit_with_source(tiers, "dag", defaults.emit)
if emit_rule is not None:
_audit_log_conf_field(EMIT, emit, context, emit_rule)
locked_fields = _compute_locked_fields(dag_rules + global_rules, EMIT_DAG_EVENTS)
return (
EmissionPolicy(
emit=emit,
extract_operator_metadata=defaults.extract_operator_metadata,
include_source_code=defaults.include_source_code,
hook_lineage=defaults.hook_lineage,
include_full_task_info=defaults.include_full_task_info,
),
locked_fields,
)
[docs]
def resolve_task_emission_policy(
operator: AnyOperator,
dag_id: str,
task_id: str,
) -> EmissionPolicy:
"""
Resolve the emission policy for a task-level event.
This is the **single authoritative entry point** for task event emission decisions.
Any active legacy options (``disabled_for_operators``, ``disable_source_code``,
``include_full_task_info``, ``selective_enable``) are *always* translated into equivalent
``emission_policy`` rules, prepended to the user-provided rules so user rules win within
each tier (last-wins). A ``DeprecationWarning`` is issued whenever a legacy option
contributes a rule — silence it by migrating those options into ``emission_policy``
exclusively.
Never raises: any failure (e.g. a misconfigured ``emission_policy``) is logged and the
built-in defaults are returned, so a bad config degrades to "emit with defaults" rather
than crashing the listener's task-notification path.
:param operator: The Airflow operator/task object.
:param dag_id: The DAG ID for this task instance.
:param task_id: The task ID for this task instance.
"""
try:
context = f"task '{task_id}' in dag '{dag_id}'"
user_rules = _parse_user_rules()
legacy_rules = _synthesize_legacy_task_rules(operator, dag_id, task_id)
_warn_legacy_with_emission_policy(legacy_rules, scope="task")
all_rules = list(legacy_rules) + list(user_rules)
from airflow.providers.openlineage.utils.utils import get_fully_qualified_class_name
fqcn = get_fully_qualified_class_name(operator)
config, locked_fields = _resolve_task_policy_from_conf_only(all_rules, fqcn, dag_id, task_id)
return _extend_policy_with_task_authoring(config, locked_fields, operator, context)
except Exception as err:
# This runs in the scheduler/listener hot path and must never raise — any failure
# (a misconfigured ``emission_policy``, or even a deprecation warning promoted to an
# error via ``-W error``) degrades to "emit with built-in defaults" rather than
# breaking task-event emission.
log.warning(
"Failed to resolve OpenLineage emission policy for task `%s` in dag `%s`; "
"emitting with default controls. Error: %s",
task_id,
dag_id,
err,
)
log.debug("Exception details:", exc_info=True)
return EmissionPolicy.defaults()
[docs]
def resolve_dag_emission_policy(dag_id: str, dag: object | None = None) -> EmissionPolicy:
"""
Resolve the emission policy for a DAG-level event.
Returns a full :class:`EmissionPolicy` for a uniform resolver API across task and
DAG scopes. Today only ``emit`` governs DAG-run events; the remaining fields
(``extract_operator_metadata``, ``include_source_code``, ``hook_lineage``,
``include_full_task_info``) are reserved — they carry their built-in defaults and
are not yet meaningful at DAG scope (no extraction happens at the DAG level). The
unified return type keeps room for future DAG-level controls (e.g. terminal-event
filtering) without re-widening the API.
Any active legacy options (currently only ``selective_enable`` affects DAG events)
are *always* translated into equivalent ``emission_policy`` rules, prepended to
the user-provided rules so user rules win within each tier (last-wins). A
``DeprecationWarning`` is issued whenever a legacy option contributes a rule.
Never raises: any failure (e.g. a misconfigured ``emission_policy``) is logged and the
built-in defaults are returned, so a bad config degrades to "emit with defaults" rather
than crashing the listener's dag-run-notification path.
:param dag_id: The DAG ID for the dag run event.
:param dag: Optional DAG object used for the ``selective_enable`` check / translation.
"""
try:
context = f"dag event '{dag_id}'"
user_rules = _parse_user_rules()
legacy_rules = _synthesize_legacy_dag_rules(dag_id, dag)
_warn_legacy_with_emission_policy(legacy_rules, scope="dag")
all_rules = list(legacy_rules) + list(user_rules)
config, locked_fields = _resolve_dag_policy_from_conf_only(all_rules, dag_id)
if dag is not None:
config = _extend_policy_with_dag_authoring(config, locked_fields, dag, context)
return config
except Exception as err:
# This runs in the scheduler/listener hot path and must never raise — any failure
# (a misconfigured ``emission_policy``, or even a deprecation warning promoted to an
# error via ``-W error``) degrades to "emit with built-in defaults" rather than
# breaking dag-run-event emission.
log.warning(
"Failed to resolve OpenLineage emission policy for dag `%s`;"
" emitting with default controls. Error: %s",
dag_id,
err,
)
log.debug("Exception details:", exc_info=True)
return EmissionPolicy.defaults()