snoozeml.config — Configuration, Worker Scaffolding & Daemon Runner#

Programmatic configuration builder, custom worker scaffolder, and local daemon runner for SnoozeML.

Generates validated models.toml, streams.toml, and models.json configuration files matching SnoozeML’s Rust configuration schema (src/config.rs), scaffolds process-isolated Python worker directories (snooze_model.py), and manages local snooze serve child processes from Python.

class snoozeml.config.GlobalDefaultsConfig(idle_timeout_secs=300, max_batch_size=16, max_batch_delay_ms=10, device='auto', max_loaded_models=None, min_residency_secs=10, autoscale_queue_threshold=None, max_replicas=None, routing_strategy=None, pinned=None)[source]#

Bases: object

Global fallback model settings matching [defaults] in models.toml.

Parameters:

  • idle_timeout_secs (int | None)

  • max_batch_size (int | None)

  • max_batch_delay_ms (int | None)

  • device (str | None)

  • max_loaded_models (int | None)

  • min_residency_secs (int | None)

  • autoscale_queue_threshold (int | None)

  • max_replicas (int | None)

  • routing_strategy (str | None)

  • pinned (bool | None)

idle_timeout_secs: int | None = 300#
max_batch_size: int | None = 16#
max_batch_delay_ms: int | None = 10#
device: str | None = 'auto'#
max_loaded_models: int | None = None#
min_residency_secs: int | None = 10#
autoscale_queue_threshold: int | None = None#
max_replicas: int | None = None#
routing_strategy: str | None = None#
pinned: bool | None = None#
to_dict()[source]#

Serializes non-None global default fields into a dictionary.

Return type:

Dict[str, Any]

class snoozeml.config.ModelConfigEntry(id, path, description=None, device=None, max_batch_size=None, max_batch_delay_ms=None, idle_timeout_secs=None, min_residency_secs=None, autoscale_queue_threshold=None, max_replicas=None, routing_strategy=None, pinned=None)[source]#

Bases: object

Per-model configuration entry matching [[models]] in models.toml.

Parameters:

  • id (str)

  • path (str)

  • description (str | None)

  • device (str | None)

  • max_batch_size (int | None)

  • max_batch_delay_ms (int | None)

  • idle_timeout_secs (int | None)

  • min_residency_secs (int | None)

  • autoscale_queue_threshold (int | None)

  • max_replicas (int | None)

  • routing_strategy (str | None)

  • pinned (bool | None)

id: str#
path: str#
description: str | None = None#
device: str | None = None#
max_batch_size: int | None = None#
max_batch_delay_ms: int | None = None#
idle_timeout_secs: int | None = None#
min_residency_secs: int | None = None#
autoscale_queue_threshold: int | None = None#
max_replicas: int | None = None#
routing_strategy: str | None = None#
pinned: bool | None = None#
to_dict()[source]#

Serializes this model entry into a dictionary.

Return type:

Dict[str, Any]

class snoozeml.config.StreamBinding(name='default-stream', broker='redis', endpoint='redis://127.0.0.1:6379', input_topic='snooze.requests', model_id=None, output_topic='snooze.responses', error_topic='snooze.dlq', consumer_group='snoozeml-group', auto_create_topics=True, concurrency=16, partitions=1, format=StreamWireFormat.JSON)[source]#

Bases: object

Event stream binding configuration matching [[streams]] in models.toml.

Parameters:
name#

Unique human-readable identifier for this stream worker binding.

Type:

str

broker#

Broker type ("kafka", "rabbitmq", "redis", or "mock").

Type:

str

endpoint#

Broker connection URL or bootstrap servers (e.g. "redis://127.0.0.1:6379").

Type:

str

input_topic#

Topic, queue name, or Redis stream key to consume requests from.

Type:

str

model_id#

Optional target model ID (if None, dynamically routes per message model_id).

Type:

str | None

output_topic#

Optional destination topic/queue/stream for responses (default "snooze.responses").

Type:

str | None

error_topic#

Optional Dead-Letter Queue (DLQ) topic for failed messages (default "snooze.dlq").

Type:

str | None

consumer_group#

Optional consumer group identifier (default "snoozeml-group").

Type:

str | None

auto_create_topics#

Whether to auto-create topics/queues on startup (default True).

Type:

bool

concurrency#

Max in-flight inference requests for backpressure control (default 16).

Type:

int

partitions#

Partition count when auto-creating topics (default 1).

Type:

int

format#

Wire serialization format ("json" or "protobuf").

Type:

snoozeml.streams.StreamWireFormat | str

name: str = 'default-stream'#
broker: str = 'redis'#
endpoint: str = 'redis://127.0.0.1:6379'#
input_topic: str = 'snooze.requests'#
model_id: str | None = None#
output_topic: str | None = 'snooze.responses'#
error_topic: str | None = 'snooze.dlq'#
consumer_group: str | None = 'snoozeml-group'#
auto_create_topics: bool = True#
concurrency: int = 16#
partitions: int = 1#
format: StreamWireFormat | str = 'json'#
to_dict()[source]#

Serializes this stream binding into a dictionary.

Return type:

Dict[str, Any]

class snoozeml.config.ApiKeyEntry(role=Role.INFERENCE, key=None, key_env=None, name=None)[source]#

Bases: object

Static API key configuration entry matching [[auth.api_keys]].

Parameters:
role: Role | str = 'inference'#
key: str | None = None#
key_env: str | None = None#
name: str | None = None#
to_dict()[source]#

Serializes this API key entry into a dictionary.

Return type:

Dict[str, Any]

class snoozeml.config.JwtSettings(secret=None, secret_env=None, algorithm=JwtAlgorithm.HS256, issuer=None, audience=None, leeway_secs=60)[source]#

Bases: object

JWT authentication configuration matching [auth.jwt].

Parameters:
secret: str | None = None#
secret_env: str | None = None#
algorithm: JwtAlgorithm | str = 'HS256'#
issuer: str | None = None#
audience: str | None = None#
leeway_secs: int = 60#
to_dict()[source]#

Serializes this JWT configuration into a dictionary.

Return type:

Dict[str, Any]

class snoozeml.config.AuthSettings(enabled=True, protect_metrics=False, api_keys=<factory>, jwt=None)[source]#

Bases: object

Authentication and RBAC settings matching [auth] in models.toml.

Parameters:
enabled: bool = True#
protect_metrics: bool = False#
api_keys: List[ApiKeyEntry]#
jwt: JwtSettings | None = None#
to_dict()[source]#

Serializes this authentication configuration into a dictionary.

Return type:

Dict[str, Any]

class snoozeml.config.SnoozeConfig(defaults=<factory>, models=<factory>, streams=<factory>, auth=None)[source]#

Bases: object

Complete programmatic configuration for a SnoozeML model directory and streaming pipelines.

Allows Python users to configure models, Kafka/RabbitMQ/Redis streams, and RBAC authentication purely in Python and export them to models.toml or models.json for the snooze binary.

Example:

from snoozeml import SnoozeConfig

cfg = (
    SnoozeConfig()
    .add_model("iris-rf", path="iris_rf.onnx", pinned=True, max_batch_size=32)
    .add_stream(
        name="redis-pipeline",
        broker="redis",
        endpoint="redis://127.0.0.1:6379",
        input_topic="snooze.requests",
        output_topic="snooze.responses",
        model_id="iris-rf",
    )
    .add_api_key(role="admin", key_env="SNOOZE_ADMIN_KEY", name="admin-cli")
)
cfg.save("models/models.toml")
Parameters:
defaults: GlobalDefaultsConfig#
models: List[ModelConfigEntry]#
streams: List[StreamBinding]#
auth: AuthSettings | None = None#
add_model(id, path, *, description=None, device=None, max_batch_size=None, max_batch_delay_ms=None, idle_timeout_secs=None, min_residency_secs=None, autoscale_queue_threshold=None, max_replicas=None, routing_strategy=None, pinned=None)[source]#

Appends a model configuration entry and returns self for fluent chaining.

Parameters:

  • id (str)

  • path (str)

  • description (str | None)

  • device (str | None)

  • max_batch_size (int | None)

  • max_batch_delay_ms (int | None)

  • idle_timeout_secs (int | None)

  • min_residency_secs (int | None)

  • autoscale_queue_threshold (int | None)

  • max_replicas (int | None)

  • routing_strategy (str | None)

  • pinned (bool | None)

Return type:

SnoozeConfig

add_stream(name='default-stream', broker='redis', endpoint='redis://127.0.0.1:6379', input_topic='snooze.requests', *, model_id=None, output_topic='snooze.responses', error_topic='snooze.dlq', consumer_group='snoozeml-group', auto_create_topics=True, concurrency=16, partitions=1, format=StreamWireFormat.JSON)[source]#

Appends a message stream binding (Kafka, RabbitMQ, Redis, or Mock) and returns self.

Parameters:
Return type:

SnoozeConfig

add_api_key(role=Role.INFERENCE, *, key=None, key_env=None, name=None, protect_metrics=None)[source]#

Adds an API key entry to [auth] (enabling authentication) and returns self.

Parameters:

  • role (Role | str)

  • key (str | None)

  • key_env (str | None)

  • name (str | None)

  • protect_metrics (bool | None)

Return type:

SnoozeConfig

set_jwt(*, secret=None, secret_env=None, algorithm=JwtAlgorithm.HS256, issuer=None, audience=None, leeway_secs=60, protect_metrics=None)[source]#

Configures [auth.jwt] verification settings and returns self.

Parameters:
Return type:

SnoozeConfig

to_dict()[source]#

Serializes the entire configuration into a JSON-compatible dictionary.

Return type:

Dict[str, Any]

to_json(indent=2)[source]#

Serializes the configuration into a formatted JSON string (models.json).

Parameters:

indent (int)

Return type:

str

to_toml()[source]#

Serializes the configuration into a TOML string compatible with src/config.rs.

Return type:

str

save(path)[source]#

Writes the configuration to disk as TOML (.toml) or JSON (.json).

Parameters:

path (str | Path) – Target filesystem path (e.g. "models/models.toml" or "models/models.json").

Returns:

Resolved pathlib.Path of the written configuration file.

Return type:

Path

classmethod from_dict(data)[source]#

Constructs a SnoozeConfig from a parsed dictionary.

Parameters:

data (Mapping[str, Any])

Return type:

SnoozeConfig

snoozeml.config.scaffold_worker(output_dir, *, model_code=None, entrypoint='snooze_model.py', class_name='Model', tensor_format='numpy', python=None, env=None, requirements=None, verify=True, verify_inputs=None, verify_device='cpu')[source]#

Scaffolds a process-isolated custom Python worker model directory (snooze_model.py).

Creates snooze_model.py, snooze_model.json, and an optional requirements.txt inside output_dir, and optionally runs snoozeml.worker.run_check to verify that the worker loads and predicts cleanly.

Parameters:

  • output_dir (str | Path) – Target model directory (e.g. "models/my-custom-worker").

  • model_code (str | None) – Python source code defining the worker model class. If None, generates a ready-to-customize template class.

  • entrypoint (str) – Entrypoint script filename (default "snooze_model.py").

  • class_name (str) – Model class name inside entrypoint (default "Model").

  • tensor_format (str) – Input tensor format passed to predict() ("numpy" or "raw").

  • python (str | None) – Optional custom Python interpreter path written to snooze_model.json.

  • env (Mapping[str, str] | None) – Optional environment variables mapping written to snooze_model.json.

  • requirements (Sequence[str] | None) – Optional list of pip requirement specifiers written to requirements.txt.

  • verify (bool) – If True, smoke-tests the generated worker immediately via run_check().

  • verify_inputs (Sequence[str] | None) – Input specification strings for run_check (default ["input:2x4"]).

  • verify_device (str) – Device string passed to run_check (default "cpu").

Returns:

Resolved pathlib.Path to the scaffolded worker directory.

Return type:

Path

snoozeml.config.find_snooze_binary(explicit_path=None)[source]#

Locates the snooze daemon executable on the system.

Checks explicit_path, SNOOZE_BIN environment variable, system PATH, and local Cargo build directories (target/release/snooze, target/debug/snooze).

Parameters:

explicit_path (str | Path | None)

Return type:

str

class snoozeml.config.SnoozeDaemon(*, config=None, models_dir=None, host='127.0.0.1', port=8080, grpc_port=50051, grpc=True, demo=False, api_key=None, jwt_secret=None, binary_path=None, startup_timeout=15.0)[source]#

Bases: object

Context-managed local snooze serve process controller for Python workflows.

Allows launching a configured SnoozeML daemon (including [[streams]], [[models]], and [auth]) directly from Python scripts, tests, or Jupyter notebooks without modifying the Rust binary.

Parameters:

  • config (Optional[SnoozeConfig])

  • models_dir (Optional[Union[str, Path]])

  • host (str)

  • port (int)

  • grpc_port (int)

  • grpc (bool)

  • demo (bool)

  • api_key (Optional[str])

  • jwt_secret (Optional[str])

  • binary_path (Optional[Union[str, Path]])

  • startup_timeout (float)

property http_endpoint: str#

Returns the HTTP base URL for this daemon instance.

property grpc_endpoint: str#

Returns the gRPC host:port target for this daemon instance.

build_command(resolved_models_dir, config_path)[source]#

Constructs the snooze serve CLI command list.

Parameters:

  • resolved_models_dir (Path)

  • config_path (Path | None)

Return type:

List[str]

start()[source]#

Starts the snooze serve child process and waits until /health/ready responds.

Return type:

SnoozeDaemon

stop()[source]#

Stops the snooze serve child process and cleans up temporary directories.

Return type:

None

client(*, use_grpc=False)[source]#

Returns a configured SnoozeClient connected to this daemon.

Parameters:

use_grpc (bool)

Return type:

SnoozeClient

async_client(*, use_grpc=False)[source]#

Returns a configured AsyncSnoozeClient connected to this daemon.

Parameters:

use_grpc (bool)

Return type:

AsyncSnoozeClient