ExternalPythonOperator got function not defined error

[yeangPan]

Goal: to use ExternalPythonOperator in docker virtual env.

Procedure:

1. rebuild image from officail and according to suggestions from Question: virtual env can't be used in docker with ExternalPythonOperator #34039 (reply in thread)
   I finally got airflow in docker venv running.

2. I write a dag:

from airflow import DAG
from airflow.operators.python import ExternalPythonOperator
import pendulum
from omsmanager import run

with DAG(dag_id="test_dag", 
         catchup=False,
         schedule="0 0 * * *",
         start_date=pendulum.datetime(2021, 1, 1, tz="UTC")) as dag:
    virtualenv_task = ExternalPythonOperator(
        task_id="test_task",
        queue="controller",
        python="/opt/airflow/feature/bin/python",
        python_callable=run,
        op_kwargs = {'module':'datastack','func':'deploy'},
        # requirements=["boto3>=1.17.43"],
        # system_site_packages=False,
        dag=dag,
    )

the omsmanager was build by me, which was installed in the used docker venv. when I run the function run outside of docker and airflow, It runned normally, I got result expected, when I wrote the dag, the task failed with error msg:

what confused me was that the function I used was in the same py file:

def getfunc(mod:str, func:str)->Callable:
    try:
        sys.path.append(CODEBASE)
        module = __import__(mod)
        func = getattr(module, func)
        return func
    except:
        raise traceback.format_exc()
    finally:
        sys.path.pop(-1)

def run(module:str,func:str, *args, **kwargs):
    function = getfunc(module, func)
    return function(*args, **kwargs)

what I want to realize is to call funcition in mod(pyfile) through run function, is there any better way to do it, or what's the best practices?

[Taragolis]

Did you try to import required dependency inside of python callable?

from airflow import DAG
from airflow.operators.python import ExternalPythonOperator
import pendulum


def venv_callable(module="", func=""):
    from omsmanager import run

    run(module=module, func=func)


with DAG(
    dag_id="test_dag", 
    catchup=False,
    schedule="0 0 * * *",
    start_date=pendulum.datetime(2021, 1, 1, tz="UTC")
) as dag:
    virtualenv_task = ExternalPythonOperator(
        task_id="test_task",
        queue="controller",
        python="/opt/airflow/feature/bin/python",
        python_callable=venv_callable,
        op_kwargs = {'module':'datastack','func':'deploy'},
        # requirements=["boto3>=1.17.43"],
        # system_site_packages=False,
        dag=dag,
    )