CODEWITHSUNDEEP
python-3.xairflow
Kay
Kay

Reputation: 2332

Airflow 2.0 - AttributeError: 'MyOperator' object has no attribute 'kwargs'

I am trying to write a custom operator for Airflow 2.0 and I can't seem to understand why the operator will not recognise the kwargs parameter.

This is my custom operator file

from airflow.models.baseoperator import BaseOperator
from airflow.utils.decorators import apply_defaults

class MyOperator(BaseOperator):

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

    def execute(self, context):

        return self.kwargs

And this is my dag:

from datetime import timedelta

from airflow import DAG
from airflow.utils.dates import days_ago
from operators.custom import MyOperator

args = {
    'owner': 'airflow',
}

with DAG(
    dag_id='ex_operator',
    default_args=args,
    schedule_interval='0 0 * * *',
    start_date=days_ago(1),
    dagrun_timeout=timedelta(minutes=60)
) as dag:

    custom_ops = MyOperator(
        task_id = 'myop_id',
        name = 'me',
        params = {
            'lib': 'rainy'
        }
    )

When I run this I get the error:

Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/models/taskinstance.py", line 1086, in _run_raw_task
    self._prepare_and_execute_task_with_callbacks(context, task)
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/models/taskinstance.py", line 1260, in _prepare_and_execute_task_with_callbacks
    result = self._execute_task(context, task_copy)
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/models/taskinstance.py", line 1300, in _execute_task
    result = task_copy.execute(context=context)
  File "/home/airflow/plugins/operators/custom.py", line 33, in execute
    return self.kwargs

AttributeError: 'MyOperator' object has no attribute 'kwargs'

Is there something I am doing wrong, and why would this not recognize kwargs that would have been recognised in any normal Python class?

Upvotes: 1

Views: 5547

Answers (1)

Kay
Kay

Reputation: 2332

args and kwargs weren't assigned and the super class wasn't doing that either. This fixed it.

class MyOperator(BaseOperator):

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

Update: In the BaseOperator super class, I saw this:

super().__init__()
        if kwargs:
            if not conf.getboolean('operators', 'ALLOW_ILLEGAL_ARGUMENTS'):
                raise AirflowException(
                    "Invalid arguments were passed to {c} (task_id: {t}). Invalid "
                    "arguments were:\n**kwargs: {k}".format(c=self.__class__.__name__, k=kwargs, t=task_id),
                )
            warnings.warn(
                'Invalid arguments were passed to {c} (task_id: {t}). '
                'Support for passing such arguments will be dropped in '
                'future. Invalid arguments were:'
                '\n**kwargs: {k}'.format(c=self.__class__.__name__, k=kwargs, t=task_id),
                category=PendingDeprecationWarning,
                stacklevel=3,
            )

At the moment, you would have to set allow_illegal_arguments in the airflow.cfg file to true to be able to pass additional/unused arguments to BaseOperator. However, it looks like Airflow has plans of deprecating it in the future.

Upvotes: 2

Related Questions