This is an automated email from the ASF dual-hosted git repository.
uranusjr pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new a785a717c10 Add documentation for the new YAML Dag format (#74748)
a785a717c10 is described below
commit a785a717c10b908378929a3983f519d9110c60e2
Author: Tzu-ping Chung <[email protected]>
AuthorDate: Mon Oct 19 13:27:48 2026 +0800
Add documentation for the new YAML Dag format (#74748)
---
.../dag-importers.rst | 25 +-
.../docs/authoring-and-scheduling/index.rst | 1 +
.../docs/authoring-and-scheduling/yaml-dags.rst | 454 +++++++++++++++++++++
.../src/airflow/config_templates/config.yml | 8 +-
devel-common/src/docs/utils/conf_constants.py | 1 +
.../docs_build/third_party_inventories.py | 1 +
.../airflow/sdk/importers/yaml_importer/builder.py | 155 +++++--
.../importers/yaml_importer/test_importer.py | 84 ++++
8 files changed, 683 insertions(+), 46 deletions(-)
diff --git a/airflow-core/docs/administration-and-deployment/dag-importers.rst
b/airflow-core/docs/administration-and-deployment/dag-importers.rst
index 81b74576e76..87c813cb4d6 100644
--- a/airflow-core/docs/administration-and-deployment/dag-importers.rst
+++ b/airflow-core/docs/administration-and-deployment/dag-importers.rst
@@ -45,6 +45,10 @@ Airflow ships with two importers:
Each :doc:`language SDK <../authoring-and-scheduling/language-sdks/index>`
with a configured coordinator
also registers an importer for its bundle format.
+Airflow also has a YAML Dag importer at
**airflow.sdk.importers.yaml_importer.YamlDagImporter**. Unlike the
+two importers above, it is not enabled by default. Register it as shown under
`Configuring importers`_.
+See :ref:`yaml-dags` for the document format.
+
How Airflow uses importers
--------------------------
@@ -73,18 +77,29 @@ Register importers in
:ref:`config:dag_processor__dag_importer_configs`. Each en
``classpath`` of the importer and, optionally, the ``extensions`` it handles
and the ``kwargs`` to create it
with:
+.. code-block:: ini
+
+ [dag_processor]
+ dag_importer_configs = [
+ {"classpath": "airflow.sdk.importers.yaml_importer.YamlDagImporter"}
+ ]
+
+Additional configurations may be added if the importer accepts them. The YAML
+importer, for example, allows customizing the file extensions it scans (by
+default, ``.yaml`` and ``.yml``):
+
.. code-block:: ini
[dag_processor]
dag_importer_configs = [
{
- "classpath": "my_company.importers.YamlDagImporter",
- "extensions": [".yaml", ".yml"]
+ "classpath": "airflow.sdk.importers.yaml_importer.YamlDagImporter",
+ "extensions": [".yaml", ".yml", ".json"]
}
]
-Without ``extensions``, the importer handles the extensions its class
declares. An entry with only a
-classpath can be given as a string, for example
``["my_company.importers.YamlDagImporter"]``.
+If no additional arguments are needed, an entry with only a classpath can also
+be given as a string, for example ``["my_company.importers.TomlDagImporter"]``.
To register an importer for one bundle only, add an ``importers`` key, in the
same format, to that bundle's
entry in :ref:`config:dag_processor__dag_bundle_config_list`:
@@ -97,7 +112,7 @@ entry in :ref:`config:dag_processor__dag_bundle_config_list`:
"name": "pipelines",
"classpath": "airflow.dag_processing.bundles.local.LocalDagBundle",
"kwargs": {"path": "/opt/airflow/pipelines"},
- "importers": ["my_company.importers.YamlDagImporter"]
+ "importers": ["my_company.importers.TomlDagImporter"]
}
]
diff --git a/airflow-core/docs/authoring-and-scheduling/index.rst
b/airflow-core/docs/authoring-and-scheduling/index.rst
index 67a7ab19f0d..837f3cb40b1 100644
--- a/airflow-core/docs/authoring-and-scheduling/index.rst
+++ b/airflow-core/docs/authoring-and-scheduling/index.rst
@@ -37,6 +37,7 @@ It's recommended that you first review the pages in
:doc:`core concepts </core-c
assets
dag-result
Non-Python Task SDKs <language-sdks/index>
+ Writing Dags in YAML <yaml-dags>
.. _scheduling-section:
diff --git a/airflow-core/docs/authoring-and-scheduling/yaml-dags.rst
b/airflow-core/docs/authoring-and-scheduling/yaml-dags.rst
new file mode 100644
index 00000000000..2d563b5d4bd
--- /dev/null
+++ b/airflow-core/docs/authoring-and-scheduling/yaml-dags.rst
@@ -0,0 +1,454 @@
+ .. 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.
+
+.. _yaml-dags:
+
+Writing Dags in YAML
+====================
+
+|experimental|
+
+.. versionadded:: 3.4.0
+
+Airflow can build a Dag from a YAML document. One YAML document describes one
+Dag, and the YAML Dag importer turns it into the same Dag structure that
Airflow
+runs like any other source. The tasks can be ordinary Airflow operators,
+or a pointer to a task handler implemented in a Language SDK. (Support for the
+Python Task SDK is planned for a future release.)
+
+A document is declarative. Everything in it is a value, and the shape of each
+value decides what it means.
+
+The following is an example YAML file defining a Dag:
+
+.. code-block:: yaml
+
+ $schema: https://airflow.apache.org/schemas/dag/2026-10-30.json
+ dag_id: daily_report
+ schedule: "0 3 * * *"
+ start_date: 2024-01-01
+
+ tasks:
+ - id: extract
+ uses: airflow.providers.standard.operators.bash.BashOperator
+ with:
+ bash_command:
+ $t: "extract.sh {{ ds }}"
+
+ - id: publish
+ uses: airflow.providers.standard.operators.bash.BashOperator
+ needs: [extract]
+ with:
+ bash_command: "publish.sh"
+
+This Dag runs every day at 03:00. The first task ``extract`` renders the date
+into its command. The second ``publish`` waits for ``extract`` through
+``needs`` and runs a fixed command.
+
+Enabling the importer
+---------------------
+
+The YAML importer is opt-in. Register it in
+:ref:`config:dag_processor__dag_importer_configs`:
+
+.. code-block:: ini
+
+ [dag_processor]
+ dag_importer_configs = [{"classpath":
"airflow.sdk.importers.yaml_importer.YamlDagImporter"}]
+
+Set this same configuration on every component that loads Dags, including the
+Dag processor, the workers, and wherever ``airflow dags`` commands run.
+
+The importer handles ``.yaml`` and ``.yml`` files that declare a Dag
``$schema``
+(see below).
+
+A document can also be written as JSON, which the YAML importer reads as-is.
+Discovery is still by file extension, so write such a document in a ``.yaml``
or
+``.yml`` file, or add ``.json`` to the importer's ``extensions``:
+
+.. code-block:: ini
+
+ [dag_processor]
+ dag_importer_configs = [
+ {
+ "classpath": "airflow.sdk.importers.yaml_importer.YamlDagImporter",
+ "extensions": [".yaml", ".yml", ".json"]
+ }
+ ]
+
+For per-bundle registration and other options, see
+:doc:`../administration-and-deployment/dag-importers`.
+
+The shape of a document
+-----------------------
+
+``$schema`` (required)
+ The format version the document is written against, given as the URL of a
+ published schema. The version it names selects the rules the document is
+ parsed with. ``$schema`` is also how Airflow recognizes that a file
contains
+ Dags. By default (see :ref:`config:core__dag_discovery_safe_mode`), a file
+ that does not carry the canonical ``$schema`` URL is skipped, so a mistyped
+ URL makes the file disappear the same way. Editors that understand
+ ``$schema`` will offer validation and completion with no extra setup.
+
+Dag-level keys
+ Dag attributes sit at the top level, such as ``dag_id``, ``schedule``,
+ ``params``. An unknown key is reported as an error.
+
+``tasks``
+ A list of tasks. Each task has an ``id`` and at most one body.
+
+Task body: ``uses`` or ``run``
+ ``uses:`` names a ready-made operator by its fully-qualified Python import
+ path (e.g. ``airflow.providers.standard.operators.bash.BashOperator``).
+ ``run:`` holds the arguments of a code task you implement elsewhere (see
+ `Code tasks`_ below). A task can have one or the other. It can even have
+ neither; such task becomes an ``EmptyOperator`` by default, which is handy
+ as a join or marker point.
+
+``with``
+ Operator-specific arguments. These are what you would pass to the
operator's
+ constructor. Read on to understand the value grammar you can use in these
+ arguments.
+
+Task-level keys
+ Scheduling and common attributes sit next to the body, distinct from
+ ``with:``. These include things like ``retries``, ``trigger_rule``,
+ ``queue``, etc. Any key that is not a structural key (``id``, ``uses``,
+ ``run``, ``with``, ``needs``, ``extends``) must be a base operator
argument;
+ an operator-specific argument goes under ``with:`` instead, and a key that
+ is neither is an error.
+
+``needs``
+ A list of upstream tasks (referenced by ``id``) that must run before this
+ one. This describes an explicit dependency edge; XCom references in
+ ``with:`` or ``run:`` are discovered automatically and do not need to be
+ repeated here.
+
+
+Templates and extends
+---------------------
+
+Tasks often share arguments. You can define a named partial task under the
+top-level ``templates:`` key, and pull it into a task with ``extends``:
+
+.. code-block:: yaml
+
+ templates:
+ retryable:
+ retries: 2
+ shell:
+ extends: [retryable] # a template can extend other templates
+ uses: airflow.providers.standard.operators.bash.BashOperator
+ with:
+ cwd: /opt/airflow/scripts
+
+ tasks:
+ - id: extract
+ extends: [shell] # pull the template into this task
+ with:
+ bash_command: "extract.sh"
+ - id: load
+ extends: [shell]
+ with:
+ bash_command: "load.sh"
+
+Both tasks run a ``BashOperator`` with ``retries: 2`` and the shared ``cwd``,
+and add only their own ``bash_command``. A template may set anything a task
can,
+and may itself ``extends`` other templates.
+
+The templates named in ``extends`` are applied left to right, and the task's
own
+inline fields are applied last. Merging is shallow: ``with:`` and ``run:`` are
+combined one key at a time, while every other field is taken as a whole rather
+than deep-merged.
+
+The assembled task is validated against the real operator, so a field the
+operator does not accept is reported as an error. It is also an error if a
+non-existent template is referenced.
+
+
+Values are literal by default
+-----------------------------
+
+A value under ``with:`` (or ``run:``, with the restrictions in `Code tasks`_)
is
+taken literally, unless it is a single-key object whose key is a reserved
+``$``-marker. For example, ``"echo foo.sh"`` and ``"{{ not rendered }}"`` are
+taken as exactly that string, not a template, even if ``bash_command`` is
listed
+on the operator as a template field.
+
+Four markers opt a value into special handling:
+
+``{$t: "..."}`` is a template string
+ A Jinja template, rendered when the task runs. This is the only way a value
+ is templated: ``{$t: "{{ ds }}"}``, ``{$t: "{{ params.region }}"}``,
+ ``{$t: "{{ conn.my_db.host }}"}``. The long form ``{$template: "..."}`` is
+ also accepted.
+
+``{$x: id}`` references an upstream output
+ The XCom output of an upstream task. ``{$x: extract}`` references
+ ``extract``'s return value, and ``{$x: {task: extract, key: rows}}`` a
+ specific key. Binding a value this way also creates the dependency edge
+ implicitly, so a task that consumes another's output does not need to also
+ list it in ``needs:``. The long form is ``{$xcom: ...}``.
+
+``{$f: path}`` reads a file's contents
+ The contents of a file are read and used. The path is resolved against the
+ directory containing the current YAML document. A path that leaves the
+ containing directory after resolution (including symlinks) is rejected.
+ ``{$t: {$f: path}}`` allows reading the file and then rendering its
contents
+ as a template. The long form is ``{$file: path}``. A file is loaded only
+ when you ask with ``$f``; it is never inferred from a filename suffix.
+
+``{$const: value}`` is a forced literal
+ This can be used to wrap contents that would otherwise carry a reserved
``$``
+ meaning. For example, ``{$const: {$x: 1}}`` is the dict ``{"$x": 1}``, not
+ an XCom.
+
+The ``$t``, ``$x`` and ``$f`` markers may only be used on a field specified by
+the operator as template-able (listed in ``template_fields``). Using them on
any
+other field causes an error. ``$const`` and plain values may be used anywhere.
+For an operator you write yourself, the operator decides which fields may
accept
+markers; see `Using your own operators`_.
+
+The ``$`` prefix in mapping keys is reserved for these markers. Prefixing any
+other key with ``$`` causes a parse error. To include a key that must start
with
+``$``, wrap the mapping in ``$const``.
+
+Markers nest. A marker inside a list, or inside a nested mapping such as an
+``env`` dict, is resolved by the same rules, so references and templates
compose
+through structure:
+
+.. code-block:: yaml
+
+ with:
+ env:
+ RUN_DATE:
+ $t: "{{ ds }}" # rendered
+ REGION: us-east-1 # literal
+
+Dates, datetimes, and deltas
+----------------------------
+
+A :class:`~datetime.date` or :class:`~datetime.datetime` argument accepts an
+ISO-8601 string or a YAML-native value, and they mean the same thing:
+
+.. code-block:: yaml
+
+ start_date: "2024-01-01" # ISO-8601 string
+ # start_date: 2024-01-01 # or a YAML-native date (same value)
+ end_date: "2024-06-01T12:00:00" # ISO-8601 datetime string
+ # end_date: 2024-06-01 12:00:00 # or a YAML-native datetime (same value)
+
+A bare date on a :class:`~datetime.datetime` argument becomes midnight. An
+ISO-8601 string is parsed.
+
+A :class:`~datetime.timedelta` or
:class:`~dateutil.relativedelta.relativedelta`
+argument accepts a duration string, written in a short form or as an ISO-8601
+duration:
+
+.. code-block:: yaml
+
+ retry_delay: 5m # short form, 5 minutes
+ execution_timeout: PT1H # ISO-8601 duration, 1 hour
+
+The short form combines ``w``, ``d``, ``h``, ``m`` and ``s`` (for example
+``1h30m``) and builds a ``timedelta``. The `ISO-8601 duration
+<https://en.wikipedia.org/wiki/ISO_8601#Durations>`_ form (``P...``) can build
a
+``timedelta`` or ``relativedelta``. Only whole-number components are
+accepted, so a fractional value like ``PT0.5H`` is not recognized; write
+``PT30M`` instead.
+
+Each conversion applies only to an argument whose **type annotation** names the
+matching type; a value bound to any other argument keeps its original type.
+
+Using your own operators
+------------------------
+
+``uses:`` takes any operator's fully-qualified Python import path, so your own
+operator class can work exactly like a provider's.
+
+.. code-block:: yaml
+
+ - id: ingest
+ uses: my_company.operators.IngestOperator
+ trigger_rule: all_done # base arguments at the task level
+ with: # your operator's own arguments
+ source: s3://bucket/incoming
+ cutoff: 2024-01-01
+
+An operator takes two kinds of argument, kept in separate places. Your
+operator's own constructor arguments go under ``with:``. The arguments every
+operator inherits from the base operator, such as ``retries``, ``retry_delay``,
+``trigger_rule``, ``queue``, ``pool``, ``priority_weight``,
``depends_on_past``,
+``start_date``, and the rest of the task model go at the task level beside the
+body. Placing a base argument under ``with:``, or an operator-specific one at
+the task level, is an error, so each argument has a single home.
+
+The task is constructed roughly like this:
+
+.. code-block:: python
+
+ IngestOperator(
+ task_id="ingest",
+ source="s3://bucket/incoming",
+ cutoff=datetime.datetime(2024, 1, 1, 0, 0),
+ trigger_rule="all_done",
+ )
+
+Templated argument rendering
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+A task argument is literal by default; only a value marked with ``{$t: ...}``
+is rendered as a Jinja template. As `Values are literal by default`_ notes, a
+marker may be used only on a field the operator templates, so an operator you
+write must list the field in its ``template_fields`` before a marker works on
it.
+
+A field left unset keeps the operator's own default, including any templating
+the operator applies to that default. An operator can also derive one of its
own
+templated fields from the arguments you give it, and that derived field still
+renders, so an unmarked value can reach a rendered field indirectly. A value is
+also never read from a file because of its name or suffix (in other words,
+Python's ``template_ext`` does not apply when the operator is used in YAML); a
+file is loaded only where you ask with ``{$f: ...}``.
+
+Literal values inside a rendered container
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+A literal also stays literal when it sits in the same list or mapping as a
+``{$t: ...}`` or ``{$x: ...}``. The marked entry renders, and the plain entries
+beside it do not. Most operators receive the finished value and never notice.
An
+operator whose ``__init__`` reads the individual entries of such a container
may
+notice edge cases, but operator authors should be aware that argument
processing
+in ``__init__`` is generally considered bad practice in the first place. Such
+processing should generally be done in ``execute`` instead, which is always
+called after template rendering has been performed.
+
+Accepting a date, datetime, or delta argument
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Whether a value is converted depends on how the receiving argument is
annotated.
+An argument annotated as a date, datetime, or delta type receives the
conversion
+described in `Dates, datetimes, and deltas`_; every other argument is left as
+written. To let your operator take these values, annotate the arguments:
+
+.. code-block:: python
+
+ import datetime
+
+ from airflow.sdk import BaseOperator
+
+
+ class IngestOperator(BaseOperator):
+ def __init__(
+ self,
+ *,
+ cutoff: datetime.datetime | None = None,
+ retention: datetime.timedelta | None = None,
+ **kwargs,
+ ):
+ super().__init__(**kwargs)
+ self.cutoff = cutoff
+ self.retention = retention
+
+With the annotations above, ``cutoff: 2024-01-01`` arrives as
+``datetime.datetime(2024, 1, 1, 0, 0)`` and ``retention: 5d`` as
+``datetime.timedelta(days=5)``. Pendulum types are also coerced automatically.
+
+The annotation is matched by the name, so the following are all recognized
+(plus other similar variants):
+
+* ``datetime.datetime``
+* ``pendulum.DateTime``
+* ``datetime.date``
+* ``pendulum.Date``
+* ``datetime.timedelta``
+* ``dateutil.relativedelta.relativedelta``
+* ``pendulum.Duration``
+
+Unions are also parsed, so things like ``datetime | None`` are recognized too.
+A type imported under another alias is not.
+
+Reserved arguments
+~~~~~~~~~~~~~~~~~~
+
+A few names are reserved and cannot be set on a task. The task id comes from
+``id``, not ``task_id``; arguments ``dag``, ``task_group``, and
``default_args``
+are not accepted. Use ``templates:`` and ``extends:`` for shared defaults in
+place of ``default_args``. (Support for task groups is planned for a future
+release.)
+
+Scheduling
+----------
+
+``schedule:`` accepts the forms Airflow already understands, chosen by shape:
+
+* a cron string (``"0 3 * * *"``) or a preset (``"@daily"``);
+* ``null`` or an omitted key, for a Dag that is only triggered manually or
+ externally;
+* an object ``{uses: <timetable import path>, with: {...}}`` to construct a
+ timetable, the same ``uses`` / ``with`` idea used for operators.
+
+Values under a timetable's ``with:`` receive the same date and delta coercion
as
+operator arguments (see `Dates, datetimes, and deltas`_), matched against the
+timetable's constructor annotations. Value markers (``$t``, ``$x``, ``$f``) are
+not interpreted there since timetables do not accept execution-time resolution.
+
+Support for asset-triggering is planned for a future release.
+
+Code tasks
+----------
+
+A task can run your own code instead of a ready-made operator with ``run:``,
+which holds the code's arguments directly:
+
+.. code-block:: yaml
+
+ - id: transform
+ queue: go-workers # to route to a coordinator
+ run:
+ source: {$x: extract}
+ factor: 2
+
+A ``run:`` task is executed by a Task SDK runtime, so it must be routed to a
+coordinator. In practice, a ``run:`` task always needs a task-level ``queue:``
+mapped to a coordinator under ``[sdk] queue_to_coordinator``. See
+:doc:`language-sdks/index` for writing the handler and configuring the
+coordinator.
+
+Each argument in ``run:`` is either a JSON-serializable literal, or an upstream
+task's output, in the ``{$x: task}`` form, which also wires the dependency.
+``$f`` and ``$const`` resolve to literals. Due to restrictions imposed by the
+cross-language task handler design, the ``run:`` syntax does not support
+a template (``$t``), a ``$x`` referencing a custom XCom key (not the return
+value), or an ``$x`` nested inside a larger value.
+
+Current limitations
+-------------------
+
+Some features are not yet available when a Dag is defined in YAML. These may be
+implemented in a future release.
+
+* **Assets.** A task's ``inlets`` and ``outlets`` are rejected, and a Dag
cannot
+ be scheduled on an asset (see `Scheduling`_).
+* **Task groups.** Tasks cannot be nested into groups (see `Reserved
+ arguments`_).
+* **Dynamic task mapping.** There is no equivalent of ``.expand()``.
+* **Arguments that must be a Python object.** This particularly affects
+ callbacks (which need a Python function).
+
+A rejected key is reported as an error rather than silently ignored, as is any
+other argument the target operator does not accept.
diff --git a/airflow-core/src/airflow/config_templates/config.yml
b/airflow-core/src/airflow/config_templates/config.yml
index 586553e4f09..9bae77fb9bf 100644
--- a/airflow-core/src/airflow/config_templates/config.yml
+++ b/airflow-core/src/airflow/config_templates/config.yml
@@ -3345,15 +3345,15 @@ dag_processor:
If ``extensions`` is omitted, the extensions declared by the importer
class are used.
An entry may also be given as a bare classpath string when it needs no
other
- settings, so ``["my_company.importers.YamlDagImporter"]`` is
equivalent to
- ``[{"classpath": "my_company.importers.YamlDagImporter"}]``.
+ settings, so ``["my_company.importers.TomlDagImporter"]`` is
equivalent to
+ ``[{"classpath": "my_company.importers.TomlDagImporter"}]``.
version_added: 3.4.0
type: string
example: >
[
{
- "classpath": "my_company.importers.YamlDagImporter",
- "extensions": [".yaml", ".yml"],
+ "classpath": "my_company.importers.TomlDagImporter",
+ "extensions": [".toml"],
"kwargs": {}
}
]
diff --git a/devel-common/src/docs/utils/conf_constants.py
b/devel-common/src/docs/utils/conf_constants.py
index fffdadea96a..fd2116bbab6 100644
--- a/devel-common/src/docs/utils/conf_constants.py
+++ b/devel-common/src/docs/utils/conf_constants.py
@@ -296,6 +296,7 @@ def get_intersphinx_mapping() -> dict[str, tuple[str,
tuple[str]]]:
[
"boto3",
"celery",
+ "dateutil",
"docker",
"hdfs",
"jinja2",
diff --git a/devel-common/src/sphinx_exts/docs_build/third_party_inventories.py
b/devel-common/src/sphinx_exts/docs_build/third_party_inventories.py
index acae9376c01..1b2fa537e0a 100644
--- a/devel-common/src/sphinx_exts/docs_build/third_party_inventories.py
+++ b/devel-common/src/sphinx_exts/docs_build/third_party_inventories.py
@@ -19,6 +19,7 @@ from __future__ import annotations
THIRD_PARTY_INDEXES = {
"boto3": "https://boto3.amazonaws.com/v1/documentation/api/latest";,
"celery": "https://docs.celeryq.dev/en/stable/";,
+ "dateutil": "https://dateutil.readthedocs.io/en/stable";,
"docker": "https://docker-py.readthedocs.io/en/stable";,
"hdfs": "https://hdfscli.readthedocs.io/en/latest";,
"jinja2": "https://jinja.palletsprojects.com/en/3.1.x";,
diff --git a/task-sdk/src/airflow/sdk/importers/yaml_importer/builder.py
b/task-sdk/src/airflow/sdk/importers/yaml_importer/builder.py
index 5baab192dfa..207ef0dc1f6 100644
--- a/task-sdk/src/airflow/sdk/importers/yaml_importer/builder.py
+++ b/task-sdk/src/airflow/sdk/importers/yaml_importer/builder.py
@@ -44,6 +44,7 @@ import re
from typing import TYPE_CHECKING, Any, NamedTuple, cast
import attrs
+import pendulum
from dateutil.relativedelta import relativedelta
from airflow.sdk import DAG, BaseOperator, XComArg, literal, task as
task_decorators
@@ -354,12 +355,23 @@ def _coerce_temporal(mapping: dict[str, Any], target:
type) -> dict[str, Any]:
Coerce scalars to datetime or time delta types expected by *target*.
Coercion follows how each argument is typed in the argument list. A value
- bound to ``datetime`` is promoted to a datetime; a YAML date becomes
- midnight, an ISO string is parsed. A value bound to ``timedelta`` or
- ``relativedelta`` is parsed from a duration string.
+ bound to ``datetime`` is promoted to a datetime. A YAML date becomes
+ midnight. An ISO string is parsed
- An argument of neither type is left untouched, so a string that merely
looks
- like a date or duration keeps its type.
+ A value bound to ``date`` is parsed from an ISO string; a YAML date or
+ datetime is passed through, so a datetime keeps its time rather than being
+ narrowed.
+
+ A value bound to ``timedelta``, ``relativedelta``, or ``pendulum.Duration``
+ is parsed from a duration string.
+
+ The concrete type follows the annotation. A ``pendulum`` annotation
+ (``DateTime``, ``Date``, or ``Duration``) yields a Pendulum value, and a
+ standard library or ``dateutil`` annotation yields a value of that type. A
+ datetime passed to a ``date`` argument is the one exception, keeping its
+ stdlib type.
+
+ An argument of none of these types is left untouched.
"""
coercer = _FieldCoercer.for_target(target)
return {key: coercer.coerce(key, value) for key, value in mapping.items()}
@@ -382,8 +394,10 @@ class _FieldCoercer(dict):
# Datetime arguments, stdlib or pendulum.
_DATETIME_TYPE_NAMES = frozenset({"datetime", "DateTime"})
+ # Date arguments, stdlib or pendulum.
+ _DATE_TYPE_NAMES = frozenset({"date", "Date"})
# Time-delta arguments, stdlib timedelta or dateutil relativedelta.
- _DELTA_TYPE_NAMES = frozenset({"timedelta", "relativedelta"})
+ _DELTA_TYPE_NAMES = frozenset({"timedelta", "relativedelta", "Duration"})
# ISO 8601 duration format.
_ISO_DURATION_RE = re.compile(
@@ -405,9 +419,11 @@ class _FieldCoercer(dict):
return coercer
for name, param in params.items():
annotation = str(param.annotation)
- if cls._is_datetime_annotation(annotation):
- coercer[name] = cls._to_datetime
- elif allowed := cls._is_delta_annotation(annotation):
+ if allowed := cls._match_datetime_annotation(annotation):
+ coercer[name] = functools.partial(cls._to_datetime,
allowed=allowed)
+ elif allowed := cls._match_date_annotation(annotation):
+ coercer[name] = functools.partial(cls._to_date,
allowed=allowed)
+ elif allowed := cls._match_delta_annotation(annotation):
coercer[name] = functools.partial(cls._to_delta,
allowed=allowed)
return coercer
@@ -419,61 +435,102 @@ class _FieldCoercer(dict):
return self[name](value)
@classmethod
- def _is_datetime_annotation(cls, annotation: str) -> bool:
- """Check if *annotation* names a datetime type."""
+ def _match_datetime_annotation(cls, annotation: str) -> frozenset[str]:
+ """Return the datetime type names *annotation* mentions, stdlib or
pendulum."""
names = (ref.rsplit(".", 1)[-1] for ref in
cls._TYPE_NAME_RE.findall(annotation))
- return any(name in cls._DATETIME_TYPE_NAMES for name in names)
+ return frozenset(name for name in names if name in
cls._DATETIME_TYPE_NAMES)
@classmethod
- def _to_datetime(cls, value: Any) -> Any:
- if isinstance(value, datetime.datetime):
+ def _to_datetime(cls, value: Any, allowed: frozenset[str]) -> Any:
+ if isinstance(value, str):
+ try:
+ result = datetime.datetime.fromisoformat(value)
+ except ValueError:
+ return value
+ elif isinstance(value, datetime.datetime):
+ result = value
+ elif isinstance(value, datetime.date):
+ result = datetime.datetime.combine(value, datetime.time.min)
+ else:
return value
- if isinstance(value, datetime.date):
- return datetime.datetime.combine(value, datetime.time.min)
+ if "DateTime" in allowed:
+ return pendulum.instance(result)
+ return result
+
+ @classmethod
+ def _match_date_annotation(cls, annotation: str) -> frozenset[str]:
+ """Return the date type names *annotation* mentions, stdlib or
pendulum."""
+ names = (ref.rsplit(".", 1)[-1] for ref in
cls._TYPE_NAME_RE.findall(annotation))
+ return frozenset(name for name in names if name in
cls._DATE_TYPE_NAMES)
+
+ @classmethod
+ def _to_date(cls, value: Any, allowed: frozenset[str]) -> Any:
if isinstance(value, str):
try:
- return datetime.datetime.fromisoformat(value)
+ result = datetime.date.fromisoformat(value)
except ValueError:
return value
- return value
+ elif isinstance(value, datetime.date):
+ result = value
+ else:
+ return value
+ # A datetime is a date subtype, so it is left as-is; only a plain date
is
+ # wrapped for a ``pendulum.Date`` annotation.
+ if "Date" in allowed and type(result) is datetime.date:
+ return pendulum.instance(result)
+ return result
@classmethod
- def _is_delta_annotation(cls, annotation: str) -> frozenset[str]:
- """Check if *annotation* names a duration delta type."""
+ def _match_delta_annotation(cls, annotation: str) -> frozenset[str]:
+ """Return the duration delta type names *annotation* mentions, stdlib,
dateutil, or pendulum."""
names = (ref.rsplit(".", 1)[-1] for ref in
cls._TYPE_NAME_RE.findall(annotation))
return frozenset(name for name in names if name in
cls._DELTA_TYPE_NAMES)
@classmethod
def _to_delta(cls, value: Any, allowed: frozenset[str]) -> Any:
"""
- Parse a duration string into a ``timedelta`` or ``relativedelta``.
+ Parse a duration string into a Python type.
+
+ This can produce a :class:`datetime.timedelta`,
+ :class:`dateutil.relativedelta.relativedelta`, or
``pendulum.Duration``.
+
+ An ISO-8601 format duration (e.g. ``P1DT6H``, ``P1Y2M``) and a short
+ form (``5m``, ``1h30m``) are accepted. A value that cannot be parsed
+ into the target duration type is returned unchanged.
- An ISO-8601 format duration (e.g. ``P1DT6H``, ``P1Y2M``) is accepted
for
- both stdlib timedelta and dateutil relativedelta arguments; a short
form
- (``5m``, ``1h30m``) is also accepted for timedelta. Years and months
- need a relativedelta, so an ISO duration carrying them is left
untouched
- unless relativedelta is allowed.
+ The short form does not support specifying years or months. Similarly,
+ when ``timedelta`` is the only accepted duration type on an argument,
an
+ ISO duration string carrying year or month components would be left
+ unparsed. A ``relativedelta`` or ``pendulum.Duration`` argument would
+ accept those components, so the same ISO duration string would be
parsed.
- A value that does not parse as a duration string is returned unchanged.
+ The concrete type follows the annotation. When more than one annotated
+ type can hold the value, a ``timedelta`` is preferred for a value
without
+ years or months, and a ``relativedelta`` over a ``pendulum.Duration``
for
+ years and months (which ``pendulum.Duration`` only approximates).
"""
if not isinstance(value, str):
return value
if (parts := cls._parse_iso_duration(value)) is not None:
if parts["years"] or parts["months"]:
- return cls._as_relativedelta(parts) if "relativedelta" in
allowed else value
+ if "relativedelta" in allowed:
+ return cls._as_relativedelta(parts)
+ if "Duration" in allowed:
+ return cls._as_duration(parts)
+ return value
if "timedelta" in allowed:
- return datetime.timedelta(
- weeks=parts["weeks"],
- days=parts["days"],
- hours=parts["hours"],
- minutes=parts["minutes"],
- seconds=parts["seconds"],
- )
+ return cls._as_timedelta(parts)
+ if "Duration" in allowed:
+ return cls._as_duration(parts)
if "relativedelta" in allowed:
return cls._as_relativedelta(parts)
return value
- if "timedelta" in allowed and (parts :=
cls._parse_short_duration(value)) is not None:
- return datetime.timedelta(**parts)
+ if (parts := cls._parse_short_duration(value)) is not None:
+ if "timedelta" in allowed:
+ return datetime.timedelta(**parts)
+ if "Duration" in allowed:
+ return pendulum.Duration(**parts)
+ return value
return value
@classmethod
@@ -489,6 +546,30 @@ class _FieldCoercer(dict):
seconds=parts["seconds"],
)
+ @classmethod
+ def _as_timedelta(cls, parts: dict[str, int]) -> datetime.timedelta:
+ """Build a timedelta from duration component ints (years/months
ignored)."""
+ return datetime.timedelta(
+ weeks=parts["weeks"],
+ days=parts["days"],
+ hours=parts["hours"],
+ minutes=parts["minutes"],
+ seconds=parts["seconds"],
+ )
+
+ @classmethod
+ def _as_duration(cls, parts: dict[str, int]) -> pendulum.Duration:
+ """Build a pendulum.Duration from duration component ints."""
+ return pendulum.Duration(
+ years=parts["years"],
+ months=parts["months"],
+ weeks=parts["weeks"],
+ days=parts["days"],
+ hours=parts["hours"],
+ minutes=parts["minutes"],
+ seconds=parts["seconds"],
+ )
+
@classmethod
def _parse_iso_duration(cls, value: str) -> dict[str, int] | None:
"""Parse an ISO-8601 duration into component ints."""
diff --git a/task-sdk/tests/task_sdk/importers/yaml_importer/test_importer.py
b/task-sdk/tests/task_sdk/importers/yaml_importer/test_importer.py
index 8c964c7e2c3..092c47c7f3d 100644
--- a/task-sdk/tests/task_sdk/importers/yaml_importer/test_importer.py
+++ b/task-sdk/tests/task_sdk/importers/yaml_importer/test_importer.py
@@ -23,6 +23,7 @@ import textwrap
import zipfile
from types import SimpleNamespace
+import pendulum
import pytest
from dateutil.relativedelta import relativedelta
@@ -82,6 +83,29 @@ class _TemporalOperator(BaseOperator):
OP_TEMPORAL = f"{__name__}.{_TemporalOperator.__name__}"
+class _DateOperator(BaseOperator):
+ """An operator with a date-typed argument (``day``)."""
+
+ def __init__(self, *, day: datetime.date | None = None, **kwargs):
+ super().__init__(**kwargs)
+ self.day = day
+
+
+OP_DATE = f"{__name__}.{_DateOperator.__name__}"
+
+
+class _PendulumOperator(BaseOperator):
+ """An operator with pendulum-typed temporal arguments."""
+
+ def __init__(self, *, when: pendulum.DateTime | None = None, day:
pendulum.Date | None = None, **kwargs):
+ super().__init__(**kwargs)
+ self.when = when
+ self.day = day
+
+
+OP_PENDULUM = f"{__name__}.{_PendulumOperator.__name__}"
+
+
class _PeriodOperator(BaseOperator):
"""An operator with time-period-typed arguments."""
@@ -91,12 +115,14 @@ class _PeriodOperator(BaseOperator):
gap: datetime.timedelta | None = None,
window: relativedelta | None = None,
either: datetime.timedelta | relativedelta | None = None,
+ span: pendulum.Duration | None = None,
**kwargs,
):
super().__init__(**kwargs)
self.gap = gap
self.window = window
self.either = either
+ self.span = span
OP_PERIOD = f"{__name__}.{_PeriodOperator.__name__}"
@@ -300,6 +326,43 @@ def test_import_coerces_temporal_with_arg(mock_bundle):
assert (cutoff.year, cutoff.month, cutoff.day) == (2024, 1, 1)
[email protected](
+ ("value", "exp_type"),
+ [
+ ("2024-01-01", datetime.date), # YAML date -> plain date
+ ("'2024-01-01'", datetime.date), # ISO date string -> plain date
+ ("2024-01-01T03:30:00", datetime.datetime), # a datetime is a date
subtype, left as-is
+ ],
+)
+def test_import_coerces_date_with_arg(mock_bundle, value, exp_type):
+ definition = _def(
+ mock_bundle,
+ "dag.yaml",
+ _doc(f"- {{id: a, uses: {OP_DATE}, with: {{day: {value}}}}}"),
+ )
+ result = YamlDagImporter().import_definition(definition,
bundle=mock_bundle)
+ assert result.success, result.errors
+ day = result.dags[0].get_task("a").day
+ assert type(day) is exp_type
+ assert (day.year, day.month, day.day) == (2024, 1, 1)
+
+
+def test_import_coerces_pendulum_temporal_args(mock_bundle):
+ # a pendulum.DateTime / pendulum.Date annotation yields pendulum values,
not stdlib
+ definition = _def(
+ mock_bundle,
+ "dag.yaml",
+ _doc(f"- {{id: a, uses: {OP_PENDULUM}, with: {{when:
2024-01-01T03:30:00, day: 2024-01-01}}}}"),
+ )
+ result = YamlDagImporter().import_definition(definition,
bundle=mock_bundle)
+ assert result.success, result.errors
+ task = result.dags[0].get_task("a")
+ assert type(task.when) is pendulum.DateTime
+ assert (task.when.year, task.when.month, task.when.day, task.when.hour) ==
(2024, 1, 1, 3)
+ assert type(task.day) is pendulum.Date
+ assert (task.day.year, task.day.month, task.day.day) == (2024, 1, 1)
+
+
def test_import_coerces_timedelta_with_arg(mock_bundle):
# a timedelta-typed argument accepts the short form and ISO-8601
definition = _def(
@@ -325,6 +388,27 @@ def
test_import_coerces_relativedelta_with_arg(mock_bundle):
assert result.dags[0].get_task("a").window == relativedelta(years=1,
months=2)
+def test_import_coerces_pendulum_duration_with_arg(mock_bundle):
+ # a pendulum.Duration arg accepts the short form and an ISO duration
carrying
+ # years/months (which a timedelta cannot represent)
+ definition = _def(
+ mock_bundle,
+ "dag.yaml",
+ _doc(
+ f"- {{id: a, uses: {OP_PERIOD}, with: {{span: 5m}}}}\n"
+ f"- {{id: b, uses: {OP_PERIOD}, with: {{span: P1Y2M}}}}",
+ ),
+ )
+ result = YamlDagImporter().import_definition(definition,
bundle=mock_bundle)
+ assert result.success, result.errors
+ span_a = result.dags[0].get_task("a").span
+ span_b = result.dags[0].get_task("b").span
+ assert type(span_a) is pendulum.Duration
+ assert span_a == pendulum.Duration(minutes=5)
+ assert type(span_b) is pendulum.Duration
+ assert (span_b.years, span_b.months) == (1, 2)
+
+
def test_import_period_union_picks_type_by_components(mock_bundle):
# for an arg that accepts both, an ISO duration with months/years becomes
a relativedelta,
# otherwise a timedelta
