# Copyright 2025 Softwell S.r.l.
# Licensed under the Apache License, Version 2.0
"""
Base executor interface and exceptions.
Purpose
=======
Defines the abstract base class for all executor implementations and
common exceptions. BaseExecutor owns the shared machinery — bypass mode,
metrics accounting around submit, pool shutdown, decorator interface —
while subclasses provide the pooled execution core.
Definition::
class BaseExecutor(ABC):
name: str
pool: Any # None in bypass mode
pool_mode: str # "process", "thread", ... (metrics label)
async def submit(self, func: Callable, *args, **kwargs) -> Any
'''Template: bypass check + metrics around _submit_pooled().'''
@abstractmethod
async def _submit_pooled(self, func: Callable, *args, **kwargs) -> Any
'''Pooled execution core (backpressure + dispatch).'''
def shutdown(self, wait: bool = True) -> None
'''Shutdown self.pool if present.'''
@property
def metrics(self) -> dict[str, Any]
'''base_metrics(); subclasses may extend with extra gauges.'''
def __call__(self, func: Callable) -> Callable
'''Decorator interface - wraps func to use submit().'''
Exceptions::
ExecutorError
Base exception for executor operations.
ExecutorOverloadError(ExecutorError)
Raised when executor has too many pending tasks (backpressure).
Design Notes
============
- BaseExecutor provides __call__ for decorator pattern (uses submit internally)
- Subclasses implement _submit_pooled() and set pool/pool_mode in __init__
- Metrics dict includes: name, mode, pending, submitted, completed, failed,
avg_duration_ms; subclasses may add gauges on top of base_metrics()
"""
from __future__ import annotations
import time
from abc import ABC, abstractmethod
from functools import wraps
from typing import Any, Callable, TypeVar
__all__ = ["BaseExecutor", "ExecutorError", "ExecutorOverloadError"]
F = TypeVar("F", bound=Callable[..., Any])
[docs]
class ExecutorError(Exception):
"""Base exception for executor operations."""
pass
[docs]
class ExecutorOverloadError(ExecutorError):
"""Raised when executor has too many pending tasks."""
pass
[docs]
class BaseExecutor(ABC):
"""
Abstract base class for all executor implementations.
Owns bypass handling, metrics accounting and pool shutdown; implements
the decorator pattern via __call__ which uses submit().
Subclasses must:
- call BaseExecutor.__init__ and set ``pool`` (None means bypass mode)
- set the ``pool_mode`` class attribute (metrics label)
- implement _submit_pooled(): backpressure + pooled dispatch
Attributes:
name: Identifier for this executor instance.
pool: The underlying pool, or None in bypass mode.
Example:
>>> class MyExecutor(BaseExecutor):
... pool_mode = "inline"
... async def _submit_pooled(self, func, *args, **kwargs):
... return func(*args, **kwargs)
>>>
>>> executor = MyExecutor("test")
>>>
>>> @executor
... def my_func(x):
... return x * 2
>>>
>>> result = await my_func(5) # returns 10
"""
__slots__ = ("name", "pool", "_metrics")
pool_mode: str = "pool"
[docs]
def __init__(self, name: str = "default") -> None:
"""Initialize shared executor state.
Args:
name: Identifier for metrics and logging.
"""
self.name = name
self.pool: Any = None
self._metrics: dict[str, Any] = {
"submitted": 0,
"completed": 0,
"failed": 0,
"total_duration_ms": 0.0,
}
[docs]
async def submit(self, func: Callable[..., Any], *args: Any, **kwargs: Any) -> Any:
"""
Submit a function for execution.
Template method: in bypass mode (no pool) runs func synchronously;
otherwise wraps _submit_pooled() with metrics accounting.
Args:
func: The function to execute.
*args: Positional arguments for func.
**kwargs: Keyword arguments for func.
Returns:
The result of func(*args, **kwargs).
Raises:
ExecutorError: If execution fails.
ExecutorOverloadError: If too many tasks are pending.
"""
if self.pool is None:
return func(*args, **kwargs)
self._metrics["submitted"] += 1
start = time.monotonic()
try:
result = await self._submit_pooled(func, *args, **kwargs)
self._metrics["completed"] += 1
return result
except Exception:
self._metrics["failed"] += 1
raise
finally:
self._metrics["total_duration_ms"] += (time.monotonic() - start) * 1000
@abstractmethod
async def _submit_pooled(self, func: Callable[..., Any], *args: Any, **kwargs: Any) -> Any:
"""Pooled execution core: backpressure + dispatch on the pool.
Called by submit() only when a pool is present; metrics are handled
by the caller.
"""
...
[docs]
def shutdown(self, wait: bool = True) -> None:
"""
Shutdown the executor and release resources.
Args:
wait: If True, wait for pending tasks to complete.
"""
if self.pool is not None:
self.pool.shutdown(wait=wait)
[docs]
def base_metrics(self) -> dict[str, Any]:
"""Common metrics core shared by all executors.
Returns:
Dict with name, mode, pending, submitted, completed, failed,
avg_duration_ms.
"""
completed = self._metrics["completed"]
return {
"name": self.name,
"mode": "bypass" if self.pool is None else self.pool_mode,
"pending": (self._metrics["submitted"] - completed - self._metrics["failed"]),
"submitted": self._metrics["submitted"],
"completed": completed,
"failed": self._metrics["failed"],
"avg_duration_ms": (
self._metrics["total_duration_ms"] / completed if completed > 0 else 0.0
),
}
@property
def metrics(self) -> dict[str, Any]:
"""
Return executor metrics (base_metrics; subclasses may extend).
Returns:
Dict containing at minimum:
- name: Executor name
- mode: pool_mode, or "bypass" when no pool
- pending: Number of pending tasks
- submitted: Total submitted tasks
- completed: Total completed tasks
- failed: Total failed tasks
- avg_duration_ms: Mean completed-task duration
"""
return self.base_metrics()
def __call__(self, func: F) -> F:
"""
Decorate a function to run in this executor.
Args:
func: The function to wrap.
Returns:
Async wrapper that runs func via submit().
"""
@wraps(func)
async def wrapper(*args: Any, **kwargs: Any) -> Any:
return await self.submit(func, *args, **kwargs)
return wrapper # type: ignore[return-value]
def __repr__(self) -> str:
"""Return string representation."""
return f"{self.__class__.__name__}(name={self.name!r})"
if __name__ == "__main__":
# Quick validation that ABC works
print("BaseExecutor is abstract - cannot instantiate directly")
try:
BaseExecutor() # type: ignore[abstract]
except TypeError as e:
print(f" Expected error: {e}")