Sending requests to the Yandex Cloud API via the Yandex Cloud Python SDK

When working with Managed Service for Apache Airflow™, you can use the Yandex Cloud Python SDK to make requests to the Yandex Cloud API. The service supports sending requests to any type of cloud resources. You do not need to set up authentication in the cloud manually. Yandex Cloud Python SDK authenticates via the IAM token of the service account linked to the Apache Airflow™ cluster.

As an example, we use a directed acyclic graph (DAG) which submits a request to the Yandex Cloud API. The request returns a list of virtual machines in the folder where the Apache Airflow™ cluster is created.

To use the Yandex Cloud Python SDK to send requests to the Yandex Cloud API:

1. Prepare your infrastructure.
2. Prepare the DAG file and run the graph.
3. Check the result.

If you no longer need the resources you created, delete them.

Prepare the infrastructurePrepare the infrastructure

1. Create a service account named airflow-sa with the following roles:

   • compute.viewer
   • managed-airflow.integrationProvider

2. Create an Yandex Object Storage bucket in any configuration.

3. Create a Managed Service for Apache Airflow™ cluster with the following parameters:

   • Service account: airflow-sa
   • Bucket name: Name of the new bucket

4. Create a VM in any configuration.

Prepare the DAG file and run the graphPrepare the DAG file and run the graph

1. Create a local file named test_python_sdk.py and copy the following script to it:

   test_python_sdk.py

   from airflow import DAG
   from airflow.models import Connection
   from airflow.operators.python_operator import PythonOperator
   from datetime import datetime
   
   import yandexcloud
   from yandex.cloud.compute.v1.instance_service_pb2 import (
       ListInstancesRequest,
   )
   from yandex.cloud.compute.v1.instance_service_pb2_grpc import InstanceServiceStub
   
   def list_instances():
       conn = Connection.get_connection_from_secrets("yandexcloud_default")
       folder_id = conn.extra_dejson.get('folder_id')
   
       sdk = yandexcloud.SDK()
       instance_service = sdk.client(InstanceServiceStub)
       response = instance_service.List(ListInstancesRequest(folder_id=folder_id))
       print("instances: ", response.instances)
   
   with DAG(
       dag_id='test_python_sdk',
       start_date=datetime(2024, 5, 24),
       schedule="@once",
       catchup=False,
   ) as dag:
       PythonOperator(
           task_id='list_instances',
           python_callable=list_instances,
       )
   

   To authenticate in the cloud, the IAM token of the service account attached to the Apache Airflow™ cluster is used. The yandexcloud.SDK() object is created with default parameters and automatically populated with the data required to authenticate with the IAM token.

2. Upload the test_python_sdk.py DAG file to the bucket you created earlier. This will automatically create a graph with the same name in the Apache Airflow™ web interface.

3. Open the Apache Airflow™ web interface.

4. Make sure a new graph named test_python_sdk has appeared in the DAGs section.

   It may take a few minutes to upload a DAG file from the bucket.

5. To run the graph, click in the line with its name.

Check the resultCheck the result

To check the result in the Apache Airflow™ web interface:

1. In the DAGs section, open the test_python_sdk graph.
2. Go to the Grid section.
3. Select the list_instances task.
4. Go to Logs.
5. Make sure the logs list the virtual machines from the folder where the Apache Airflow™ cluster is created. This means the query was successful.

Delete the resources you createdDelete the resources you created

Some resources are not free of charge. Delete the resources you no longer need to avoid paying for them:

1. Service account
2. Object Storage bucket
3. Managed Service for Apache Airflow™ cluster
4. Virtual machine