dag.test() fails when HttpHook is initialized in Operator init method

[aa-matthias]

Apache Airflow version

3.0.3

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

No response

What happened?

To avoid redundant init of httphook in multiple class methods and reuse the connection across multiple methods we are initializing the hook in the operator init method. while this works fine debugging such a task is no longer possible in Airflow (since 3.0.0).

Debugging the task via

if __name__ == "__main__":
    dag_object.test()

works fine in previous versions of Airflow (e.g. 2.10.0).

Full Stack Trace:

'TCPKeepAliveAdapter' object has no attribute 'socket_options'
  File "/venv/lib/python3.12/site-packages/requests_toolbelt/adapters/socket_options.py", line 66, in init_poolmanager
    socket_options=self.socket_options
                   ^^^^^^^^^^^^^^^^^^^
  File "/venv/lib/python3.12/site-packages/requests/adapters.py", line 236, in __setstate__
    self._pool_connections, self._pool_maxsize, block=self._pool_block

        )
  File "/usr/lib/python3.12/copy.py", line 261, in _reconstruct
    y.__setstate__(state)
  File "/usr/lib/python3.12/copy.py", line 162, in deepcopy
    y = _reconstruct(x, memo, *rv)
        ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/lib/python3.12/copy.py", line 221, in _deepcopy_dict
    y[deepcopy(key, memo)] = deepcopy(value, memo)
                             ^^^^^^^^^^^^^^^^^^^^^
  File "/usr/lib/python3.12/copy.py", line 136, in deepcopy
    y = copier(x, memo)
        ^^^^^^^^^^^^^^^
  File "/usr/lib/python3.12/copy.py", line 259, in _reconstruct
    state = deepcopy(state, memo)
            ^^^^^^^^^^^^^^^^^^^^^
  File "/usr/lib/python3.12/copy.py", line 162, in deepcopy
    y = _reconstruct(x, memo, *rv)
        ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/venv/lib/python3.12/site-packages/airflow/sdk/bases/operator.py", line 1264, in __deepcopy__
    v = copy.deepcopy(v, memo)
        ^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/lib/python3.12/copy.py", line 143, in deepcopy
    y = copier(memo)
        ^^^^^^^^^^^^
  File "/usr/lib/python3.12/copy.py", line 221, in _deepcopy_dict
    y[deepcopy(key, memo)] = deepcopy(value, memo)
                             ^^^^^^^^^^^^^^^^^^^^^
  File "/usr/lib/python3.12/copy.py", line 136, in deepcopy
    y = copier(x, memo)
        ^^^^^^^^^^^^^^^
  File "/venv/lib/python3.12/site-packages/airflow/sdk/definitions/dag.py", line 752, in __deepcopy__
    object.__setattr__(result, k, copy.deepcopy(v, memo))
                                  ^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/lib/python3.12/copy.py", line 143, in deepcopy
    y = copier(memo)
        ^^^^^^^^^^^^
  File "/usr/lib/python3.12/copy.py", line 221, in _deepcopy_dict
    y[deepcopy(key, memo)] = deepcopy(value, memo)
                             ^^^^^^^^^^^^^^^^^^^^^
  File "/usr/lib/python3.12/copy.py", line 136, in deepcopy
    y = copier(x, memo)
        ^^^^^^^^^^^^^^^
  File "/usr/lib/python3.12/copy.py", line 259, in _reconstruct
    state = deepcopy(state, memo)
            ^^^^^^^^^^^^^^^^^^^^^
  File "/usr/lib/python3.12/copy.py", line 162, in deepcopy
    y = _reconstruct(x, memo, *rv)
        ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/venv/lib/python3.12/site-packages/airflow/models/dag.py", line 1779, in create_task_groups
    new_task_group = copy.deepcopy(task_group)
                     ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/venv/lib/python3.12/site-packages/airflow/models/dag.py", line 1791, in from_sdk_dag
    create_task_groups(dag.task_group)
  File "/venv/lib/python3.12/site-packages/airflow/models/dag.py", line 1459, in clear_dags
    dag = DAG.from_sdk_dag(dag)
          ^^^^^^^^^^^^^^^^^^^^^
  File "/venv/lib/python3.12/site-packages/airflow/sdk/definitions/dag.py", line 1096, in test
    dags=[self],

                start_date=logical_date,

                end_date=logical_date,

                dag_run_state=False,  # type: ignore

            )
  File "/home/matthias/airflow/dags/fxl_to_spf_products.py", line 154, in <module>
    _dag.test()
  File "/usr/lib/python3.12/runpy.py", line 88, in _run_code
    exec(code, run_globals)
  File "/usr/lib/python3.12/runpy.py", line 198, in _run_module_as_main (Current frame)
    return _run_code(code, main_globals, None,
AttributeError: 'TCPKeepAliveAdapter' object has no attribute 'socket_options'

What you think should happen instead?

No response

How to reproduce

please see one of the two minimal repo examples attached.

minimal_repo_dag_test_issue_class.txt
minimal_repo_dag_test_issue_dag.txt

Operating System

Ubuntu 24.04

Versions of Apache Airflow Providers

No response

Deployment

Virtualenv installation

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

[aa-matthias]

seems to be caused by #44302 and moving instantiation of the TCPKeepAliveAdapter to __init__ as suggested here: #44302 (comment)

[kyungjunleeme]

@aa-matthias ur history is very helpful.

I did not encounter any issues when testing this in Airflow 2.
Also, the code you provided had some incorrect parts, so I fixed them and tested it myself.

I've been thinking about how to handle this issue.
It appears to be caused by changes introduced during the upgrade from Airflow 2 to 3, and I believe there’s room for improvement here.

In particular, we should update the implementation so that users can utilize the test functionality more easily and conveniently, while still writing code in the correct way.

If you assign this to me, I’d be happy to work on the fix and submit a PR.

[potiuk]

assigned

[kyungjunleeme]

Based on the code you shared earlier, I ran some tests and both versions work without any issues. From what I can see, it’s more likely that the problem is with the connection you’ve configured.

# filename: http_task_demo.py
import os
import pendulum
from airflow.sdk import dag, task
from airflow.providers.http.hooks.http import HttpHook

LOCAL_TZ = pendulum.timezone("Europe/Berlin")
DEFAULT_CONN_ID = "httpbin_default" 

@task()
def fetch_data_from_api(endpoint: str = "json", method: str = "GET"):
    hook = HttpHook(method=method, http_conn_id=DEFAULT_CONN_ID)
    resp = hook.run(endpoint=endpoint)
    resp.raise_for_status()
    return resp.json()

@dag(
    dag_id="exec_dag",
    schedule=None,
    start_date=pendulum.datetime(2025, 1, 1, tz=LOCAL_TZ),
    catchup=False,
)
def exec_dag():
    fetch_data_from_api()  # 기본값: httpbin /json

dag_object = exec_dag()

if __name__ == "__main__":
    dag_object.test()

import pendulum
from airflow.sdk import dag
from airflow.providers.http.hooks.http import HttpHook
from airflow.models import BaseOperator

local_tz = pendulum.timezone("Europe/Berlin")


class TestBaseOperator(BaseOperator):
    def __init__(self, http_conn_id, **kwargs):
        super().__init__(**kwargs)
        self.http_hook = HttpHook(http_conn_id=http_conn_id)

    def step1(self):
        response = self.http_hook.run(endpoint="anything/test-endpoint")
        return response.json()

    def step2(self):
        response = self.http_hook.run(endpoint="anything/test-endpoint")
        return response.json()

    def step3(self):
        response = self.http_hook.run(endpoint="anything/test-endpoint")
        return response.json()

    def execute(self, context):
        result1 = self.step1()
        result2 = self.step2()
        result3 = self.step3()
        return {"result1": result1, "result2": result2, "result3": result3}


@dag(dag_id="class_exec_dag", schedule=None, start_date=pendulum.datetime(2025, 1, 1, tz=local_tz), catchup=False)
def exec_dag():

    task1 = TestBaseOperator(http_conn_id="httpbin_default", task_id="test_operator_1")

    task2 = TestBaseOperator(http_conn_id="httpbin_default", task_id="test_operator_2")

    task1 >> task2

dag_object = exec_dag()

# debug
if __name__ == "__main__":
    dag_object.test()
# /debug

[aa-matthias]

seems to work in 3.1.1 - can debug successfully now