Skip to content

ARCHITECTURE.md

Latest commit

 

History

History
181 lines (147 loc) · 5.21 KB

ARCHITECTURE.md

File metadata and controls

181 lines (147 loc) · 5.21 KB
Raw

Streamix Queue - Package Structure

Project Layout

Streamix/
├── src/streamix_queue/          # Main package source
│   ├── __init__.py              # Public API exports (publish, consume)
│   ├── _version.py              # Version metadata
│   ├── simple_api.py            # High-level API (publish, consume functions)
│   ├── consumer.py              # ConsumerEngine - retry/DLQ orchestration
│   ├── handler.py               # HandlerExecutor - handler invocation logic
│   ├── producer.py              # Producer class - event publishing
│   ├── transport.py             # RedisStreamTransport - Redis I/O layer
│   └── schemas.py               # StreamMessage class - event schema
├── examples/
│   ├── tasks_app.py             # Example: Consumer (listener)
│   └── publish_event.py         # Example: Publisher
├── pyproject.toml               # Modern Python packaging config
├── setup.py                     # Setuptools entry point
├── MANIFEST.in                  # File inclusion rules for distribution
├── LICENSE                      # MIT license
├── .gitignore                   # Git ignore rules
├── README.md                    # Full documentation for PyPI
└── PYPI_RELEASE.md             # Instructions for publishing to PyPI

dist/                            # Generated (after running `python -m build`)
├── streamix_queue-0.1.0-py3-none-any.whl     # Wheel distribution
└── streamix_queue-0.1.0.tar.gz               # Source distribution

Core Modules

simple_api.py - Public API

Provides the main entry points:

  • publish(event, data, ...) → Returns StreamMessage
  • consume(event, handler, ...) → Blocks and processes events

consumer.py - Consumer Engine

Handles the event consumption loop:

  • Reads messages from Redis consumer groups
  • Executes handler functions
  • Implements retry logic (increments retries on failure)
  • Sends permanently failed messages to DLQ
  • Acknowledges processed messages

transport.py - Redis Transport

Low-level Redis Streams operations:

  • XADD - Add message to stream
  • XREADGROUP - Read from consumer group
  • XACK - Acknowledge message
  • XAUTOCLAIM - Reclaim stale messages
  • XGROUP CREATE - Create consumer group

schemas.py - Message Schema

Defines the event message structure:

  • StreamMessage - Full message object
  • MessageTimestamps - Tracking creation and updates
  • Structured JSON serialization/deserialization

handler.py - Handler Execution

Intelligent handler invocation:

  • Detects handler signature (1 or 2+ parameters)
  • Calls handler(data) or handler(data, message) accordingly

producer.py - Producer

Wrapper around transport for publishing:

  • Creates StreamMessage
  • Calls transport.publish()

Configuration - pyproject.toml

Key Sections:

  • [project] - Package metadata (name, version, description)
  • [project.dependencies] - Runtime dependencies (redis>=5.0.0)
  • [project.optional-dependencies] - Dev tools (pytest, mypy, black, ruff)
  • [project.urls] - Links to GitHub, docs, issues
  • [tool.setuptools] - Package discovery and source location
  • [tool.black] - Code formatter config
  • [tool.ruff] - Linter config

Building for PyPI

One-Command Build

python -m build

Creates:

  • dist/streamix_queue-0.1.0-py3-none-any.whl (wheel - faster install)
  • dist/streamix_queue-0.1.0.tar.gz (sdist - source distribution)

Upload to PyPI

pip install twine
twine upload dist/*

Installation (After PyPI)

pip install streamix-queue

Then import:

from streamix_queue import publish, consume

Data Flow

Publishing Flow:

User calls publish()
  ↓
simple_api.py wraps Producer
  ↓
Producer creates StreamMessage
  ↓
transport.publish() calls XADD to Redis
  ↓
Message stored in Redis Stream

Consuming Flow:

User calls consume()
  ↓
simple_api.py creates ConsumerEngine
  ↓
ConsumerEngine.run_forever() loops:
  ├─ transport.read_group() → XREADGROUP from Redis
  ├─ transport.claim_stale_messages() → XAUTOCLAIM for recovery
  ├─ For each message:
  │  ├─ handler.execute() → calls user's handler function
  │  ├─ transport.ack() → XACK to mark processed
  │  └─ On error:
  │     ├─ Retry limit not reached: republish with incremented retries
  │     └─ Retry limit exceeded: publish_dead_letter() to <stream>:failed

Version Updates

To release a new version:

  1. Update version in pyproject.toml:

    version = "0.2.0"

  2. Update src/streamix_queue/_version.py if desired

  3. Rebuild:

    python -m build

  4. Upload:

    twine upload dist/* --skip-existing

Testing Locally

# Install in dev mode
pip install -e .

# Run examples
python examples/publish_event.py
python examples/tasks_app.py (in another terminal)

# Verify imports
python -c "from streamix_queue import publish, consume; print('OK')"

Resources

  • See PYPI_RELEASE.md for detailed PyPI upload instructions
  • See README.md for full API documentation
  • See examples/ directory for usage examples