Executors¶
Executor abstraction for running extraction tasks.
LambdaExecutor(
function_name,
staging_bucket,
storage=None,
failure_mode=_STRICT_MODE,
endpoint_url=None,
max_concurrent_invokes=10,
invoke_timeout=900,
)
Execute extraction tasks remotely via AWS Lambda.
Each :class:ExtractionTask is serialized, staged to S3, and dispatched as a separate Lambda invocation. The Lambda handler is expected to deserialize the task, run the extraction, upload the results, and return a JSON payload with a manifest_uri key.
Because boto3 is an optional dependency, it is imported lazily inside :meth:__init__. Install it with pip install boto3 before using this executor.
Create a new Lambda executor.
| PARAMETER | DESCRIPTION |
|---|---|
function_name | AWS Lambda function name or ARN. TYPE: |
staging_bucket | S3 bucket used to stage serialized tasks. TYPE: |
storage | Optional storage backend used to load result manifests. When TYPE: |
failure_mode |
TYPE: |
endpoint_url | Optional boto3 endpoint URL (e.g. TYPE: |
max_concurrent_invokes | Maximum number of concurrent Lambda invocations. TYPE: |
invoke_timeout | Read timeout in seconds for the boto3 Lambda client. TYPE: |
| RAISES | DESCRIPTION |
|---|---|
ImportError | If |
Source code in components/aereo/executors/_lambda.py
def __init__(
self,
function_name: str,
staging_bucket: str,
storage: StorageBackend | None = None,
failure_mode: Literal["strict", "best_effort"] = _STRICT_MODE,
endpoint_url: str | None = None,
max_concurrent_invokes: int = 10,
invoke_timeout: int = 900,
) -> None:
"""Create a new Lambda executor.
Args:
function_name: AWS Lambda function name or ARN.
staging_bucket: S3 bucket used to stage serialized tasks.
storage: Optional storage backend used to load result manifests.
When ``None``, a backend is resolved from each manifest URI.
failure_mode: ``"strict"`` aborts on the first failed task;
``"best_effort"`` skips failed tasks and returns successful ones.
endpoint_url: Optional boto3 endpoint URL (e.g. ``http://localhost:4566``
for LocalStack emulation).
max_concurrent_invokes: Maximum number of concurrent Lambda invocations.
invoke_timeout: Read timeout in seconds for the boto3 Lambda client.
Raises:
ImportError: If ``boto3`` is not installed.
"""
self.function_name = function_name
self.staging_bucket = staging_bucket
self.storage = storage
self.failure_mode = failure_mode
self.endpoint_url = endpoint_url
self.max_concurrent_invokes = max_concurrent_invokes
self._invoke_timeout = invoke_timeout
self._serializer = _TaskSerializer()
self._staging = _CloudTaskStaging(
bucket=staging_bucket, endpoint_url=endpoint_url
)
try:
import boto3 # pyright: ignore[reportMissingImports]
from botocore.config import Config # pyright: ignore[reportMissingImports]
except ImportError as exc:
raise ImportError(
"boto3 is required for LambdaExecutor. "
"Install it with: pip install boto3"
) from exc
self._lambda_client = boto3.client(
"lambda",
endpoint_url=endpoint_url,
config=Config(
read_timeout=invoke_timeout,
retries={"max_attempts": 3, "mode": "adaptive"},
),
)
Bases: RuntimeError
Raised when a Lambda invocation fails with a retryable error.
Bases: Protocol
Protocol for pluggable task executors.
An executor turns a sequence of :class:ExtractionTask objects into a single validated artifact GeoDataFrame.
LocalExecutor(
workers=1,
failure_mode=_STRICT_MODE,
cache=None,
use_threads=False,
)
Execute extraction tasks locally.
Wraps :func:aereo.execution.run_task with optional caching, failure handling, and local parallelism through joblib.
By default joblib's loky backend is used for process-based parallelism. loky starts clean interpreter processes, which avoids the fork-after-read deadlock that happens when worker processes inherit netCDF/HDF5 state from the parent.
Create a new LocalExecutor.
| PARAMETER | DESCRIPTION |
|---|---|
workers | Maximum number of parallel workers. TYPE: |
failure_mode |
TYPE: |
cache | Optional per-task artifact catalog cache. TYPE: |
use_threads | When TYPE: |
Source code in components/aereo/executors/core.py
def __init__(
self,
workers: int | None = 1,
failure_mode: Literal["strict", "best_effort"] = _STRICT_MODE,
cache: TaskResultCache | None = None,
use_threads: bool = False,
) -> None:
"""Create a new LocalExecutor.
Args:
workers: Maximum number of parallel workers. ``None`` or ``1`` runs
tasks sequentially in the current process. ``>1`` dispatches
tasks through a joblib pool. If -1 is passed, the number of
workers is set to the number of CPUs in the system.
failure_mode: ``"strict"`` aborts on the first failed task;
``"best_effort"`` skips failed tasks and returns successful ones.
cache: Optional per-task artifact catalog cache.
use_threads: When ``True`` and *workers* > 1, use joblib's
``threading`` backend instead of ``loky``.
"""
self.workers = workers
if self.workers == -1:
import multiprocessing
self.workers = multiprocessing.cpu_count()
self.failure_mode = failure_mode
self.cache = cache
self.use_threads = use_threads
shutdown(_wait=True)
No-op for API compatibility.
joblib's loky backend already reuses and cleans up its worker pool automatically.
Source code in components/aereo/executors/core.py
def shutdown(self, _wait: bool = True) -> None:
"""No-op for API compatibility.
``joblib``'s ``loky`` backend already reuses and cleans up its worker
pool automatically.
"""
return None