FrankYang0529 opened a new pull request, #73081:
URL: https://github.com/apache/airflow/pull/73081
## Why
- `PythonVirtualenvOperator` uses a virtualenv built from `requirements`,
and `ExternalPythonOperator` uses the interpreter given in `python`. An `async
def` callable never reaches `execute_callable()`, so `requirements`, `python`,
and `venv_cache_path` are ignored.
- Either the callable runs with the worker's own packages and the task still
succeeds, or it fails with an `ImportError` that looks like a broken virtualenv.
## How
- `_BasePythonVirtualenvOperator.__init__` raises `ValueError` for async
callables. The Dag now fails at parse time.
## Verification
- Unit test: `uv run --frozen --project providers/standard pytest
providers/standard/tests/unit/standard/operators/test_python.py`
- Integration test:
1. Setup
```sh
export DEMO_DIR="${TMPDIR:-/tmp}/airflow-async-venv-demo"
export AIRFLOW_HOME="$DEMO_DIR/airflow-home"
export AIRFLOW__CORE__DAGS_FOLDER="$DEMO_DIR/dags"
export AIRFLOW__CORE__LOAD_EXAMPLES=False
mkdir -p "$AIRFLOW__CORE__DAGS_FOLDER"
uv run --frozen --no-dev --project providers/standard --with "pandas>=3,<4"
airflow db migrate
uv venv --allow-existing "$DEMO_DIR/pandas2-venv"
uv pip install --python "$DEMO_DIR/pandas2-venv/bin/python" "pandas>=2.2,<3"
```
2. Create Dags
```sh
cat > "$AIRFLOW__CORE__DAGS_FOLDER/virtualenv_example.py" <<'EOF'
from __future__ import annotations
from airflow.sdk import dag, task
@dag(schedule=None)
def virtualenv_example():
@task.virtualenv(requirements=["pandas>=2.2,<3"],
system_site_packages=False)
def virtualenv_sync():
import sys
import pandas as pd
return {"python": sys.executable, "pandas": pd.__version__}
@task.virtualenv(requirements=["pandas>=2.2,<3"],
system_site_packages=False)
async def virtualenv_async():
import sys
import pandas as pd
return {"python": sys.executable, "pandas": pd.__version__}
virtualenv_sync()
virtualenv_async()
example_dag = virtualenv_example()
if __name__ == "__main__":
dag_run = example_dag.test()
for ti in sorted(dag_run.get_task_instances(), key=lambda ti:
ti.task_id):
print(f"EXAMPLE STATE {ti.task_id} {ti.state}")
EOF
cat > "$AIRFLOW__CORE__DAGS_FOLDER/external_python_example.py" <<'EOF'
from __future__ import annotations
import os
from airflow.sdk import dag, task
PANDAS2_PYTHON = os.path.join(os.environ["DEMO_DIR"], "pandas2-venv", "bin",
"python")
@dag(schedule=None)
def external_python_example():
@task.external_python(python=PANDAS2_PYTHON)
def external_python_sync():
import sys
import pandas as pd
return {"python": sys.executable, "pandas": pd.__version__}
@task.external_python(python=PANDAS2_PYTHON)
async def external_python_async():
import sys
import pandas as pd
return {"python": sys.executable, "pandas": pd.__version__}
external_python_sync()
external_python_async()
example_dag = external_python_example()
if __name__ == "__main__":
dag_run = example_dag.test()
for ti in sorted(dag_run.get_task_instances(), key=lambda ti:
ti.task_id):
print(f"EXAMPLE STATE {ti.task_id} {ti.state}")
EOF
```
3. Run dags to check result
On main branch:
```sh
uv run --frozen --no-dev --project providers/standard --with "pandas>=3,<4"
python "$AIRFLOW__CORE__DAGS_FOLDER/virtualenv_example.py" 2>&1 | grep -oE
"running task <TaskInstance: [^ ]+|Returned value was: \{[^}]*\}|EXAMPLE STATE
.*"
uv run --frozen --no-dev --project providers/standard --with "pandas>=3,<4"
python "$AIRFLOW__CORE__DAGS_FOLDER/external_python_example.py" 2>&1 | grep -oE
"running task <TaskInstance: [^ ]+|Returned value was: \{[^}]*\}|EXAMPLE STATE
.*"
```
All four tasks succeed. The sync tasks report pandas 2 from the virtualenv
or the external environment. The async tasks report pandas 3 from the `uv run`
environment, which shows they ignored `requirements` and `python`:
```text
running task <TaskInstance: virtualenv_example.virtualenv_sync
Returned value was: {'python': '.../venvymv4y504/bin/python', 'pandas':
'2.3.3'}
running task <TaskInstance: virtualenv_example.virtualenv_async
Returned value was: {'python': '.../builds-v0/.tmpDC6uaL/bin/python',
'pandas': '3.0.5'}
EXAMPLE STATE virtualenv_sync success
EXAMPLE STATE virtualenv_async success
running task <TaskInstance: external_python_example.external_python_sync
Returned value was: {'python':
'.../airflow-async-venv-demo/pandas2-venv/bin/python', 'pandas': '2.3.3'}
running task <TaskInstance: external_python_example.external_python_async
Returned value was: {'python': '.../builds-v0/.tmp90SjJP/bin/python',
'pandas': '3.0.5'}
EXAMPLE STATE external_python_sync success
EXAMPLE STATE external_python_async success
```
On this branch, both Dags now fail at parse time.
```sh
uv run --frozen --no-dev --project providers/standard --with "pandas>=3,<4"
python "$AIRFLOW__CORE__DAGS_FOLDER/virtualenv_example.py" 2>&1 | grep
ValueError
uv run --frozen --no-dev --project providers/standard --with "pandas>=3,<4"
python "$AIRFLOW__CORE__DAGS_FOLDER/external_python_example.py" 2>&1 | grep
ValueError
```
```text
raise ValueError(
ValueError: _PythonVirtualenvDecoratedOperator does not support async
functions as python_callable. Call asyncio.run() inside a regular function
instead.
raise ValueError(
ValueError: _PythonExternalDecoratedOperator does not support async
functions as python_callable. Call asyncio.run() inside a regular function
instead.
```
<!-- SPDX-License-Identifier: Apache-2.0
https://www.apache.org/licenses/LICENSE-2.0 -->
<!--
Thank you for contributing!
Please provide above a brief description of the changes made in this pull
request.
Write a good git commit message following this guide:
https://chris.beams.io/posts/git-commit/
Please make sure that your code changes are covered with tests.
And in case of new features or big changes remember to adjust the
documentation.
For user-facing UI changes, please attach before/after screenshots (or a
short
screen recording) so reviewers can assess the visual impact.
Feel free to ping (in general) for the review if you do not see reaction for
a few days
(72 Hours is the minimum reaction time you can expect from volunteers) - we
sometimes miss notifications.
In case of an existing issue, reference it using one of the following:
* closes: #ISSUE
* related: #ISSUE
-->
---
##### Was generative AI tooling used to co-author this PR?
<!--
If generative AI tooling has been used in the process of authoring this PR,
please
change below checkbox to `[X]` followed by the name of the tool, uncomment
the "Generated-by".
-->
- [X] Yes - Claude Code
<!--
Generated-by: [Tool Name] following [the
guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
-->
---
* Read the **[Pull Request
Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)**
for more information. Note: commit author/co-author name and email in commits
become permanently public when merged.
* For fundamental code changes, an Airflow Improvement Proposal
([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals))
is needed.
* When adding dependency, check compliance with the [ASF 3rd Party License
Policy](https://www.apache.org/legal/resolved.html#category-x).
* For significant user-facing changes create newsfragment:
`{pr_number}.significant.rst`, in
[airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments).
You can add this file in a follow-up commit after the PR is created so you
know the PR number.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
