Architecture Handlers

A minimal, extensible Python library providing a base class (ArchitectureHandler) for building handler components that accept inputs, optionally perform network requests, and publish results via thread-safe queues. Ships with concrete handlers for AI/agent tasks: speech-to-text, text-to-speech, face recognition, gesture generation, LLM agent interaction, and Ollama chat.

Package: scs-architecture-handlers · Python: >= 3.10 · License: MIT

Table of Contents

• Features
• Installation
• Handlers Overview
• Design Decisions
• Usage
  • Direct synchronous generation
  • Threaded pipeline
  • Async wrappers
  • Callbacks & non-blocking execution
• Extension Points
• Configuration Reference
• Stats & Monitoring
• Development
  • Setup
  • Running Tests
  • System Dependencies
  • CI/CD & Releasing to PyPI
• License

Features

• Threaded processing loop — consumes items from an input queue, processes them, and writes results to a result queue.
• Pluggable core logic — override generate_results() or supply generate_results_callback in config.
• Template methods — validate_item, preprocess_item, postprocess_result for easy adaptation.
• Optional type checking — expected_type catches mismatched inputs at the boundary.
• Dropping policy — bounded queues with configurable capacity; drops are counted and logged.
• Runtime instrumentation — timestamps and counters (processed/dropped/errors/network errors) via stats().
• Async wrappers — all sync methods have async_* counterparts (powered by asyncio.to_thread).
• Hook methods — 9 instrumentation hooks (_hook_before_feed, _hook_after_generate, etc.) for monitoring.
• Non-blocking callbacks — dedicated callback worker thread so user callbacks never stall the hot path.
• Singleton sync core — async wrappers delegate to the same sync implementation; no code duplication.

Installation

From PyPI

pip install scs-architecture-handlers

from scs_architecture_handlers.base_handler import ArchitectureHandler

Development setup

# Clone the repository
git clone <repo-url>
cd architecture-handlers

# Create a virtual environment (Python >= 3.10)
python3.11 -m venv .venv
source .venv/bin/activate

# Install dependencies (using pip)
pip install -r requirements.txt

# Or using uv
uv sync

Handlers Overview

Design Decisions

1. Single sync core, async wrappers on top. All async methods delegate to the sync implementation via asyncio.to_thread. This means there is exactly one implementation of the processing logic — no duplicated sync/async code paths.

2. Template Method pattern. The base class defines the skeleton of the feed → validate → preprocess → generate → postprocess → publish pipeline. Subclasses override only the steps they need (validate_item, preprocess_item, generate_results, postprocess_result).

3. Bounded queues with backpressure. Both input and result queues have configurable max sizes. When a queue is full, items are silently dropped and dropped_count is incremented. This prevents unbounded memory growth under load.

4. Dedicated async callback thread. External callbacks (e.g., database writes, network calls) can be scheduled via _emit_async_callback to run on a separate worker thread, keeping the main processing path fast and predictable.

5. Hook-based instrumentation. All major lifecycle events fire hook methods. Subclasses and monitoring tools can override these hooks without modifying the core pipeline.

6. Dry-run mode in all network handlers. Every handler that performs network I/O supports a dry_run flag that returns stub results without making HTTP calls. This makes testing and offline development straightforward.

7. Weak reference finalizer. A weakref.finalize ensures background threads are joined when the handler is garbage-collected, preventing orphaned threads.

Usage

1) Direct synchronous generation

Subclass ArchitectureHandler and override generate_results:

from scs_architecture_handlers.base_handler import ArchitectureHandler


class EchoHandler(ArchitectureHandler):
    def generate_results(self, item, extra):
        return {"echo": item, "meta": extra}


h = EchoHandler(run_as_thread=False)
out = h.generate("hello", tag=1)
print(out)  # {"echo": "hello", "meta": {"tag": 1}}

h.cleanup()

2) Threaded pipeline

Use the internal worker thread to consume items from the input queue and push processed results to the result queue:

import time
from scs_architecture_handlers.base_handler import ArchitectureHandler


class Doubler(ArchitectureHandler):
    def generate_results(self, item, extra):
        return item * 2


h = Doubler(run_as_thread=True, disable_thread=False)
h.feed(10)
print(h.get_result(timeout=1.0))  # 20
h.cleanup()

3) Async wrappers

All public sync methods have async counterparts:

import asyncio
from scs_architecture_handlers.base_handler import ArchitectureHandler


class EchoHandler(ArchitectureHandler):
    def generate_results(self, item, extra):
        return {"echo": item}


async def main():
    h = EchoHandler(run_as_thread=False)
    out = await h.async_generate("async-hello")
    print(out)  # {"echo": "async-hello"}
    await h.async_cleanup()


asyncio.run(main())

Available async wrappers: async_feed, async_get_result, async_generate, async_start, async_cleanup.

4) Callbacks & non-blocking execution

Schedule external callbacks on a dedicated thread via _emit_async_callback:

import time
from scs_architecture_handlers.base_handler import ArchitectureHandler


def on_result(item, extra):
    time.sleep(0.1)
    print("callback got:", item)


class NonBlockingHandler(ArchitectureHandler):
    def generate_results(self, item, extra):
        result = {"value": str(item).upper(), "extra": extra}
        self._emit_async_callback(result, extra)
        return result


h = NonBlockingHandler(
    run_as_thread=False,
    disable_thread=False,
    generate_results_callback=on_result,
    callback_queue_size=8,
)

for i in range(5):
    out = h.generate(f"msg-{i}")
    assert out["value"] == f"MSG-{i}"

time.sleep(0.5)
h.cleanup()

Notes:

• If the callback queue fills, new tasks are dropped and dropped_count increments.
• For producer-style handlers (e.g., microphone/STT), use _emit_async_callback inside the production path.

Extension Points

Override these template methods to adapt the handler:

Instrumentation hooks (optional overrides):

Configuration Reference

Stats & Monitoring

stats = handler.stats()
print(stats)
# {
#   "client_id": "client-...",
#   "created_at": ...,
#   "last_feed_at": ...,
#   "last_generate_at": ...,
#   "last_result_at": ...,
#   "last_network_call_at": ...,
#   "input_queue_size": 0,
#   "result_queue_size": 0,
#   "run_as_thread": True,
#   "processed_count": 42,
#   "dropped_count": 0,
#   "error_count": 0,
#   "network_error_count": 0,
# }

All counters are updated automatically by the default hook implementations. Override hooks for custom monitoring.

Development

Setup

python3.11 -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt

Running Tests

Tests use pytest with pytest-asyncio and pytest-timeout. They mock external services so no live servers are needed.

pytest -q
# or
pytest tests/

Individual test files:

System Dependencies

On Debian/Ubuntu, you may need:

sudo apt-get update
sudo apt-get install -y portaudio19-dev

CI/CD & Releasing to PyPI

The project uses GitLab CI (.gitlab-ci.yml) with four stages:

1. test — runs pytest -q on all branches and merge requests.
2. secret-detection — GitLab Secret Detection template.
3. build — builds sdist+wheel on version tags.
4. deploy — publishes to PyPI via Twine on version tags.

To release a new version:

# 1. Update version in pyproject.toml
# 2. Commit and tag
git tag v0.2.6
git push origin v0.2.6

The deploy job verifies the tag matches pyproject.toml version automatically.

Prerequisites:

• Set a protected, masked CI variable PYPI_TOKEN with your PyPI API token in GitLab → Settings → CI/CD → Variables.
• Optionally override PYPI_REPOSITORY_URL to point at TestPyPI for testing.

License

MIT