pynenc.core_tasks

Module Contents

Classes

CoreTaskFunction	Wrapper for core task functions that will be bound to an app at runtime.
CoreTaskDefinition
CoreTaskRegistry	Registry for core tasks definitions that should run in any Pynenc instance.

Functions

get_app_and_runner_ctx	Get the app and runner context from the current execution context
recover_pending_invocations	Recovers PENDING invocations that exceeded the allowed pending time
recover_running_invocations	Recovers PENDING invocations that exceeded the allowed pending time

Data

core_tasks_registry

API

class pynenc.core_tasks.CoreTaskFunction(func: collections.abc.Callable)

Wrapper for core task functions that will be bound to an app at runtime.

This wrapper allows the function to be identified as a core task during deserialization, enabling proper function extraction in Task._from_json.

Initialization

__call__(*args: Any, **kwargs: Any) → Any

class pynenc.core_tasks.CoreTaskDefinition

Bases: typing.NamedTuple

func: collections.abc.Callable

None

options: dict[str, Any]

None

config_cron: str | None

None

class pynenc.core_tasks.CoreTaskRegistry

Registry for core tasks definitions that should run in any Pynenc instance.

Initialization

task(config_cron: str | None = None, **options: Any) → collections.abc.Callable[[collections.abc.Callable], pynenc.core_tasks.CoreTaskFunction]

Deferred decorator that stores the function and options

pynenc.core_tasks.get_app_and_runner_ctx() → tuple[pynenc.app.Pynenc, pynenc.runner.runner_context.RunnerContext]

Get the app and runner context from the current execution context

pynenc.core_tasks.core_tasks_registry

‘CoreTaskRegistry(…)’

pynenc.core_tasks.recover_pending_invocations() → None

Recovers PENDING invocations that exceeded the allowed pending time

pynenc.core_tasks.recover_running_invocations() → None

Recovers PENDING invocations that exceeded the allowed pending time