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(2021, 1, 1),
11    catchup=False,
12    tags=["example"],
13) as dag:
14
15    # [START howto_operator_http_sensor_check]
16    dag_sensor_check = HttpSensor(
17        task_id="http_sensor_check",
18        http_conn_id="http_default",
19        endpoint="",
20        request_params={},
21        response_check=lambda response: "httpbin" in response.text,
22        poke_interval=5,
23    )
24    # [END howto_operator_http_sensor_check]
25
26    # [START howto_operator_simple_http_operator_get]
27    get_op = SimpleHttpOperator(
28        task_id="get_op",
29        method="GET",
30        endpoint="get",
31        data={"param1": "value1", "param2": "value2"},
32        headers={},
33    )
34    # [END howto_operator_simple_http_operator_get]
35
36    # [START howto_operator_simple_http_operator_post]
37    post_op = SimpleHttpOperator(
38        task_id="post_op",
39        endpoint="post",
40        data=json.dumps({"priority": 5}),
41        headers={"Content-Type": "application/json"},
42        response_check=lambda response: response.json()["json"]["priority"] == 5,
43    )
44    # [END howto_operator_simple_http_operator_post]
45
46    dag_sensor_check >> get_op >> post_op