example_dag_decorator DAG failing

[vatsrahul1001]

Apache Airflow version

3.0.0

If "Other Airflow 2 version" selected, which one?

No response

What happened?

example_dag_decorator dag is failing with below error

[2025-04-10, 09:55:33] ERROR - Task failed with exception: source="task"
TypeError: Invalid type for url.  Expected str or httpx.URL, got <class 'airflow.sdk.definitions.param.DagParam'>: <airflow.sdk.definitions.param.DagParam object at 0xffff68267760>
File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/task_runner.py", line 775 in run

File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/task_runner.py", line 1038 in _execute_task

File "/opt/airflow/task-sdk/src/airflow/sdk/bases/operator.py", line 409 in wrapper

File "/opt/airflow/airflow-core/src/airflow/example_dags/example_dag_decorator.py", line 41 in execute

File "/usr/local/lib/python3.9/site-packages/httpx/_api.py", line 198 in get

File "/usr/local/lib/python3.9/site-packages/httpx/_api.py", line 106 in request

File "/usr/local/lib/python3.9/site-packages/httpx/_client.py", line 814 in request

File "/usr/local/lib/python3.9/site-packages/httpx/_client.py", line 345 in build_request

File "/usr/local/lib/python3.9/site-packages/httpx/_client.py", line 375 in _merge_url

File "/usr/local/lib/python3.9/site-packages/httpx/_urls.py", line 11[9](http://localhost:28080/dags/example_dag_decorator/runs/manual__2025-04-10T09:55:31.872693+00:00/tasks/get_ip?try_number=1#9) in __init__

What you think should happen instead?

No response

How to reproduce

1. Run breeze start-airflow --executor CeleryExecutor --backend postgres --load-example-dags
2. Execute example_dag_decorator

Operating System

linux

Versions of Apache Airflow Providers

No response

Deployment

Official Apache Airflow Helm Chart

Deployment details

No response

Anything else?

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

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

[tirkarthi]

Looks like there is a test for task decorator being resolved. DagParam is being resolved for decorated tasks and not the operator's constructor is my guess here. The test is from commit aaaea35 . cc @ashb in case of any thoughts.

airflow/task-sdk/tests/task_sdk/definitions/test_dag.py

Lines 500 to 517 in aaf1bee

	def test_dag_param_resolves(self):
	"""Test that dag param is correctly resolved by operator"""
	from airflow.decorators import task
	@dag_decorator(schedule=None, default_args=self.DEFAULT_ARGS)
	def xcom_pass_to_op(value=self.VALUE):
	@task
	def return_num(num):
	return num
	xcom_arg = return_num(value)
	self.operator = xcom_arg.operator
	xcom_pass_to_op()
	assert isinstance(self.operator.op_args[0], DagParam)
	self.operator.render_template_fields({})
	assert self.operator.op_args[0] == 42

In below example dag print_url also accesses the url but has string type resolved. Meanwhile print(type(self.url)) inside __init__ for GetRequestOperator gives DagParam.

from __future__ import annotations

from typing import TYPE_CHECKING

import httpx
import pendulum

from airflow.models.baseoperator import BaseOperator
from airflow.sdk import dag, task

if TYPE_CHECKING:
    from airflow.sdk import Context


class GetRequestOperator(BaseOperator):
    """Custom operator to send GET request to provided url"""

    def __init__(self, *, url: str, **kwargs):
        super().__init__(**kwargs)
        self.url = url

    def execute(self, context: Context):
        return httpx.get(self.url).json()


# [START dag_decorator_usage]
@dag(
    schedule=None,
    start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
    catchup=False,
    tags=["example"],
)
def gh49046(url: str = "http://httpbin.org/get"):
    """
    DAG to get IP address and echo it via BashOperator.

    :param url: URL to get IP address from. Defaults to "http://httpbin.org/get".
    """
    get_ip = GetRequestOperator(task_id="get_ip", url=url)

    @task()
    def print_url(url_arg):
        print(f"{url_arg =}")
        return url_arg

    print_url(url)


example_dag = gh49046()

[tirkarthi]

Setting url to be a templated field works so just wondering if there was a case where before task-sdk in Airflow 2 all attributes of a task where resolved including dag params before execute even though the field is not templated which is not done in Airflow 3. The dag used to work in Airflow 2 as an example dag. The dag params implementation was done in #46176 cc: @amoghrajesh @kaxil

class GetRequestOperator(BaseOperator):
    """Custom operator to send GET request to provided url"""
    template_fields = ["url"]

    def __init__(self, *, url: str, **kwargs):
        super().__init__(**kwargs)
        self.url = url

    def execute(self, context: Context):
        return httpx.get(self.url).json()

[tirkarthi]

In 42db67b the url which used to be hard-coded string got changed to pick this value from dag param. So probably it's the example dag that needs to be updated to ensure url is templated or to continue using hard-coded url like it used to before?

Just wondering in Airflow 2 if dag params used to resolve during execute for all attributes i.e. even for attributes not specified int the template_fields of an operator.

--- a/airflow/example_dags/example_dag_decorator.py
+++ b/airflow/example_dags/example_dag_decorator.py
@@ -24,7 +24,7 @@ import pendulum
 
 from airflow.decorators import dag, task
 from airflow.models.baseoperator import BaseOperator
-from airflow.operators.email import EmailOperator
+from airflow.providers.standard.operators.bash import BashOperator
 
 if TYPE_CHECKING:
     from airflow.sdk.definitions.context import Context
@@ -48,27 +48,24 @@ class GetRequestOperator(BaseOperator):
     catchup=False,
     tags=["example"],
 )
-def example_dag_decorator(email: str = "example@example.com"):
+def example_dag_decorator(url: str = "http://httpbin.org/get"):
     """
-    DAG to send server IP to email.
+    DAG to get IP address and echo it via BashOperator.
 
-    :param email: Email to send IP to. Defaults to example@example.com.
+    :param url: URL to get IP address from. Defaults to "http://httpbin.org/get".
     """
-    get_ip = GetRequestOperator(task_id="get_ip", url="http://httpbin.org/get")
+    get_ip = GetRequestOperator(task_id="get_ip", url=url)

@vatsrahul1001 I don't have an Airflow 2 environment ready. Can you please check if below works in Airflow 2?

from __future__ import annotations

from typing import TYPE_CHECKING

import httpx
import pendulum

from airflow.models.baseoperator import BaseOperator
from airflow.sdk import dag, task

if TYPE_CHECKING:
    from airflow.sdk import Context


class GetRequestOperator(BaseOperator):
    """Custom operator to send GET request to provided url"""

    def __init__(self, *, url: str, **kwargs):
        super().__init__(**kwargs)
        self.url = url

    def execute(self, context: Context):
        return httpx.get(self.url).json()


# [START dag_decorator_usage]
@dag(
    schedule=None,
    start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
    catchup=False,
    tags=["example"],
)
def gh49046(url: str = "http://httpbin.org/get"):
    """
    DAG to get IP address and echo it via BashOperator.

    :param url: URL to get IP address from. Defaults to "http://httpbin.org/get".
    """
    get_ip = GetRequestOperator(task_id="get_ip", url=url)

    @task()
    def print_url(url_arg):
        print(f"{url_arg =}")
        return url_arg

    print_url(url)


example_dag = gh49046()

[kaxil]

Oh yea, you are right, PR incoming!