Customize Operator Extra Links

Airflow includes plugins for customizing the Airflow UI. One small but impactful customization is adding extra links in the Details view for operators in the UI. These extra links can point to static websites, such as documentation for the operator, or dynamic links created from information during the task instance run.

This tutorial shows how to add both static and dynamic extra links using the AirflowExtraLinkPlugin to existing and custom operators.

After you complete this tutorial, you’ll be able to:

• Add a static operator extra link to any operator using an Airflow plugin.
• Modify an existing operator to push an additional value to XCom.
• Add a dynamic operator extra link to any operator using an Airflow plugin.

Time to complete

This tutorial takes approximately 1 hour to complete.

Assumed knowledge

To get the most out of this tutorial, make sure you have an understanding of:

• Navigating an Airflow Project. See Get started with Apache Airflow.
• Airflow Plugins. See Import plugins to Airflow.
• Airflow Connections. See Manage connections in Apache Airflow.
• Intermediate knowledge of Python. See the official Python documentation.

Prerequisites

• The Astro CLI.

Step 1: Create an Astro project

Set up Airflow by creating a new Astro project:

1
$ mkdir astro-extra-link-tutorial && cd astro-extra-link-tutorial
2
$ astro dev init

Step 2: Create a DAG using the SimpleHttpOperator

You’ll first add a static link to the SimpleHttpOperator which goes to the Mozilla HTTP documentation. This link will appear on every task instance created by this operator.

1. Create a new Python file named plugin_test_dag.py in the dags folder of your Airflow project.

2. Copy and paste the following DAG code into your file:

   1
   from airflow.models.dag import DAG
   2
   from airflow.providers.http.operators.http import SimpleHttpOperator
   3
   from pendulum import datetime
   4
   5
   with DAG(
   6
       dag_id="plugin_test_dag",
   7
       start_date=datetime(2022, 11, 1),
   8
       schedule=None,
   9
       catchup=False
   10
   ):
   11
   12
       call_api_simple = SimpleHttpOperator(
   13
           task_id="call_api_simple",
   14
           http_conn_id="random_user_api_conn",
   15
           method="GET"
   16
       )

This DAG has one SimpleHttpOperator task that posts a GET request to an API as defined in the random_user_api_conn connection.

Step 3: Add a static operator extra link

Create an Airflow plugin to add an extra link to the operator.

1. Create a new Python file named my_extra_link_plugin.py in the plugins folder of your Airflow project.

2. Copy paste the following code into your file.

   1
   from airflow.plugins_manager import AirflowPlugin
   2
   from airflow.models.baseoperator import BaseOperatorLink
   3
   from airflow.providers.http.operators.http import SimpleHttpOperator
   4
   5
   # define the extra link
   6
   class HTTPDocsLink(BaseOperatorLink):
   7
       # name the link button in the UI
   8
       name = "HTTP docs"
   9
   10
       # add the button to one or more operators
   11
       operators = [SimpleHttpOperator]
   12
   13
       # provide the link
   14
       def get_link(self, operator, *, ti_key=None):
   15
           return "https://developer.mozilla.org/en-US/docs/Web/HTTP"
   16
   17
   # define the plugin class
   18
   class AirflowExtraLinkPlugin(AirflowPlugin):
   19
       name = "extra_link_plugin"
   20
       operator_extra_links = [
   21
           HTTPDocsLink(),
   22
       ]

This script accomplishes the following:

• Defines an operator extra link called HTTPDocsLink which will create an extra link button with the name HTTP docs. Customize this string to change the name on the button displayed in the UI.
• Adds the SimpleHttpOperator to the list of operators this extra link will be applied to. You can add as many operators as you’d like, including custom operators.
• Defines the get_link() method which determines the website the operator extra link will link to. You can change this function to any Python function that returns a valid link. See Step 9 for instructions on how to make this link dynamically change between task instances.
• Creates an instance of the AirflowPlugin class which will be automatically picked up by Airflow to install the plugin named extra_link_plugin in your Airflow instance.
• Adds the HTTPDocsLink plugin to the extra_link_plugin. You can add several operator extra links to the same Airflow plugin.

Step 4: Add an HTTP connection

1. Run astro dev start in your Astro project directory to start up Airflow. If your Airflow instance is already running, use astro dev restart to restart it in order to load any changes made in the plugins folder.

2. Add an HTTP connection called random_user_api_conn to http://randomuser.me/api/ in the Airflow UI. This API will return data about a randomly generated user persona. Feel free to use a different API, the content returned will not be relevant for this tutorial. Learn more about connections in the Manage connections in Apache Airflow guide.

   

Step 5: Use your static operator extra link

1. Run the plugins_test_dag.

2. In the Grid view, click the green square representing the successful run of the call_api_simple task. Select the Details tab and scroll down to see the extra link button called HTTP docs.

   

3. Click on the button to visit the HTTP docs on Mozilla.

Step 6: Create a custom operator

Another core feature of extra links is that you can dynamically generate them based on information returned by an operator at run time. The second half of this tutorial will cover how to modify an operator to push the value you need to XComs and retrieve that value for use in an extra link.

1. Create a new file called cat_http.py in the include folder of your Airflow project.

2. Copy the following code into the file.

   1
   from airflow.providers.http.operators.http import SimpleHttpOperator
   2
   from airflow.providers.http.hooks.http import HttpHook
   3
   from airflow.utils.operator_helpers import determine_kwargs
   4
   from airflow.exceptions import AirflowException
   5
   6
   class CatHttpOperator(SimpleHttpOperator):
   7
   8
       # initialize with identical arguments to the parent class
   9
       def __init__(self, **kwargs):
   10
           super().__init__(**kwargs)
   11
   12
       def execute(self, context):
   13
   14
           http = HttpHook(
   15
               self.method,
   16
               http_conn_id=self.http_conn_id,
   17
               auth_type=self.auth_type,
   18
               tcp_keep_alive=self.tcp_keep_alive,
   19
               tcp_keep_alive_idle=self.tcp_keep_alive_idle,
   20
               tcp_keep_alive_count=self.tcp_keep_alive_count,
   21
               tcp_keep_alive_interval=self.tcp_keep_alive_interval,
   22
           )
   23
   24
           self.log.info("Calling HTTP method")
   25
   26
           response = http.run(
   27
               self.endpoint, self.data, self.headers, self.extra_options
   28
           )
   29
           if self.log_response:
   30
               self.log.info(response.text)
   31
           if self.response_check:
   32
               kwargs = determine_kwargs(self.response_check, [response], context)
   33
               if not self.response_check(response, **kwargs):
   34
                   raise AirflowException("Response check returned False.")
   35
           if self.response_filter:
   36
               kwargs = determine_kwargs(self.response_filter, [response], context)
   37
               return self.response_filter(response, **kwargs)
   38
   39
           # pushing the HTTP status response to XComs
   40
           context["ti"].xcom_push(key="status_code", value=response.status_code)
   41
   42
           return response.text

   This code defines a custom version of the SimpleHttpOperator called the CatHttpOperator. This operator has a one-line customization before the return statement of the .execute() method:context["ti"].xcom_push(key="status_code", value=response.status_code).

   This line pushes the status_code attribute of the response object to XComs where it can be called from your plugin.

3. Add an empty Python file called __init__.py to your include folder. This file enables module imports from the folder.

Step 7: Create a DAG with your custom operator

1. In the plugin_test_dag.py file import the CatHttpOperator by adding the following import statement at the start of the code:

   1
   from include.cat_http import CatHttpOperator

2. Create a second task in the existing DAG context using the new operator with the code snippet below.

   1
   call_api_cat = CatHttpOperator(
   2
       task_id="call_api_cat",
   3
       http_conn_id="random_user_api_conn",
   4
       method="GET"
   5
   )

This task will post the same GET request to the API you defined with the connection ID random_user_api_conn as the first task, but this time using the CatHttpOperator.

Step 8: Run your DAG and view modified XComs

1. Run the plugins_test_dag, which now consists of two tasks.

2. Select the latest run of the call_api_cat task in the Grid view and click on the XCom tab to view XComs returned by this task.

   

3. Verify that the XCom for this task instance contains an entry for status_code. In the screenshot below the HTTP status code returned was 200.

   

Step 9: Add a dynamic extra link to your custom operator

Next, you will create a dynamic extra link using an Airflow plugin by following these steps:

1. Open my_extra_link_plugin.py in your plugins folder.

2. Add the following import statements at the start of the file:

   1
   from include.cat_http import CatHttpOperator
   2
   from airflow.models import XCom

3. Copy paste the following code below the definition of the HTTPDocsLink class and above the definition of the AirflowExtraLinkPlugin class.

   1
   class CatLink(BaseOperatorLink):
   2
       # name the link button in the UI
   3
       name = "HTTP cat"
   4
   5
       # add the button to one or more operators
   6
       operators = [CatHttpOperator]
   7
   8
       # provide the link
   9
       def get_link(self, operator, *, ti_key=None):
   10
           status_code = XCom.get_value(key="status_code", ti_key=ti_key) or ""
   11
           return f"https://http.cat/{status_code}"

   This code creates a class called CatLink derived from BaseOperatorLink. The .get_link() method retrieves the status_code you pushed to XCom and appends it to the HTTP cat Api link.

4. Update the operator_extra_links list in the AirflowExtraLinkPlugin class with the new CatLink(). The class should look like this:

   1
   class AirflowExtraLinkPlugin(AirflowPlugin):
   2
       name = "extra_link_plugin"
   3
       operator_extra_links = [
   4
           CatLink(), # add this line
   5
           HTTPDocsLink()
   6
       ]

5. Save the file and restart your Airflow instance with astro dev restart.

Step 10: See your new extra link in action

Your second extra link has now been added to the CatHttpOperator.

1. In the Airflow UI, run plugins_test_dag again.

2. Navigate to the Graph View and click on the call_api_cat task.

3. Click the HTTP Cat button to find the response of your last API call illustrated with a fitting cat.

   

4. (Optional) See all your plugins listed under Admin -> Plugins.

   

Conclusion

Congratulations! You added two operator extra links as an Airflow plugin. On the way you also learned how to modify an existing operator to pass an additional value to XCom.