using airflow plugin to import external python function

Question

On on_failure_callback I am calling sns method to send notification which is working fine.
I want to declare sns function as a separate function using airflow plugins.

import boto3
from airflow import DAG
from airflow.operators import BashOperator
from datetime import datetime, timedelta

# t1, t2, t3 and t4 are examples of tasks created using operators

def sns(state):
  client = boto3.client('sns',aws_access_key_id='abcd',aws_secret_access_key='abcd',region_name='us-east-1')
  response = client.publish(
    TargetArn='Topicarn',
    Message='Test')

t3 = BashOperator(
    task_id='task_3',
    bash_command='fail',
    on_failure_callback=sns,
    dag=dag)

Thanks,
Rajeev

Have you read the official documentation on Plugins? airflow.apache.org/plugins.html - there is a very nice example there which you can try for yourself. If you have a specific problem while doing that, then you can ask a more specific question. — bosnjak
– bosnjak, Commented Mar 11, 2019 at 12:47
I have read the official documentation.Looking answer for below questions: — Rajeev Reddy
– Rajeev Reddy, Commented Mar 11, 2019 at 13:10
You should update your original question with those points. It will get answered then. The way it is now is too broad — bosnjak
– bosnjak, Commented Mar 11, 2019 at 13:13

bosnjak · Accepted Answer · 2019-03-11 13:17:28Z

1

(replying to actual questions posted to comments, please move those to the original question)

Since this will be a basically a notification operator, you should inherit from BaseOperator.
Plugin folder is usually located in the project's root, next to the dag folder.
As for the plugin itself, once you inherit from AirflowPlugin, it will be accessible for import as airflow.operators.my_operator. You can use it as any other operator.

answered Mar 11, 2019 at 13:17

bosnjak

8,6342 gold badges25 silver badges47 bronze badges

Sign up to request clarification or add additional context in comments.

Comments

Rajeev Reddy · Accepted Answer · 2019-03-12 09:59:18Z

Below code worked for me :

plugin code :

import boto3
from airflow.models import BaseOperator
from airflow.plugins_manager import AirflowPlugin
from airflow.utils.decorators import apply_defaults

class snsoperator(BaseOperator):
  @apply_defaults
  def __init__(self, *args, **kwargs):

   super(snsoperator, self).__init__(*args, **kwargs)

  def __call__(self, *args, **kwargs):
    print("calling execute")
    client = boto3.client('sns', aws_access_key_id='abc',
                              aws_secret_access_key='abc', region_name='us-east-1')
    response = client.publish(
            TargetArn='abc',
            Message='Test')

class snsplugin(AirflowPlugin):
  name = "snsplugin"
  operators = [snsoperator]

DAG code :

import boto3
from airflow import DAG
from airflow.operators import BashOperator
from datetime import datetime, timedelta
from airflow.operators.snsplugin import snsoperator

t3 = BashOperator(
    task_id='task_3',
    bash_command='fail',
    on_failure_callback=snsoperator(task_id='sns'),
    dag=dag)

Collectives™ on Stack Overflow

using airflow plugin to import external python function

2 Answers 2

Comments

Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

2 Answers 2

Comments

Comments

Your Answer

Sign up or log in

Post as a guest

Related