diff --git a/pyproject.toml b/pyproject.toml index 8ace60a..5abe3de 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -94,6 +94,7 @@ ignore = [ "yt_framework/yt/clients/client_dev.py" = ["ARG002", "PLR0913"] # Complexity: remaining high-arity API surfaces are isolated to explicit modules. "yt_framework/operations/_internal/dependency_strategy.py" = ["PLR0913"] +"yt_framework/operations/command_ops/map.py" = ["PLR0913"] "yt_framework/operations/command_ops/map_reduce.py" = ["PLR0913"] "yt_framework/operations/command_ops/map_reduce_support.py" = ["D103"] "yt_framework/operations/s3.py" = ["PLR0913"] diff --git a/tests/test_map.py b/tests/test_map.py index 20c7832..7a502a6 100644 --- a/tests/test_map.py +++ b/tests/test_map.py @@ -156,3 +156,12 @@ def test_run_map_raises_when_mapper_missing_and_builder_command_missing( ctx = _map_stage_context(tmp_path) with pytest.raises(ValueError, match="Command not provided"): run_map(ctx, _minimal_map_config()) + + +def test_run_map_sync_false_returns_operation_without_waiting( + tmp_path: Path, +) -> None: + ctx = _map_stage_context(tmp_path) + result = run_map(ctx, _minimal_map_config(), sync=False) + assert result is ctx.deps.yt_client.run_map_submit.return_value + ctx.deps.yt_client.wait_for_operation.assert_not_called() diff --git a/yt_framework/operations/command_ops/map.py b/yt_framework/operations/command_ops/map.py index 77824e9..2540694 100644 --- a/yt_framework/operations/command_ops/map.py +++ b/yt_framework/operations/command_ops/map.py @@ -32,6 +32,7 @@ import logging from pathlib import Path + from yt.wrapper import Operation from yt.wrapper.schema import TableSchema from yt_framework.contracts import StageContext @@ -173,8 +174,10 @@ def run_map( output_schema: TableSchema | None = None, mapper: object | None = None, job: object | None = None, -) -> bool: - """Run YT map operation and wait for completion. + *, + sync: bool = True, +) -> bool | Operation: + """Run YT map operation and optionally wait for completion. All job parameters (pool, memory, CPU, Docker image, etc.) are automatically extracted from operation_config. Operation config should be passed from @@ -188,9 +191,12 @@ def run_map( mapper: Optional mapper leg (legacy name). When omitted, framework uses command wrapper. Can be a TypedJob instance or command string. job: Preferred mapper leg alias. Can be a TypedJob instance or command string. + sync: If True (default), wait for operation completion and return bool. + If False, submit the operation and return the Operation object immediately. Returns: - True if successful, False otherwise + If ``sync=True``: True if successful, False otherwise. + If ``sync=False``: the submitted ``yt.wrapper.Operation`` object. """ logger = context.logger @@ -274,6 +280,9 @@ def run_map( ), ) + if not sync: + return operation + # Wait for completion success = context.deps.yt_client.wait_for_operation(operation)