Can't import Airflow plugins

Question

Following Airflow tutorial here. Problem: The webserver returns the following error ``` Broken DAG: [/usr/local/airflow/dags/test_operator.py] cannot import name MyFirstOperator ``` Notes: The directory structure looks like this: ``` airflow_home ├── airflow.cfg ├── airflow.db ├── dags │ └── test_operators.py ├── plugins │ └── my_operators.py └── unittests.cfg ``` I am attempting to import the plugin in 'test_operators.py' like this: ``` from airflow.operators import MyFirstOperator ``` The code is all the same as what is found in the tutorial.

Accepted Answer

After struggling with the Airflow documentation and trying some of the answers here without success, I found this approach from astronomer.io.

As they point out, building an Airflow Plugin can be confusing and perhaps not the best way to add hooks and operators going forward.

Custom hooks and operators are a powerful way to extend Airflow to meet your needs. There is however some confusion on the best way to implement them. According to the Airflow documentation, they can be added using Airflow’s Plugins mechanism. This however, overcomplicates the issue and leads to confusion for many people. Airflow is even considering deprecating using the Plugins mechanism for hooks and operators going forward.

So instead of messing around with the Plugins API I followed Astronomer's approach, setting up Airflow as shown below.

```
dags
└── my_dag.py               (contains dag and tasks)
plugins
├── __init__.py
├── hooks
│   ├── __init__.py
│   └── mytest_hook.py      (contains class MyTestHook)
└── operators
    ├── __init__.py
    └── mytest_operator.py  (contains class MyTestOperator)
```

With this approach, all the code for my operator and hook live entirely in their respective files - and there's no confusing plugin file. All the `__init__.py` files are empty (unlike some equally confusing approaches of putting Plugin code in some of them).

For the `import`s needed, consider how Airflow actually uses the plugins directory:

When Airflow is running, it will add dags/, plugins/, and config/ to PATH

This means that doing `from airflow.operators.mytest_operator import MyTestOperator` probably isn't going to work. Instead `from operators.mytest_operator import MyTestOperator` is the way to go (note the alignment to`from directory/file.py import Class` in my setup above).

Working snippets from my files are shown below.

my_dag.py:

```
from airflow import DAG
from operators.mytest_operator import MyTestOperator
default_args = {....}
dag = DAG(....)
....
mytask = MyTestOperator(task_id='MyTest Task', dag=dag)
....
```

my_operator.py:

```
from airflow.models import BaseOperator
from hooks.mytest_hook import MyTestHook

class MyTestOperator(BaseOperator):
    ....
    hook = MyTestHook(....)
    ....
```

my_hook.py:

```
class MyTestHook():
    ....
```

This worked for me and was much simpler than trying to subclass AirflowPlugin. However it might not work for you if you want changes to the webserver UI:

Note: The Plugins mechanism still must be used for plugins that make changes to the webserver UI.

As an aside, the errors I was getting before this (that are now resolved):

```
ModuleNotFoundError: No module named 'mytest_plugin.hooks.mytest_hook'
ModuleNotFoundError: No module named 'operators.mytest_plugin'
```

Can't import Airflow plugins

Solution

Problem

Related problems