forked from apache/airflow-client-python
-
Notifications
You must be signed in to change notification settings - Fork 0
/
test_python_client.py
130 lines (117 loc) · 5.05 KB
/
test_python_client.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
#
# PEP 723 compliant inline script metadata (not yet widely supported)
# /// script
# requires-python = ">=3.8"
# dependencies = [
# "apache-airflow-client",
# "rich",
# ]
# ///
from __future__ import annotations
import sys
import uuid
import airflow_client.client
try:
# If you have rich installed, you will have nice colored output of the API responses
from rich import print
except ImportError:
print("Output will not be colored. Please install rich to get colored output: `pip install rich`")
pass
from airflow_client.client.api import config_api, dag_api, dag_run_api
from airflow_client.client.model.dag_run import DAGRun
# The client must use the authentication and authorization parameters
# in accordance with the API server security policy.
# Examples for each auth method are provided below, use the example that
# satisfies your auth use case.
#
# In case of the basic authentication below, make sure that Airflow is
# configured also with the basic_auth as backend additionally to regular session backend needed
# by the UI. In the `[api]` section of your `airflow.cfg` set:
#
# auth_backend = airflow.api.auth.backend.session,airflow.api.auth.backend.basic_auth
#
# Make sure that your user/name are configured properly - using the user/password that has admin
# privileges in Airflow
# Configure HTTP basic authorization: Basic
configuration = airflow_client.client.Configuration(
host="http://localhost:8080/api/v1", username="admin", password="admin"
)
# Make sure in the [core] section, the `load_examples` config is set to True in your airflow.cfg
# or AIRFLOW__CORE__LOAD_EXAMPLES environment variable set to True
DAG_ID = "example_bash_operator"
# Enter a context with an instance of the API client
with airflow_client.client.ApiClient(configuration) as api_client:
errors = False
print("[blue]Getting DAG list")
dag_api_instance = dag_api.DAGApi(api_client)
try:
api_response = dag_api_instance.get_dags()
print(api_response)
except airflow_client.client.OpenApiException as e:
print(f"[red]Exception when calling DagAPI->get_dags: {e}\n")
errors = True
else:
print("[green]Getting DAG list successful")
print("[blue]Getting Tasks for a DAG")
try:
api_response = dag_api_instance.get_tasks(DAG_ID)
print(api_response)
except airflow_client.client.exceptions.OpenApiException as e:
print(f"[red]Exception when calling DagAPI->get_tasks: {e}\n")
errors = True
else:
print("[green]Getting Tasks successful")
print("[blue]Triggering a DAG run")
dag_run_api_instance = dag_run_api.DAGRunApi(api_client)
try:
# Create a DAGRun object (no dag_id should be specified because it is read-only property of DAGRun)
# dag_run id is generated randomly to allow multiple executions of the script
dag_run = DAGRun(
dag_run_id="some_test_run_" + uuid.uuid4().hex,
)
api_response = dag_run_api_instance.post_dag_run(DAG_ID, dag_run)
print(api_response)
except airflow_client.client.exceptions.OpenApiException as e:
print(f"[red]Exception when calling DAGRunAPI->post_dag_run: {e}\n")
errors = True
else:
print("[green]Posting DAG Run successful")
# Get current configuration. Note, this is disabled by default with most installation.
# You need to set `expose_config = True` in Airflow configuration in order to retrieve configuration.
conf_api_instance = config_api.ConfigApi(api_client)
try:
api_response = conf_api_instance.get_config()
print(api_response)
except airflow_client.client.OpenApiException as e:
if "FORBIDDEN" in str(e):
print(
"[yellow]You need to set `expose_config = True` in Airflow configuration"
" in order to retrieve configuration."
)
print("[bright_blue]This is OK. Exposing config is disabled by default.")
else:
print(f"[red]Exception when calling DAGRunAPI->post_dag_run: {e}\n")
errors = True
else:
print("[green]Config retrieved successfully")
if errors:
print("\n[red]There were errors while running the script - see above for details")
sys.exit(1)
else:
print("\n[green]Everything went well")