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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
