airflow_http_operator_sensor_get_post_requests_dag.py

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