Allow PythonOperator to override task interpreter mode

[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.