Creating custom links

[pabloalcain]

What do you see as an issue?

I'm trying to recreate the Define an operator extra link guide, but I cannot get it to work. From the documentation, I understood (perhaps incorrectly) that creating this DAG should be enough:

from airflow import DAG
from airflow.models.baseoperator import BaseOperator, BaseOperatorLink
from airflow.plugins_manager import AirflowPlugin
from airflow.models.taskinstancekey import TaskInstanceKey
from airflow.utils.dates import days_ago

class GoogleLink(BaseOperatorLink):
    name = "Google"

    def get_link(self, operator: BaseOperator, *, ti_key: TaskInstanceKey):
        return "https://www.google.com"


class MyFirstOperator(BaseOperator):

    operator_extra_links = (GoogleLink(),)

    def __init__(self, **kwargs):
        super().__init__(**kwargs)

    def execute(self, context):
        self.log.info("Hello World!")


# Defining the plugin class
class AirflowExtraLinkPlugin(AirflowPlugin):
    name = "extra_link_plugin"
    operator_extra_links = [
        GoogleLink(),
    ]

dag = DAG("extra_link_test", start_date=days_ago(2))
with dag:
    MyFirstOperator(task_id="my_first_operator")

but I cannot see the link in the operator, as I expect from the image in the documentation

I feel like I'm doing something wrong, maybe related to how I'm registering the extra link plugin (is what I'm doing enough?), but I cannot find in the documentation how to properly make it work.

Solving the problem

I think that having minimal DAG generation in which we can see the operator rendered in the UI with the extra link can clarify the situation.

Anything else

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[boring-cyborg[bot]]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.

[phanikumv]

Hi @pabloalcain thank you for creating the issue. I will take a closer look at this and come back to you!

[hussein-awala]

@pabloalcain you need to move this code to the plugins folder (doc), otherwise the new link will not be visible for Airflow webserver.

[pabloalcain]

Thanks Hussein! I had tried that the first time and it didn't work, and I realize how it was because I wasn't importing the plugin link correctly. I didn't know that $AIRFLOW_HOME/plugins was already loaded to $PYTHONPATH. So, just for further reference if someone is having the issues i was having: it's only a matter of declaring the link as a plugin

#file: AIRFLOW_HOME/plugins/google_link.py

from airflow.models.baseoperator import BaseOperator
from airflow.models.taskinstance import TaskInstanceKey
from airflow.models.baseoperator import BaseOperatorLink
from airflow.plugins_manager import AirflowPlugin


class GoogleLink(BaseOperatorLink):
    name = "Google"

    def get_link(self, operator: BaseOperator, *, ti_key: TaskInstanceKey):
        return "https://www.google.com"


class AirflowExtraLinkPlugin(AirflowPlugin):
    name = "extra_link_plugin"
    operator_extra_links = [
        GoogleLink(),
    ]

and then importing it directly

#file: AIRFLOW_HOME/dags/extra_link.py
from airflow import DAG
from airflow.models.baseoperator import BaseOperator
from airflow.utils.dates import days_ago

# We can import like this because `plugins/` is already loaded to PYTHONPATH
from google_link import GoogleLink


class MyFirstOperator(BaseOperator):

    operator_extra_links = (GoogleLink(),)

    def __init__(self, **kwargs):
        super().__init__(**kwargs)

    def execute(self, context):
        self.log.info("Hello World!")


dag = DAG("extra_link_test", start_date=days_ago(2))
with dag:
    MyFirstOperator(task_id="my_first_operator")

Thanks again Hussein!