[fat-catTW]

Why

[core] execute_tasks_new_python_interpreter currently applies globally, so enabling fresh Python interpreter startup affects every task in the deployment.

Some Python tasks need that behavior, for example to pick up plugin changes immediately, but forcing all tasks onto the slower execution mode is unnecessary.

relates: #41093

Solution

Add execute_tasks_new_python_interpreter to PythonOperator as a task-level override.

When set to True or False, the task value is carried through Dag serialization, the ExecuteTask workload, and the task supervisor. When left as None, task execution keeps using the global [core] execute_tasks_new_python_interpreter configuration.

Was generative AI tooling used to co-author this PR?

-[X]Yes (please specify the tool below)

Generated-by: [Codex] following the guidelines

---

• Read the 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) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[ignacioparicio]

Not a review, just something I ran into that seems relevant.

I think this PR does more than add a per-task override. The config fallback in _should_use_exec_for_task also makes [core] execute_tasks_new_python_interpreter work again for CeleryExecutor and LocalExecutor, which has had no effect since 3.0.

The Celery code that reads it sits behind an if not AIRFLOW_V_3_0_PLUS: gate, and as far as I can tell the Edge worker is the only thing still honouring it on 3.x.

I opened #70901 in parallel to document that and get airflow config lint to flag it (happy to rework it if this lands first).

So if this merges it's a behaviour change for anyone still running Celery or Local with execute_tasks_new_python_interpreter set to True, which might deserve a newsfragment of its own.

[fat-catTW]

Thanks, that makes sense. Since this PR restores the global [core] execute_tasks_new_python_interpreter behavior for LocalExecutor and CeleryExecutor, I’ll add a newsfragment for that. Your #70901 can then probably focus on the config lint/documentation side depending on which PR lands first.

[uranusjr]

If I understand this correctly… this won’t work. ExecuteTask.make() reads ti.task.execute_tasks_new_python_interpreter, but ti.task is not guarenteed to be set at this point.

[fat-catTW]

Good catch, thanks for review.
I changed this so ExecuteTask.make() no longer reads from ti.task.
The scheduler now resolves the task from the serialized Dag during enqueue, reads execute_tasks_new_python_interpreter from that serialized task, and passes the value explicitly into ExecuteTask.make(). If the serialized Dag or task is unavailable, it falls back to None, so the existing global config fallback behavior is preserved.

[uranusjr]

I don’t think this is going to fly either, at least as-is. ExecuteTask.make() is used in providers, you need to add this to more places. Unfortunately, providers also introduce complex version compatibility issues, so you’ll need to figure out a way to ensure things work correctly between different version combinations. And for the combinations where things can’t work, you’ll need to be able to do error handling, preferrably at parse time. The two easiest ways this can go down are either to silently make the argument do nothing (or drop it with a warning), or make the scheduler crash. Neither is acceptable IMO.

[fat-catTW]

Thanks, that makes sense.
The PythonOperator-level override has broader provider/core compatibility implications than I initially accounted for.

While working on this, I noticed a related bug.
The existing global [core] execute_tasks_new_python_interpreter setting was not being honored by the task supervisor path. I narrowed this PR to only restore that fallback behavior, and removed the per-task override changes from this PR.

So this PR is now a smaller bugfix related to #41093, rather than closing it. The per-task override can be handled separately with a more complete design for provider executor compatibility.

[potiuk]

Thanks for narrowing this down, but the bug this PR now fixes has already landed on main in #72164, and the remaining hunk would undo how that PR scoped the option, so I'm closing this one as superseded rather than asking for a rebase.

Superseded by #72164

def _should_use_exec() -> bool:
    """Whether forked children should ``exec`` a fresh interpreter."""
    return sys.platform in _FORK_EXEC_PLATFORMS or conf.getboolean(   # ← already on main as _task_process_uses_exec()
        "core", "execute_tasks_new_python_interpreter", fallback=False
    )

#72164 ("Allow opting into fork+exec task processes via execute_tasks_new_python_interpreter", merged 2026-09-14) added _task_process_uses_exec() with exactly this or conf.getboolean(...) fallback, a TestTaskProcessUsesExec test, the 72164.significant.rst newsfragment and an updated config.yml description. This PR's diff no longer applies to main, and the green CI from 2026-08-08 predates that merge.

The hunk also changes the Dag processor and the triggerer

_should_use_exec() is not task-only; it is the shared platform gate:

airflow-core/src/airflow/dag_processing/processor.py:619:  use_exec = target is _parse_file_entrypoint and supervisor._should_use_exec()
airflow-core/src/airflow/jobs/triggerer_job_runner.py:551: use_exec=supervisor._should_use_exec(),

Reading the config here means execute_tasks_new_python_interpreter=True execs a fresh interpreter for every Dag-file parse and for the triggerer too, which the option has never described. #72164 avoided exactly this by introducing a separate _task_process_uses_exec() and leaving _should_use_exec() platform-only; its newsfragment states that the Dag processor, the triggerer and task callbacks keep bare fork. If this PR were rebased with this hunk kept, it would silently revert that scoping.

Smaller observations

• test_should_use_exec_honors_platform_and_config stubs supervisor.conf.getboolean with lambda *args, **kwargs: config_value, which returns the value for any section/key; conf_vars (already imported in this file) is the way to override config in tests.
• The title, body and labels still describe the per-task PythonOperator override that was removed on 2026-08-08, and 70280.bugfix.rst duplicates the merged 72164.significant.rst entry.
• If you want to pursue the original per-task override, that needs a new design that answers uranusjr's points (providers construct ExecuteTask themselves; unsupported version combinations must fail at parse time), best as a fresh PR.

---

This review was drafted by an AI-assisted tool and
confirmed by an Apache Airflow maintainer. After you've
addressed the points above and pushed an update, an Apache Airflow
maintainer — a real person — will take the next look
at the PR. The findings cite the project's review criteria;
if you think one of them is mis-applied, please reply on the
PR and a maintainer will weigh in.

More on how Apache Airflow handles maintainer review:
contributing-docs/05_pull_requests.rst.

[potiuk]

Closing as superseded by #72164 (see the review above). Thanks for the work on this; a fresh PR is the best route if you want to pursue the per-task override.