Re: [I] dag.test() fails when HttpHook is initialized in Operator init method [airflow]

["kyungjunleeme (via GitHub)" <gi...@apache.org>]

kyungjunleeme commented on issue #53371:
URL: https://github.com/apache/airflow/issues/53371#issuecomment-3182282494

   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.
   
   ```python
   
   # 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()
   
   ```
   
   
   ```python
   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
   ```
   
   <img width="37" height="92" alt="Image" src="https://github.com/user-attachments/assets/1f02e24f-354f-4837-a708-74d66aec3174" />


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org