PythonOperator with external module

[lforesta]

I would like to use the PythonOperator with a function that needs to be imported from an external module, but I want to avoid having to install this module on the webserver/scheduler. I have not yet found a clean way to do this.

For example, the following sample DAG is valid only if some_module is available on both the webserver/scheduler and on the worker that serves queue queue_1.

from datetime import datetime, timedelta
from airflow import DAG

from airflow.operators import PythonOperator
from some_module import some_function

default_args = {
    'owner': "someone",
    'start_date': datetime.combine(datetime.today() - timedelta(1), datetime.min.time()),
}

dag = DAG(dag_id="12345",
          description="No description provided",
          catchup=True,
          max_active_runs=1,
          schedule_interval=None,
          default_args=default_args)

task_1 = PythonOperator(task_id='task_1',
                         dag=dag,
                         python_callable=some_function,
                         op_kwargs={}]},
                         queue='queue_1'
                         )

A workaround is to create a custom operator that dynamically imports that module and use that operator instead of the Python one, but there must be a better and more straightforward way to do this.
All examples I find using the PythonOperator only use functions defined in the DAG itself.

[turbaszek]

How about using this as callable?

def my_function(*args, **kwargs):
    from some_module import some_function
    return some_function(*args, **kwargs)

You still have to have the some_module installed on worker but it should not be necessary on scheduler and worker.

[lforesta]

Hi thanks, yes that is eventually what I implemented (in a custom operator instead of in a function). I was hoping there would be a better way to directly use external modules with the PythonOperator (i.e. having them being loaded at runtime on the worker where the task is executed).

It seems to me this would be a basic feature of this operator?

[turbaszek]

having them being loaded at runtime on the worker where the task is executed

If I'm not mistaken that's what happens if you move the import to the callable, isn't it?

[lforesta]

yes but in this case I always need a wrapper around the module I want to use, right?
This is what seems strange to me, that PythonOperator does not allow this out of the box (as far as I have understood). I wanted to double check it this is actually the case, or if I missed something there.

[turbaszek]

@lforesta I think this is the most pythonic approach. Other option I see would be possibly passing required modules by strings which is terrible imho - no support from IDE autocomplete and high possibility of typos.

[lforesta]

Ok thanks a lot for the support.
I'm marking your first comment as answer, since there is no other approach with PythonOperator.