airflow.providers.modal.utils.python_sandbox¶
How @task.modal runs a function in a Modal sandbox and gets its return value back.
The sandbox’s command is a short bootstrap program (BOOTSTRAP) run with python -c.
It reads one JSON payload from stdin holding the function’s source and its pickled arguments,
defines the function and calls it. Everything the function prints, including what its
subprocesses write and any traceback, goes to the sandbox’s stderr, in order; the task log
follows that stream. Stdout carries only the return value, pickled, base64-encoded and
prefixed with its length (<length>:<base64>), written once the function has returned, so a
function that raises leaves stdout empty and exits nonzero.
The bootstrap needs only the Python standard library, so the image needs neither Airflow nor the Modal client. The shared virtualenv script template is deliberately not used: it reconnects to an Airflow supervisor when it finds Airflow installed, and a sandbox has none.
Both streams outlive the sandbox and can be read through a new handle from
Sandbox.from_id, which is what lets the deferrable path follow the output on the triggerer
and collect the return value back on a worker. Both the decorator, on a worker, and the
trigger, on the triggerer, use this module.
Attributes¶
Functions¶
|
Return the stdin payload for |
|
Write the payload to the sandbox's stdin in pieces that fit the SDK's buffer, then close it. |
|
Wait for the sandbox to exit, leaving its exit code on |
|
Async counterpart of |
|
Return the function's return value from the sandbox's stdout. |
Module Contents¶
- airflow.providers.modal.utils.python_sandbox.BOOTSTRAP = Multiline-String[source]¶
Show Value
""" import __future__, base64, json, os, pickle, sys, tempfile, types # Keep the real stdout for the result, and point fd 1 at stderr so everything else the # function prints, from Python or from a subprocess, lands on stderr. result = os.fdopen(os.dup(1), "wb") os.dup2(2, 1) payload = json.loads(sys.stdin.read()) # Compiling from a file, rather than a string, lets tracebacks show the function's source lines. path = os.path.join(tempfile.mkdtemp(prefix="airflow-modal-task-"), payload["function_name"] + ".py") with open(path, "w") as f: f.write(payload["source"]) # A real, registered module: dataclasses and other tools look a class's module up in sys.modules. module = types.ModuleType("__airflow_modal_task__") sys.modules[module.__name__] = module # Postponed annotations, as in a Dag file with `from __future__ import annotations`: the # function's annotations may name types that only the Dag file imported. code = compile( payload["source"], path, "exec", flags=__future__.annotations.compiler_flag, dont_inherit=True ) exec(code, module.__dict__) arguments = pickle.loads(base64.b64decode(payload["arguments"])) value = getattr(module, payload["function_name"])(*arguments["args"], **arguments["kwargs"]) # Length-prefixed, so a value Modal kept only the end of is detected instead of misread. encoded = base64.b64encode(pickle.dumps(value)) result.write(b"%d:" % len(encoded) + encoded) result.close() """
- airflow.providers.modal.utils.python_sandbox.build_payload(*, source, function_name, args, kwargs)[source]¶
Return the stdin payload for
BOOTSTRAP.- Parameters:
source (str) – Source of the function, without its decorators.
function_name (str) – Name the source defines the function under.
args (collections.abc.Iterable[Any]) – Positional arguments to call it with.
kwargs (dict[str, Any]) – Keyword arguments to call it with.
- airflow.providers.modal.utils.python_sandbox.send_payload(sandbox, payload)[source]¶
Write the payload to the sandbox’s stdin in pieces that fit the SDK’s buffer, then close it.
- airflow.providers.modal.utils.python_sandbox.wait_for_exit(sandbox)[source]¶
Wait for the sandbox to exit, leaving its exit code on
sandbox.returncode.Sandbox.waitraisesSandboxTimeoutErrorfor a sandbox that hit its timeout even withraise_on_termination=False. The exit code (124) is recorded before it raises, so the caller reports a timeout the same way as any other exit code.
- async airflow.providers.modal.utils.python_sandbox.await_exit(sandbox)[source]¶
Async counterpart of
wait_for_exit(), for the triggerer’s event loop.
- airflow.providers.modal.utils.python_sandbox.decode_result(stdout)[source]¶
Return the function’s return value from the sandbox’s stdout.
- Raises:
ValueError – When stdout is empty, which happens when the function did not return, or does not hold a whole value, which happens when Modal kept only the end of a large one.