ferruzzi commented on code in PR #62645:
URL: https://github.com/apache/airflow/pull/62645#discussion_r2996394549
##########
task-sdk/src/airflow/sdk/execution_time/supervisor.py:
##########
@@ -2120,3 +2120,99 @@ def supervise(
if close_client and client:
with suppress(Exception):
client.close()
+
+
+def supervise_workload(
+ workload: ExecutorWorkload,
+ *,
+ server: str | None = None,
+ dry_run: bool = False,
+ client: Client | None = None,
+ subprocess_logs_to_stdout: bool = False,
+ proctitle: str | None = None,
+) -> int:
+ """
+ Run any workload type to completion in a supervised subprocess.
+
+ Dispatch to the appropriate supervisor based on workload type.
Workload-specific
+ attributes (log_path, sentry_integration, bundle_info, etc.) are read from
the
+ workload object itself.
+
+ :param workload: The ``ExecutorWorkload`` to execute.
+ :param server: Base URL of the API server (used by task workloads).
+ :param dry_run: If True, execute without actual task execution (simulate
run).
+ :param client: Optional preconfigured client for communication with the
server.
+ :param subprocess_logs_to_stdout: Should task logs also be sent to stdout
via the main logger.
+ :param proctitle: Process title to set for this workload. If not provided,
defaults to
+ ``"airflow supervisor: <workload.display_name>"``. Executors may pass
a custom title
+ that includes executor-specific context (e.g. team name).
+ :return: Exit code of the process.
+ """
+ # Imports deferred to avoid an SDK/core dependency at module load time.
+ from airflow.executors.workloads.callback import ExecuteCallback
+ from airflow.executors.workloads.task import ExecuteTask
Review Comment:
Moving to a shared library looks like the "most correct" answer and would
mostly be a clean break. I think we can safely move everything from the
`executors/workloads` path into a new location at `shared/workloads` *except*
the `types.py`. As Amogh hints, moving the types file will get a bit messy
since the type hints are scattered all over the codebase and will end up
causing a fairly large diff. I think most of that work would just be updating
import paths, but there are a lot of them. A quick grep (don't quote me on
this part, I didn't do a thorough search) looks like ~60 imports over ~35 files.
That's not a no, but as you said, that feels like it is worth its own PR.
If we're aiming for 3.2.1 at this point, that gives me about two weeks for that.
As for what we can do right now in this PR... would it work if we move
`supervise_workload` into core (I'd propose a new location at
`airflow.executors.workload_dispatcher`?) and leave the `supervise_task` and
`supervise_callback` where they are? I think that meets the barrier design...
`supervise_workload` depends on core workload types but dispatches to
`supervise_task` and `supervise_callback`, which both live in the SDK and (I'm
pretty sure) they both have a core import for type-hinting, but at runtime they
only depend on SDK-side objects.
--
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]
