external_python_task decorator doesn't work if airflow isn't in external python env

[JCoder01]

Apache Airflow version

Other Airflow 2 version (please specify below)

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

2.7.3

What happened?

2024-02-01T15:01:41.731+0000] {process_utils.py:186} INFO - Output:
[2024-02-01T15:01:41.912+0000] {process_utils.py:190} INFO - Traceback (most recent call last):
[2024-02-01T15:01:41.912+0000] {process_utils.py:190} INFO -   File "/tmp/tmdfow99m6c/script.py", line 17, in <module>
[2024-02-01T15:01:41.913+0000] {process_utils.py:190} INFO -     @external_python_task(task_id = 'task_runner', python=PYENV, expect_airflow=False)
[2024-02-01T15:01:41.913+0000] {process_utils.py:190} INFO -      ^^^^^^^^^^^^^^^^^^^^
[2024-02-01T15:01:41.913+0000] {process_utils.py:190} INFO - NameError: name 'external_python_task' is not defined

When generating python_callable_source the airflow decorator is included in the source

What you think should happen instead?

Airflow should strip the decorator from the function definition

How to reproduce

Create python env that doesn't install airflow (PYENV)

run something to effect of:

from airflow.decorators import external_python_task
from airflow.models.taskinstance import TaskInstance

@external_python_task(python=PYENV, expect_airflow=False)
def task_runner():
    pass


with DAG(dag_id="test", start_date=pendulum.now()):
    task_runner()

task = dag.get_task('task_runner')
t1 = TaskInstance(task=task, execution_date=dag.get_latest_execution_date())
t1.render_templates()
task.execute(t1.get_template_context())

Operating System

linux

Versions of Apache Airflow Providers

No response

Deployment

Official Apache Airflow Helm Chart

Deployment details

No response

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

[Taragolis]

You should use @task.external_python instead, the function which you use is a private

airflow/airflow/decorators/external_python.py

Lines 43 to 49 in 3904206

	Wrap a callable into an Airflow operator to run via a Python virtual environment.
	Accepts kwargs for operator kwarg. Can be reused in a single DAG.
	This function is only used during type checking or auto-completion.
	:meta private:

I can't reproduce it locally on Airflow 2.8 it works as expected

from datetime import datetime, timezone
from pathlib import Path

from airflow import DAG
from airflow.decorators import task, external_python_task


ANOTHER_VENV = Path("/Users/taragolis/.pyenv/versions/llama-runtime-py311/bin/python")
assert ANOTHER_VENV.exists()
assert ANOTHER_VENV.is_file()


@task.external_python(python=str(ANOTHER_VENV), expect_airflow=False)
def task_runner():
    import sys

    if sys.version_info < (3, 10):
        from importlib_metadata import version as metadata_version
    else:
        from importlib.metadata import version as metadata_version

    try:
        ver = metadata_version("apache-airflow")
    except ModuleNotFoundError:
        print("`apache-airflow` package not found")
    else:
        print(f"found `apache-airflow` package version {ver}")

    return sys.executable


with DAG(
    dag_id="issue_37136",
    start_date=datetime(2024, 1, 1, tzinfo=timezone.utc),
    schedule=None,
    tags=["venv", "37136"],
):
    task_runner()

[2024-02-02, 10:57:31 UTC] {taskinstance.py:1957} INFO - Dependencies all met for dep_context=non-requeueable deps ti=<TaskInstance: issue_37136.task_runner manual__2024-02-02T10:57:28.849895+00:00 [queued]>
[2024-02-02, 10:57:31 UTC] {taskinstance.py:1957} INFO - Dependencies all met for dep_context=requeueable deps ti=<TaskInstance: issue_37136.task_runner manual__2024-02-02T10:57:28.849895+00:00 [queued]>
[2024-02-02, 10:57:31 UTC] {taskinstance.py:2171} INFO - Starting attempt 1 of 1
[2024-02-02, 10:57:31 UTC] {taskinstance.py:2192} INFO - Executing <Task(_PythonExternalDecoratedOperator): task_runner> on 2024-02-02 10:57:28.849895+00:00
[2024-02-02, 10:57:31 UTC] {standard_task_runner.py:60} INFO - Started process 35155 to run task
[2024-02-02, 10:57:31 UTC] {standard_task_runner.py:87} INFO - Running: ['airflow', 'tasks', 'run', 'issue_37136', 'task_runner', 'manual__2024-02-02T10:57:28.849895+00:00', '--job-id', '17', '--raw', '--subdir', 'DAGS_FOLDER/foo.py', '--cfg-path', '/var/folders/g7/vy97v5m51_v9hdpf_qxqzqvr0000gn/T/tmp8tbefl9b']
[2024-02-02, 10:57:31 UTC] {standard_task_runner.py:88} INFO - Job 17: Subtask task_runner
[2024-02-02, 10:57:31 UTC] {task_command.py:423} INFO - Running <TaskInstance: issue_37136.task_runner manual__2024-02-02T10:57:28.849895+00:00 [running]> on host andreys-mbp.lan
[2024-02-02, 10:57:31 UTC] {taskinstance.py:2481} INFO - Exporting env vars: AIRFLOW_CTX_DAG_OWNER='airflow' AIRFLOW_CTX_DAG_ID='issue_37136' AIRFLOW_CTX_TASK_ID='task_runner' AIRFLOW_CTX_EXECUTION_DATE='2024-02-02T10:57:28.849895+00:00' AIRFLOW_CTX_TRY_NUMBER='1' AIRFLOW_CTX_DAG_RUN_ID='manual__2024-02-02T10:57:28.849895+00:00'
[2024-02-02, 10:57:31 UTC] {process_utils.py:182} INFO - Executing cmd: /Users/taragolis/.pyenv/versions/llama-runtime-py311/bin/python /var/folders/g7/vy97v5m51_v9hdpf_qxqzqvr0000gn/T/venv-calld6yc0zua/script.py /var/folders/g7/vy97v5m51_v9hdpf_qxqzqvr0000gn/T/venv-calld6yc0zua/script.in /var/folders/g7/vy97v5m51_v9hdpf_qxqzqvr0000gn/T/venv-calld6yc0zua/script.out /var/folders/g7/vy97v5m51_v9hdpf_qxqzqvr0000gn/T/venv-calld6yc0zua/string_args.txt /var/folders/g7/vy97v5m51_v9hdpf_qxqzqvr0000gn/T/venv-calld6yc0zua/termination.log
[2024-02-02, 10:57:31 UTC] {process_utils.py:186} INFO - Output:
[2024-02-02, 10:57:31 UTC] {process_utils.py:190} INFO - `apache-airflow` package not found
[2024-02-02, 10:57:31 UTC] {python.py:201} INFO - Done. Returned value was: /Users/taragolis/.pyenv/versions/llama-runtime-py311/bin/python
[2024-02-02, 10:57:31 UTC] {taskinstance.py:1138} INFO - Marking task as SUCCESS. dag_id=issue_37136, task_id=task_runner, execution_date=20240202T105728, start_date=20240202T105731, end_date=20240202T105731
[2024-02-02, 10:57:31 UTC] {local_task_job_runner.py:234} INFO - Task exited with return code 0
[2024-02-02, 10:57:31 UTC] {taskinstance.py:3281} INFO - 0 downstream tasks scheduled from follow-on schedule check

[JCoder01]

Ahh, I see. Yeah, it works with task.external_python.
This still feels a bit error prone. As a user, if I'm used to doing from airflow.decorators import task and then I see my IDE pop up with external_python_task as an option I don't think it's unreasonable to think that's what I want to use, it's even included in the __all__. If it's meant to be private, cound it (and all other non public) be import external_python_task as _external_python_task to at least indicate it's private?

[rkarthik29]

I am having a similar issue with task.external_python. The logs fails when trying to check airflow version , this code here

airflow/providers/standard/src/airflow/providers/standard/operators/python.py

Line 1046 in 49181cf

from importlib.metadata import version

. I am using version 2.10.3