`pynenc.builder`¶

Builder pattern implementation for configuring Pynenc applications with plugin support.

This module provides a fluent, chainable builder interface that can be extended by plugins through Python’s entry points system. Plugins automatically register their builder methods when installed, enabling seamless integration of backend-specific functionality.

Key components:

PynencBuilder: Core builder class with plugin support
Plugin registration system via entry points
Dynamic method resolution for plugin-provided methods
Validation system for plugin configurations

Module Contents¶

Classes¶

PynencBuilder

A builder pattern implementation for creating and configuring Pynenc applications.

API¶

class pynenc.builder.PynencBuilder[source]¶

A builder pattern implementation for creating and configuring Pynenc applications.

This builder simplifies the configuration process by providing intuitive method chaining to set up various components of a Pynenc application. Backend-specific methods are provided by installed plugins through Python’s entry points system.

The plugin system automatically discovers and registers methods from installed plugins, enabling backend-specific configuration methods to become available when plugins are installed.

Example:

# With backend plugins installed
pynenc_app = (
    PynencBuilder()
    .app_id("my_application")
    .serializer_pickle()
    .multi_thread_runner(min_threads=1, max_threads=4)
    .logging_level("info")
    .build()
)

# Memory-based configuration for development
pynenc_app = (
    PynencBuilder()
    .app_id("my_application")
    .memory()  # Built-in memory backend
    .mem_client_data_store(min_size_to_cache=1024)
    .build()
)

Initialization

Initialize a new PynencBuilder with plugin discovery and empty configuration.

Automatically discovers and loads plugin methods via entry points on first instantiation.

_plugin_methods: dict[str, collections.abc.Callable]¶: None

_plugin_validators: list[collections.abc.Callable]¶: []

_plugins_loaded¶: False

classmethod _load_plugins() → None[source]¶

Load and register plugin methods via Python entry points system.

Discovers all plugins registered under ‘pynenc.plugins’ entry point group and registers their builder methods for dynamic resolution.

classmethod _load_plugins_importlib() → None[source]¶: Load plugins using importlib.metadata.

classmethod _load_plugins_pkg_resources() → None[source]¶: Load plugins using pkg_resources.

classmethod _register_plugin_from_entry_point(entry_point: Any) → None[source]¶: Register a single plugin from an entry point.

classmethod register_plugin_method(method_name: str, method_func: collections.abc.Callable) → None[source]¶

This allows plugins to extend the builder with their own configuration methods. Plugins should call this during their registration process.

Parameters:

method_name (str) – Name of the method to register (e.g., ‘redis’, ‘mongodb’)
method_func (Callable) – The method function that takes builder as first argument

Raises:

ValueError – If method name conflicts with existing core methods

classmethod register_plugin_validator(validator_func: collections.abc.Callable) → None[source]¶

Validators should raise ValueError if the configuration is invalid.

Parameters:: validator_func (Callable) – Function that takes (config: dict) and raises ValueError if invalid

__getattr__(name: str) → collections.abc.Callable[source]¶

Dynamic method resolution for plugin-registered methods.

This enables plugins to add methods that are automatically available when the plugin is installed.

Parameters:: name (str) – Method name being accessed
Returns:: Bound method from plugin
Raises:: AttributeError – If method not found in plugins with helpful error message

__dir__() → list[str][source]¶: Include plugin methods in dir() for discoverability.

app_id(app_id: str) → pynenc.builder.PynencBuilder[source]¶

Set the application ID for the Pynenc application.

The application ID uniquely identifies this Pynenc application instance and is used in logging, monitoring, and component configuration.

Parameters:: app_id (str) – The unique identifier for this application
Returns:: The builder instance for method chaining

memory() → pynenc.builder.PynencBuilder[source]¶

Configure in-memory components for the Pynenc application.

This sets up all components (orchestrator, broker, state backend, and argument cache) to use in-memory backends. This is primarily for testing and development purposes.

Note: In-memory components are only compatible with certain runners.

Returns:: The builder instance for method chaining

sqlite(sqlite_db_path: str | None = None) → pynenc.builder.PynencBuilder[source]¶

Configure SQLite components for the Pynenc application.

This sets up all components (orchestrator, broker, state backend, and argument cache) to use SQLite backends. This is primarily for testing multiprocess scenarios on a single machine.

Note: SQLite will not work in distributed environments with independent DB files.

Parameters:: sqlite_db_path (str | None) – Path to the SQLite database file. If None, uses the default location.
Returns:: The builder instance for method chaining

mem_client_data_store(min_size_to_cache: int = 1024, local_cache_size: int = 1024) → pynenc.builder.PynencBuilder[source]¶

Configure memory-based client data store.

This sets up in-memory argument caching for development and testing purposes. Arguments larger than the specified threshold will be cached locally. Also configures the ClientDataStore with the same settings.

Parameters:

min_size_to_cache (int) – Minimum string length required to cache an argument
local_cache_size (int) – Maximum number of items to cache locally

Returns:

The builder instance for method chaining

disable_client_data_store() → pynenc.builder.PynencBuilder[source]¶

Disable client data store completely.

This turns off all client data store functionality, which may be useful for debugging or when caching is not desired.

Returns:: The builder instance for method chaining

mem_trigger(scheduler_interval_seconds: int = 60, enable_scheduler: bool = True) → pynenc.builder.PynencBuilder[source]¶

Configure memory-based trigger system.

This sets up in-memory triggers for development and testing purposes. Time-based triggers will be checked at the specified interval.

Parameters:

scheduler_interval_seconds (int) – Interval for the scheduler to check time-based triggers
enable_scheduler (bool) – Whether to enable the scheduler for time-based triggers

Returns:

The builder instance for method chaining

multi_thread_runner(min_threads: int = 1, max_threads: int = 1, enforce_max_processes: bool = False) → pynenc.builder.PynencBuilder[source]¶

Configure the MultiThreadRunner for concurrent task execution.

The MultiThreadRunner uses threads to execute tasks concurrently within the same process, providing efficient parallel execution with shared memory.

Parameters:

min_threads (int) – The minimum number of threads to keep in the thread pool
max_threads (int) – The maximum number of threads allowed in the thread pool
enforce_max_processes (bool) – If True, enforces the maximum number of processes

Returns:

The builder instance for method chaining

persistent_process_runner(num_processes: int = 0) → pynenc.builder.PynencBuilder[source]¶

Configure the PersistentProcessRunner for concurrent task execution.

The PersistentProcessRunner maintains a pool of persistent processes for task execution, providing true parallel execution across multiple CPU cores, with isolated memory spaces.

Parameters:: num_processes (int) – The number of processes to create in the process pool
Returns:: The builder instance for method chaining

thread_runner(min_threads: int = 1, max_threads: int = 0) → pynenc.builder.PynencBuilder[source]¶

Configure the ThreadRunner for task execution.

The ThreadRunner uses a thread pool to execute tasks concurrently within the same process, providing efficient parallel execution with shared memory.

Parameters:

min_threads (int) – The minimum number of threads to keep in the thread pool
max_threads (int) – The maximum number of threads allowed in the thread pool

Returns:

The builder instance for method chaining

process_runner() → pynenc.builder.PynencBuilder[source]¶

Configure the ProcessRunner for task execution.

The ProcessRunner creates a new process for each task execution, providing isolated execution context for each task.

Returns:: The builder instance for method chaining

dummy_runner() → pynenc.builder.PynencBuilder[source]¶

Configure the DummyRunner for task execution.

The DummyRunner executes tasks in the main thread of the application. This is useful for testing and debugging purposes.

Returns:: The builder instance for method chaining

dev_mode(force_sync_tasks: bool = True) → pynenc.builder.PynencBuilder[source]¶

Enable development mode for easier debugging.

In development mode, tasks can be forced to run synchronously, making debugging and testing easier.

Parameters:: force_sync_tasks (bool) – If True, forces all tasks to run synchronously
Returns:: The builder instance for method chaining

logging(level: str | None = None, *, stream: str | None = None, colors: bool | None = None, fmt: str | None = None) → pynenc.builder.PynencBuilder[source]¶

Configure all logging options in a single call.

Each parameter is optional — only non-None values are applied, so you can mix this with the individual logging_* methods freely.

Parameters:

level (str | None) – Log level — "debug", "info", "warning", "error", "critical"
stream (str | None) – Output stream — "stdout" or "stderr"
colors (bool | None) – True force, False disable, None auto-detect
fmt (str | None) – Log format — "text" or "json"

Returns:

The builder instance for method chaining

Raises:

ValueError – If any provided value is invalid

logging_level(level: str) → pynenc.builder.PynencBuilder[source]¶

Set the logging level for the application.

Parameters:: level (str) – Log level — "debug", "info", "warning", "error", "critical"
Returns:: The builder instance for method chaining
Raises:: ValueError – If an invalid logging level is provided

logging_colors(use_colors: bool | None = None) → pynenc.builder.PynencBuilder[source]¶

Control ANSI colour output in logs.

Pass True to force colours, False to disable them, or None (the default) to let pynenc auto-detect based on whether the log stream is an interactive TTY. Set to False when running in containers or CI environments that do not support ANSI escape codes.

Parameters:: use_colors (bool | None) – True to force colours, False to disable, None for auto-detect
Returns:: The builder instance for method chaining

logging_stream(stream: str) → pynenc.builder.PynencBuilder[source]¶

Set which output stream log messages are written to.

Use "stdout" when deploying in containers (e.g. Kubernetes / GKE) so that log lines are not misclassified as errors by the container runtime’s default stderr-to-ERROR mapping.

Parameters:: stream (str) – Output stream — "stdout" or "stderr"
Returns:: The builder instance for method chaining
Raises:: ValueError – If an invalid stream name is provided

logging_format(fmt: str) → pynenc.builder.PynencBuilder[source]¶

Set the log output format.

"text" (default) emits human-readable lines compatible with pynmon Log Explorer. "json" emits structured JSON objects — ideal for log-aggregation pipelines such as GCP Cloud Logging, Datadog, or the ELK stack. The text field in the JSON payload preserves the human-readable representation so pynmon can still parse JSON logs.

Parameters:: fmt (str) – Log format — "text" or "json"
Returns:: The builder instance for method chaining
Raises:: ValueError – If an invalid format name is provided

runner_tuning(runner_loop_sleep_time_sec: float = 0.01, invocation_wait_results_sleep_time_sec: float = 0.01, min_parallel_slots: int = 1) → pynenc.builder.PynencBuilder[source]¶

Configure runner performance tuning parameters.

Parameters:

runner_loop_sleep_time_sec (float) – Sleep time between runner loop iterations
invocation_wait_results_sleep_time_sec (float) – Sleep time when waiting for results
min_parallel_slots (int) – Minimum number of parallel execution slots

Returns:

The builder instance for method chaining

task_control(blocking_control: bool = False, queue_timeout_sec: float = 0.1) → pynenc.builder.PynencBuilder[source]¶

Configure task control parameters.

Parameters:

blocking_control (bool) – Whether to enable blocking control for concurrent tasks
queue_timeout_sec (float) – Timeout for queue operations in seconds

Returns:

The builder instance for method chaining

serializer_json_pickle() → pynenc.builder.PynencBuilder[source]¶

Configure the JSON-Pickle hybrid serializer

Returns:: The builder instance for method chaining

serializer_json() → pynenc.builder.PynencBuilder[source]¶

Configure the JSON serializer

Returns:: The builder instance for method chaining

serializer_pickle() → pynenc.builder.PynencBuilder[source]¶

Configure the Pickle serializer

Returns:: The builder instance for method chaining

concurrency_control(running_concurrency: str | pynenc.conf.config_task.ConcurrencyControlType | None = None, registration_concurrency: str | pynenc.conf.config_task.ConcurrencyControlType | None = None) → pynenc.builder.PynencBuilder[source]¶

Configure concurrency control default behaviors for all tasks.

Parameters:

running_concurrency (Optional[Union[str, ConcurrencyControlType]]) – Controls runtime concurrency behavior
registration_concurrency (Optional[Union[str, ConcurrencyControlType]]) – Controls registration concurrency behavior

Returns:

The builder instance for method chaining

max_pending_seconds(seconds: float) → pynenc.builder.PynencBuilder[source]¶

Set the maximum time a task can remain in PENDING state.

Parameters:: seconds (float) – Maximum time in seconds a task can remain in PENDING state
Returns:: The builder instance for method chaining

argument_print_mode(mode: str | pynenc.conf.config_pynenc.ArgumentPrintMode, truncate_length: int = 32) → pynenc.builder.PynencBuilder[source]¶

Configure how task arguments are printed in logs.

Parameters:

mode (Union[str, ArgumentPrintMode]) – The print mode to use
truncate_length (int) – Maximum length for printed argument values in TRUNCATED mode

Returns:

The builder instance for method chaining

Raises:

ValueError – If an invalid mode string is provided

hide_arguments() → pynenc.builder.PynencBuilder[source]¶

Configure logs to hide all task arguments.

Returns:: The builder instance for method chaining

show_argument_keys() → pynenc.builder.PynencBuilder[source]¶

Configure logs to show only argument names.

Returns:: The builder instance for method chaining

show_full_arguments() → pynenc.builder.PynencBuilder[source]¶

Configure logs to show complete argument values without truncation.

Returns:: The builder instance for method chaining

show_truncated_arguments(truncate_length: int = 32) → pynenc.builder.PynencBuilder[source]¶

Configure logs to show truncated argument values.

Parameters:: truncate_length (int) – Maximum length for printed argument values
Returns:: The builder instance for method chaining

custom_config(**kwargs: Any) → pynenc.builder.PynencBuilder[source]¶

Add arbitrary configuration values.

For common configuration values, prefer using the dedicated methods instead of this generic method.

Parameters:: kwargs (Any) – Custom configuration values to add
Returns:: The builder instance for method chaining

trigger_task_modules(modules: collections.abc.Iterable[str]) → pynenc.builder.PynencBuilder[source]¶

Declare modules that contain trigger-dependent tasks.

Runners will import these modules at startup to ensure tasks that depend on triggers (cron, event, status, etc.) are registered. This avoids importing all task modules during normal application import time and prevents eager connections to external systems.

Parameters:: modules (Iterable[str]) – Iterable of module import paths (e.g. [“myapp.tasks.scheduled”]).
Returns:: The builder instance for method chaining

This method accepts any iterable of strings and stores the value as a deduplicated set[str] in the configuration. Using a set ensures uniqueness and avoids ordering assumptions.

_validate_plugin_compatibility() → None[source]¶

Validate plugin configurations before building.

Runs all registered plugin validators to ensure the configuration is valid.

_validate_memory_compatibility() → None[source]¶

Validate that the selected runner is compatible with memory components.

Raises:: ValueError – If memory components are used with an incompatible runner

build() → pynenc.Pynenc[source]¶

Build and return a configured Pynenc instance.

This method creates a new Pynenc instance using the configuration values that have been set through the builder methods.

Returns:: A configured Pynenc instance ready for use
Raises:: ValueError – If the configuration is invalid

pynenc.builder¶

Module Contents¶

Classes¶

API¶

`pynenc.builder`¶