airflow_http_operator_sensor_get_post_requests_dag.py

1import json
2from datetime import datetime
3
4from airflow import DAG
5from airflow.providers.http.operators.http import SimpleHttpOperator
6from airflow.providers.http.sensors.http import HttpSensor
7
8with DAG(
9    dag_id="example_http_operator",
10    start_date=datetime(2022, 1, 1),
11    schema=None,
12    catchup=False,
13) as dag:
14
15    # [START howto_operator_http_sensor]
16    check_api = HttpSensor(
17        task_id="check_api",
18        http_conn_id="http_default",
19        endpoint="get",
20        request_params={"param": "value"},
21        response_check=lambda response: "httpbin" in response.text,
22        poke_interval=5,
23    )
24    # [END howto_operator_http_sensor]
25
26    # [START howto_operator_http_get]
27    get_op = SimpleHttpOperator(
28        task_id="get_op",
29        http_conn_id="http_default",
30        method="GET",
31        endpoint="get",
32        data={"param1": "value1", "param2": "value2"},
33        headers={"Content-Type": "application/json"},
34    )
35    # [END howto_operator_http_get]
36
37    # [START howto_operator_http_post]
38    post_op = SimpleHttpOperator(
39        task_id="post_op",
40        http_conn_id="http_default",
41        method="POST",
42        endpoint="post",
43        data=json.dumps({"priority": 5}),
44        headers={"Content-Type": "application/json"},
45        response_filter=lambda response: response.json(),
46    )
47    # [END howto_operator_http_post]
48
49    check_api >> get_op >> post_op