`yggdrasil.concurrent.threading`¶

threading ¶

yggdrasil.concurrent.threading¶

Thread-specific async job primitives.

Canonical exports¶

:class:AsyncJob — abstract awaitable handle (base.py)
:class:ThreadJob — daemon-thread implementation (thread_job.py)

Backward-compatible re-exports¶

Job, JobResult, and JobPoolExecutor are also exported here so that existing from yggdrasil.concurrent.threading import … statements continue to work without modification.

AsyncJob ¶

Bases: ABC, Generic[T]

Abstract handle for an async operation that can be awaited.

Implementors must provide :meth:wait, :attr:is_done, and :meth:result.

is_done `abstractmethod` `property` ¶

is_done: bool

True when the job has finished (success or failure).

wait `abstractmethod` ¶

wait(wait: WaitingConfigArg = None, raise_error: bool = True) -> Optional[T]

Block until the job completes and return its result.

Parameters:

Name	Type	Description	Default
`wait`	`WaitingConfigArg`	`None` / `True` → block indefinitely. `False` → non-blocking poll; returns `None` if not yet finished. `WaitingConfig` or a numeric number of seconds → timed wait.	`None`
`raise_error`	`bool`	Re-raise the job's exception when `True`; return `None` when `False`.	`True`

Returns:

Type	Description
`Optional[T]`	The job's return value, or `None` on failure / timeout / non-blocking
`Optional[T]`	miss (when `raise_error=False`).

Raises:

Type	Description
`TimeoutError`	If a timed wait elapses before the job finishes.
`BaseException`	Any exception raised by the job (when `raise_error=True`).

result `abstractmethod` ¶

result() -> Optional[JobResult[T]]

Return the :class:~yggdrasil.concurrent.job_result.JobResult if done, else None.

ThreadJob `dataclass` ¶

ThreadJob(job: Job[T])

Bases: AsyncJob[T]

A :class:~yggdrasil.concurrent.job.Job running in a daemon :class:~threading.Thread, started immediately on construction.

Use :meth:wait to block and retrieve the result::

handle = Job.make(fetch_data, url).fire_and_forget()

# … do other work …

data = handle.wait()           # block until thread finishes
data = handle.wait(wait=5.0)   # block at most 5 seconds
data = handle.wait(wait=False) # non-blocking poll

is_done `property` ¶

is_done: bool

True when the thread has finished (success or failure).

result ¶

result() -> Optional[JobResult[T]]

Return the :class:~yggdrasil.concurrent.job_result.JobResult if done, else None.

wait ¶

wait(wait: WaitingConfigArg = None, raise_error: bool = True) -> Optional[T]

Block until the thread finishes.

Parameters:

Name	Type	Description	Default
`wait`	`WaitingConfigArg`	`None` / `True` → block indefinitely. `False` → non-blocking poll. `WaitingConfig` or numeric seconds → timed wait.	`None`
`raise_error`	`bool`	Re-raise the thread's exception when `True`.	`True`

Returns:

Type	Description
`Optional[T]`	The job return value, or `None` on failure / timeout / non-blocking
`Optional[T]`	miss (when `raise_error=False`).

Raises:

Type	Description
`TimeoutError`	When a timed wait elapses before the thread finishes.
`BaseException`	The exception from the job (when `raise_error=True`).

Job ¶

Job(
    func: Callable[..., T],
    args: Tuple[Any, ...] = _EMPTY_ARGS,
    kwargs: Dict[str, Any] = _EMPTY_KWARGS,
)

Bases: Generic[T]

Minimal callable bundle: a function plus its bound args / kwargs.

Deliberately not a :func:dataclass. Job is on the hot path for every fan-out (JobPoolExecutor.as_completed), every fire-and-forget thread spawn, and every retry layer — the saved dataclass init / equality / repr overhead matters per job. A plain slotted class with a hand-written __init__ is roughly half the per-construction cost.

Build via :meth:make (collects *args / **kwargs like a normal call site) or pass them explicitly to the constructor.

Usage::

job = Job.make(my_func, arg1, arg2, key=val)
result = job.run()                # synchronous
handle = job.fire_and_forget()    # background thread, returns ThreadJob
handle.wait()                     # block until done

make `classmethod` ¶

make(func: Callable[..., T], *args: Any, **kwargs: Any) -> 'Job[T]'

Build a :class:Job from a callable and its arguments.

run ¶

run() -> T

Execute the job synchronously in the calling thread.

thread ¶

thread() -> 'ThreadJob[T]'

Start this job in a daemon thread and return a :class:~yggdrasil.concurrent.threading.ThreadJob handle.

The thread is started immediately. Call :meth:~yggdrasil.concurrent.threading.ThreadJob.wait to block until completion and retrieve the result.

fire_and_forget ¶

fire_and_forget() -> 'ThreadJob[T]'

Alias for :meth:thread — kept for the more readable call site.

JobResult `dataclass` ¶

JobResult(result: Optional[T], exception: Optional[BaseException])

Bases: Generic[T]

Immutable outcome of a :class:~yggdrasil.concurrent.job.Job execution.

Either holds a result (success) or an exception (failure). Both fields are always present; exactly one is None in the normal case.

ok `property` ¶

ok: bool

True when the job succeeded (no exception captured).

get ¶

get() -> T

Return the result or re-raise the captured exception.

Raises:

Type	Description
`BaseException`	The exception that was captured during job execution.

JobPoolExecutor ¶

JobPoolExecutor(
    max_workers: int | None = None,
    max_in_flight: int | None = None,
    job_name_prefix: str = "",
)

Bases: ThreadPoolExecutor

ThreadPoolExecutor helper for large / infinite :class:Job streams.

Keeps at most max_in_flight futures submitted at any time so the caller is never overwhelmed when the job source is unbounded.

Parameters:

Name	Type	Description	Default
`max_workers`	`int \| None`	Thread-pool size (`None` → `os.cpu_count()`).	`None`
`max_in_flight`	`int \| None`	Max in-flight futures (`None` → `max_workers * 2`).	`None`
`job_name_prefix`	`str`	Thread-name prefix forwarded to `ThreadPoolExecutor`.	`''`

Yields (via :meth:as_completed): :class:JobResult in completion order (ordered=False) or submission order (ordered=True).

submit_job ¶

submit_job(job: Job) -> Future

Submit a single :class:Job and return its :class:Future.

as_completed ¶

as_completed(
    jobs: Iterable[Job],
    *,
    ordered: bool = False,
    max_in_flight: int | None = None,
    cancel_on_exit: bool = False,
    shutdown_on_exit: bool = False,
    shutdown_wait: bool = False,
    raise_error: bool = True
) -> Iterator[JobResult]

Consume a (possibly infinite) :class:Job iterable and yield :class:JobResult objects.

Parameters:

Name	Type	Description	Default
`jobs`	`Iterable[Job]`	Source of :class:`Job` objects (may be infinite).	required
`ordered`	`bool`	`False` (default) → completion order. `True` → strict submission order.	`False`
`max_in_flight`	`int \| None`	Override the instance-level window for this call.	`None`
`cancel_on_exit`	`bool`	Cancel pending futures on generator close / error.	`False`
`shutdown_on_exit`	`bool`	Shut down the pool on generator close / error.	`False`
`shutdown_wait`	`bool`	Wait for running threads when shutting down.	`False`
`raise_error`	`bool`	Re-raise job exceptions at the yield site.	`True`

Yields:

Type	Description
`JobResult`	class:`JobResult` — one per submitted job.

yggdrasil.concurrent.threading¶

threading ¶