pynenc.broker.sqlite_broker

SQLite-based broker for cross-process testing.

This module provides a SQLite-based broker implementation that enables true cross-process coordination for testing process runners.

Module Contents

Classes

Tables
SQLiteBroker	A SQLite-based implementation of the broker for cross-process testing.

API

class pynenc.broker.sqlite_broker.Tables

QUEUE

‘broker_message_queue’

class pynenc.broker.sqlite_broker.SQLiteBroker(app: pynenc.app.Pynenc)

Bases: pynenc.broker.base_broker.BaseBroker

A SQLite-based implementation of the broker for cross-process testing.

Uses SQLite for cross-process message queue coordination and implements all required abstract methods from BaseBroker. It’s designed specifically for testing process runners.

Warning

The SQLiteBroker class is designed for testing purposes only and should not be used in production systems. It uses temporary SQLite files for state.

Initialization

_init_tables() → None

Initialize SQLite tables for broker.

conf() → pynenc.conf.config_broker.ConfigBrokerSQLite

send_message(invocation_id: pynenc.identifiers.invocation_id.InvocationId) → None

Send a message (invocation ID) to the queue.

route_invocation(invocation_id: pynenc.identifiers.invocation_id.InvocationId) → None

Route a single invocation ID by sending it to the message queue.

route_invocations(invocation_ids: list[pynenc.identifiers.invocation_id.InvocationId]) → None

Route multiple invocation IDs by sending them to the message queue.

retrieve_invocation() → InvocationId | None

Atomically retrieve and remove a single invocation from the queue. Ensures that no two processes can retrieve the same invocation.

Returns:

The next invocation ID in the queue, or None if empty.

count_invocations() → int

Count the number of invocations in the queue.

purge() → None

Clear all broker messages.