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


class MyOperator(BaseOperator):
    @apply_defaults
    def __init__(self,
                 message,
                 *args, **kwargs):
        super(MyOperator, self).__init__(*args, **kwargs)
        self.message = message

    def
 
 execute(self, context):
        print("Message:[%s]" % self.message)

    def on_kill(self):
        print("Killed MyOperator")


# Defining the plugin class

class AirflowTestPlugin(AirflowPlugin):
    name = "myplugin"
    operators = [MyOperator]

from airflow.{type, like "operators", "sensors"}.{name specificed inside the plugin class} import *

from datetime import datetime
from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import PythonOperator
from airflow.operators.bash_operator import BashOperator

from airflow.operators.myplugin import MyOperator

def print_hello():
    return 'Hello world!'

dag = DAG('hello_world', description='Simple tutorial DAG',
          schedule_interval='10 * * * *',
          start_date=datetime(2017, 3, 20), catchup=False)

dummy_operator = MyOperator(task_id='dummy_task', message="Hello World", dag=dag)

hello_operator = PythonOperator(task_id='hello_task', python_callable=print_hello, dag=dag)

t1 = BashOperator(
    task_id='print_date',
    bash_command='date',
    dag=dag)


dummy_operator >> hello_operator >> t1