Migrate DAG979Sensor: Leverage Public Airflow API

Nov 6, 2025 by SLV Team 50 views

Hey folks! Let's dive into a crucial update for our Airflow setup. We're talking about revamping the DAG979Sensor to leverage the power of the public Airflow API. This is not just a simple tweak, it's a necessary evolution to keep our workflows running smoothly, especially with the transition to Airflow 3. So, buckle up as we explore the 'why' and 'how' of this migration, ensuring our digital pipelines stay robust and efficient. We will mainly talk about the challenges of the current DAG979Sensor and how we can use the Airflow API to solve this problem.

The Problem: DAG979Sensor and Airflow 3 Compatibility

Alright, let's get down to the nitty-gritty. The current DAG979Sensor is designed to monitor the state and configurations of specific dagruns, namely those associated with the digital_bookplate_979 DAGs. It does this by directly accessing the DagRun model within Airflow. The main issue here is that the DagRun model is no longer directly accessible in Airflow 3. That's a big problem, right? This means the sensor, as it stands, won't function correctly in the newer version of Airflow. We need a way to monitor the dag runs, and also a way to get the state and the conf. We are in a state of crisis, as we need to adapt to the new system, so we can monitor all the digital_bookplate_979 DAGs.

This incompatibility forces us to find an alternative solution. We can't just ignore it and hope for the best. We need to actively find a way to make sure the process works, without the use of DagRun model. We could use another method or, perhaps, create a new one to fit with the Airflow 3 standard. The most common fix is to use the API.

Think of it like this: your old car's engine isn't compatible with the new fuel, so you need a new engine. In our case, the 'engine' is the DAG979Sensor, the 'old fuel' is the DagRun model, and the 'new fuel' is the public Airflow API. The aim of this change is to prevent any disruption in our workflow, especially for the digital_bookplate_979 DAGs. This will help make sure that everything stays on track. The migration is not just about adapting to the new Airflow version, it's about making sure that our workflows are stable and efficient.

So, what is the API? It is a series of REST endpoints. These endpoints make sure that we can interact with the Airflow instance, without using the old models. The endpoint we need is the Get Dag Run endpoint. This API will provide us with all the info that we need. We can get the state, the configuration, and other useful data that can help us monitor the state of the DAGs.

Solution: Leveraging the Public Airflow API

The good news is that there's a straightforward solution: utilize the public Airflow API. Specifically, we'll be focusing on the Get Dag Run endpoint. This API endpoint allows us to retrieve detailed information about a specific DAG run, including its state and configuration. This is exactly what the DAG979Sensor needs to function. It will work by giving us a JSON response. This response is an array of information that we can parse and use in the poll_979s_dags task.

Let's break down how this works. We will be sending a request to the API, and the API will return us a JSON response containing the info. This means we will get information about the dag_run_id, dag_id, logical_date, and many other attributes that describe the dag run. The most important fields for the DAG979Sensor are the state and the conf. The sensor polls the API regularly, checks the state, and waits until the state is success or failed. We will need to rewrite the code to make it work. The conf field gives us the configuration details of the DAG run, which is also very helpful. Using the Airflow API makes sure that we can monitor the state of the DAG runs, without using the DagRun model.

Here's an example of the JSON response you can expect from the API:

{
  "dag_run_id": "string",
  "dag_id": "string",
  "logical_date": "2019-08-24T14:15:22Z",
  "queued_at": "2019-08-24T14:15:22Z",
  "start_date": "2019-08-24T14:15:22Z",
  "end_date": "2019-08-24T14:15:22Z",
  "duration": 0,
  "data_interval_start": "2019-08-24T14:15:22Z",
  "data_interval_end": "2019-08-24T14:15:22Z",
  "run_after": "2019-08-24T14:15:22Z",
  "last_scheduling_decision": "2019-08-24T14:15:22Z",
  "run_type": "backfill",
  "state": "queued",
  "triggered_by": "cli",
  "triggering_user_name": "string",
  "conf": {},
  "note": "string",
  "dag_versions": [
    {
      "id": "497f6eca-6276-4993-bfeb-53cbbbba6f08",
      "version_number": 0,
      "dag_id": "string",
      "bundle_name": "string",
      "bundle_version": "string",
      "created_at": "2019-08-24T14:15:22Z",
      "dag_display_name": "string",
      "bundle_url": "string"
    }
  ],
  "bundle_version": "string",
  "dag_display_name": "string"
}

As you can see, the response contains a wealth of information. This includes the state of the DAG run, which is crucial for the sensor, along with other metadata that can be useful for debugging and monitoring. This API will provide us the data that we need to keep our process running, and also makes it so that we can use it with Airflow 3. Using the API makes it so that we don't depend on the implementation details of the model, which can change in the future.

Implementation Steps: Configuring JWT with FAB Auth Manager

Now, let's talk about the practical side of things. To successfully use the public Airflow API, we'll need to configure our JWT (JSON Web Token). We will be using the FAB (Flask-AppBuilder) auth manager, since our plugins use the FAB provider. This will make it easier to authorize our requests to the API. This will guarantee that only authorized users can monitor the state of the DAGs.

First, we need to generate a JWT. This involves creating a set of credentials and using them to generate a token. The API documentation has a guide on this. Once the token is generated, we must store it securely. We do not want to expose the JWT in the code, which is a big security risk. We must use secure configurations to save it. When our DAG979Sensor makes requests to the API, it will include the JWT in the Authorization header. This will tell the API that the request is authorized. This allows us to use the public API, which is a big leap forward. Remember to also handle the situation if the token expires. We must refresh it to make sure that our sensor works.

Here are the key steps involved:

JWT Generation: You'll need to generate a JWT using the FAB auth manager. This is your key to accessing the API. Follow the steps in the official documentation to generate your token. This is the first step to being able to interact with the API, and will need to be done correctly.
Secure Storage: Store the JWT securely. Never hardcode it into your sensor code. Use environment variables or a secrets management system. Keeping it safe is essential for maintaining the security of the Airflow instance.
API Requests: When making requests to the API, include the JWT in the Authorization header. This tells the API that your request is authorized. This will make the API see that you are an authorized user, and that you have permissions to see the state of the DAGs.
Error Handling: Implement robust error handling to deal with potential API issues, such as invalid credentials or network problems. You should be prepared to handle all the potential situations. This will make sure that the system is stable, even when there are problems with the API.

Code Example: (Illustrative - Needs Adaptation)

I can't provide the complete code here, but I can give you a starting point. This is an example to illustrate the process. It's important that you adapt it to your specific needs. The code below shows how to send a request, but you will need to fill in the implementation for your code. The exact implementation depends on how you are using the DAG979Sensor. Here's a basic outline:

import requests
import os

class DAG979Sensor:
  def __init__(self, dag_run_id, api_endpoint, jwt_token):
    self.dag_run_id = dag_run_id
    self.api_endpoint = api_endpoint
    self.jwt_token = jwt_token

  def poke(self, context):
    headers = {
      "Authorization": f"Bearer {self.jwt_token}",
      "Content-Type": "application/json"
    }
    url = f"{self.api_endpoint}/{self.dag_run_id}"

    try:
      response = requests.get(url, headers=headers)
      response.raise_for_status()
      dag_run_info = response.json()
      state = dag_run_info["state"]
      print(f"DAG Run {self.dag_run_id} is in state: {state}")
      return state in ["success", "failed"]
    except requests.exceptions.RequestException as e:
      print(f"API request failed: {e}")
      return False

# Example Usage (Adapt this to your needs)
if __name__ == "__main__":
    dag_run_id = "your_dag_run_id"  # Replace with actual DAG run ID
    api_endpoint = "http://your-airflow-instance/api/v1/dags/your_dag_id/dagRuns"  # Replace with your API endpoint
    jwt_token = os.environ.get("AIRFLOW_API_TOKEN")  # Get the token from an environment variable

    if not jwt_token:
      print("Error: AIRFLOW_API_TOKEN environment variable not set.")
    else:
      sensor = DAG979Sensor(dag_run_id, api_endpoint, jwt_token)
      # The sensor will now poll the API
      # You can integrate this with your Airflow DAG
      # For example, use it within a PythonOperator or a custom sensor
      # to wait for the DAG to complete
      # For example:
      # if sensor.poke(context={}):
      #   print("DAG run completed successfully")
      # else:
      #   print("DAG run failed or is still running")
      sensor.poke(context={})

This is the basis of our approach. We will be using the requests library to send a GET request to the API. We'll be sending the JWT in the Authorization header. This shows that we are authorized. The API will respond with JSON that we can parse to get the state. We'll include basic error handling to make sure that the program is resilient. Adapt it to your code.

Remember to replace the placeholders with your actual values. Also, you will need to add exception handling for potential issues. The code is a basic framework and needs to be adapted to your system.

Benefits of the Migration

Migrating to the public Airflow API offers several advantages, solidifying our infrastructure and making it more future-proof. Let's explore these benefits:

Airflow 3 Compatibility: The most immediate benefit is compatibility with Airflow 3. Using the public API ensures that the DAG979Sensor will work seamlessly with the newer version. This means that we can upgrade to the new system, without having to worry about breaking the system.
Future-Proofing: We are less dependent on internal model implementations. This will give us long-term stability and resilience, as it's less prone to breaking when Airflow updates its internal structure. This ensures the workflow stays stable, even as Airflow evolves. No need to worry about the future.
Enhanced Security: The use of JWTs and secure configurations will improve the security of our Airflow instance. This means that we will be less likely to have issues with security breaches, since the system will be more secure. Protect your data.
Maintainability: Using the API simplifies the code. This makes it easier to understand, maintain, and debug. A cleaner, more focused sensor is more manageable in the long run. We make it simple to update the code, and solve any problems.
Standardization: We use the standard API, which means that we are using the official interface to communicate with Airflow. This will make it easier to follow best practices, and use standardized methods. Standardizing our approach will help us in the long term.

Conclusion: A Step Towards a Robust Airflow Setup

Migrating the DAG979Sensor to use the public Airflow API is a necessary and beneficial change. It ensures compatibility with Airflow 3, improves security, and provides a more maintainable and future-proof solution. While it requires some upfront effort in terms of configuration and code adaptation, the long-term benefits in terms of stability, security, and maintainability are well worth the investment. It's a key step in keeping our data pipelines robust and efficient. By following the steps outlined, you can successfully migrate your sensor and ensure your workflows continue to run smoothly. Happy coding!