Dynamic Task Mapping Failure with External Python Decorator

[sc-ndrf]

Apache Airflow version

2.10.2

If "Other Airflow 2 version" selected, which one?

No response

What happened?

I am using dynamic task mapping to create two instances of a task make_paths that returns a single string.
The return values are then passed to a task that expects a list of string for further processing.
However, before executing the second task, print_paths, an exception is thrown: ModuleNotFoundError: No module named '***'. The log file is attached.

When the tasks are decorated with @task instead of task.external_python , the DAG runs successfully and produces the expected output.

error.log

What you think should happen instead?

The DAG should run without error also using the task.external_python decorator.

How to reproduce

• Download and unpack the attached zip archive.
• Build and start the Docker containers using docker compose build --no-cache && docker compose up -d
• Open the web GUI at http://localhost:8090/dags/test_dag and run the test DAG. Use airflow:airflow as login.

airflow_external_fail.zip

Operating System

Ubuntu 24.04 as host OS.

Versions of Apache Airflow Providers

none used.

Deployment

Docker-Compose

Deployment details

Docker container based on the official apache/airflow:2.10.2-python3.12 image.

Docker Compose version 2.24.6+ds1-0ubuntu2

Anything else?

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[boring-cyborg[bot]]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.

[potiuk]

Likely - your external venv does not have airflow installed. "***" in log replaces sensitive passwords, and it means that in your system you use "airflow" as a password - so all occurences of "airflow" are replaced in the log. I suggest to use "real" passwords also in testing - leaving default password even while you are stage-testing things is a very bad idea.

[sc-ndrf]

Thank you for the quick reply.

Likely - your external venv does not have airflow installed.

Yes, indeed. The software that is supposed to run in this external venv has rather complicated dependencies, so I would prefer to keep them separate and from the documentation I understood that this should be possible (though I would not have access to stuff in op_kwargs.

Anyways, I found an easy way around by adding an intermediate task using the same venv as airflow:

@task
def reduce(items: list[str]) -> list[str]:
    return items

Which is then inserted into the DAG right before the task run in external python needing the list.

    pathlist = make_paths.partial(base_path=str(BASE_PATH / "xcom")).expand(x=[1, 2])
    reduced = reduce(pathlist)
    print_paths(input_paths=reduced)

I suggest to use "real" passwords also in testing - leaving default password even while you are stage-testing things is a very bad idea.

I'll check and update my settings.

[potiuk]

BTW. For such issues, please use "discussions" not "issues". Issues (as explained when you open one) should be used to report issues with Airflow not to ask for help in debugging your configuration/setup.