pynenc.call

Module Contents

Classes

Call	Base class for task calls with common functionality.
PreSerializedCall	Represents a call optimized for parallel routing with pre-serialized arguments. This call type is used for batch processing tasks with disabled concurrency control, where some arguments are pre-serialized (e.g., large shared data).

API

class pynenc.call.Call

Bases: typing.Generic[pynenc.types.Params, pynenc.types.Result]

Base class for task calls with common functionality.

Parameters:

task (Task[Params, Result]) – The task associated with the call.

task: Task[Params, Result]

None

_arguments: pynenc.arguments.Arguments

‘field(…)’

_serialized_arguments: dict[str, str] | None

None

property app: pynenc.app.Pynenc

Gets the Pynenc application instance associated with the task.

Returns:

The Pynenc application instance.

property arguments: pynenc.arguments.Arguments

Get the arguments for this call. This property allows subclasses to override argument handling.

Returns:

Arguments object containing call arguments

property call_id: str

Generates a unique identifier for the call based on the task ID and the arguments.

Returns:

A string representing the unique identifier of the call.

serialized_arguments() → dict[str, str]

Serializes the call arguments into strings.

Returns:

A dictionary of serialized argument strings.

serialized_args_for_concurrency_check() → dict[str, str] | None

Determines the call arguments required for the task concurrency check.

Returns:

A dictionary of serialized argument strings required for concurrency control, or None if concurrency control is disabled.

deserialize_arguments(serialized_arguments: dict[str, str]) → pynenc.arguments.Arguments

Deserializes the given serialized arguments.

Parameters:

serialized_arguments (dict[str, str]) – The serialized arguments to deserialize.

Returns:

An Arguments object representing the deserialized arguments.

to_json() → str

Serializes the call into a JSON string.

Returns:

A JSON string representing the serialized call.

__getstate__() → dict

Gets the state of the Call object for serialization purposes.

Returns:

A dictionary representing the state of the Call object.

__setstate__(state: dict) → None

Sets the state of the Call object from the provided dictionary.

Parameters:

state (dict) – A dictionary representing the state to set.

classmethod from_json(app: pynenc.app.Pynenc, serialized: str) → pynenc.call.Call

Creates a Call object from a serialized JSON string.

Parameters:

• app (Pynenc) – The Pynenc application instance.

• serialized (str) – The serialized JSON string representing the call.

Returns:

A Call object created from the serialized data.

__str__() → str

__repr__() → str

__hash__() → int

__eq__(other: Any) → bool

class pynenc.call.PreSerializedCall

Bases: pynenc.call.Call[pynenc.types.Params, pynenc.types.Result]

Represents a call optimized for parallel routing with pre-serialized arguments. This call type is used for batch processing tasks with disabled concurrency control, where some arguments are pre-serialized (e.g., large shared data).

Parameters:

• task (Task[Params, Result]) – The task associated with the call.

• other_args (dict[str, Any]) – Unique arguments for this specific call.

• pre_serialized_args (dict[str, str]) – Pre-serialized common arguments.

other_args: dict[str, Any]

‘field(…)’

pre_serialized_args: dict[str, str]

‘field(…)’

_call: pynenc.call.Call[pynenc.types.Params, pynenc.types.Result] | None

‘field(…)’

_cached_arguments: pynenc.arguments.Arguments | None

‘field(…)’

_serialized_arguments: dict[str, str] | None

‘field(…)’

_args_hash: str | None

‘field(…)’

property call: Call[Params, Result]

property arguments: pynenc.arguments.Arguments

property call_id: str

Get the call_id from the underlying Call object.

Returns:

A string representing the unique identifier of the call.

property serialized_arguments: dict[str, str]

Serializes the call arguments into strings.

Returns:

A dictionary of serialized argument strings.

abstract property serialized_args_for_concurrency_check: dict[str, str] | None

__getstate__() → dict

__setstate__(state: dict) → None

abstract classmethod from_json(app: pynenc.app.Pynenc, serialized: str) → pynenc.call.Call

PreSerializedCall doesn’t support from_json as it’s meant for batch routing.

Raises:

NotImplementedError – This method is not implemented for PreSerializedCall.

__str__() → str

abstract __hash__() → int

__eq__(other: Any) → bool