splurge-pub-sub 2025.3.2

A lightweight pub-sub framework.

splurge-pub-sub

A lightweight, thread-safe publish-subscribe framework for Python applications. Splurge provides simple, Pythonic in-process event communication with full type safety and comprehensive error handling.

Features

• Lightweight: Zero external dependencies, minimal footprint
• Thread-Safe: Full concurrency support with reentrant locks
• Type-Safe: Complete type annotations with mypy strict mode compliance
• Async Publishing: Non-blocking message dispatch via background worker thread
• Simple API: Subscribe, publish, unsubscribe with intuitive methods
• Decorator Syntax: @bus.on("topic") for simplified subscriptions
• Topic Filtering: Wildcard pattern matching for selective message delivery
• Correlation IDs: Cross-library event tracking and coordination
• Error Handling: Custom error handlers for failed callbacks
• PubSubAggregator: Aggregate messages from multiple PubSub instances
• PubSubSolo: Scoped singleton PubSub instances for multi-package scenarios
• Context Manager: Automatic resource cleanup with with statement
• 95% Coverage: Comprehensive test coverage across all features

Quick Start

Installation

pip install splurge-pub-sub

Basic Usage

from splurge_pub_sub import PubSub, Message

# Create a pub-sub bus
bus = PubSub()

# Subscribe to a topic
def handle_event(msg: Message) -> None:
    print(f"Received: {msg.data}")

sub_id = bus.subscribe("user.created", handle_event)

# Publish a message (non-blocking, returns immediately)
bus.publish("user.created", {"id": 123, "name": "Alice"})
bus.drain()  # Wait for message to be delivered (optional)
# Output: Received: {'id': 123, 'name': 'Alice'}

# Unsubscribe when done
bus.unsubscribe("user.created", sub_id)

Decorator API

@bus.on("user.updated")
def handle_user_updated(msg: Message) -> None:
    print(f"User updated: {msg.data}")

bus.publish("user.updated", {"id": 123, "status": "active"})
bus.drain()  # Wait for delivery if needed

Topic Filtering

from splurge_pub_sub import TopicPattern

# Create patterns with wildcards
pattern = TopicPattern("user.*")
pattern.matches("user.created")  # True
pattern.matches("user.updated")  # True
pattern.matches("order.created")  # False

# Patterns with ? for single character
pattern = TopicPattern("user.?.created")
pattern.matches("user.a.created")  # True
pattern.matches("user.ab.created")  # False

Correlation ID for Cross-Library Coordination

# Multiple libraries using same correlation_id
correlation_id = "workflow-123"

dsv_bus = PubSub(correlation_id=correlation_id)
tabular_bus = PubSub(correlation_id=correlation_id)

# Monitor all events with same correlation_id
monitor_bus = PubSub()
monitor_bus.subscribe("*", lambda m: print(f"[{m.correlation_id}] {m.topic}"), 
                      correlation_id=correlation_id)

dsv_bus.publish("dsv.file.loaded", {"file": "data.csv"})
tabular_bus.publish("tabular.table.created", {"rows": 100})
monitor_bus.drain()  # Wait for messages to be delivered
# Both messages received by monitor

Error Handling

def my_error_handler(exc: Exception, topic: str) -> None:
    print(f"Error on topic '{topic}': {exc}")

bus = PubSub(error_handler=my_error_handler)

@bus.on("risky.operation")
def handle_event(msg: Message) -> None:
    raise ValueError("Something went wrong!")

bus.publish("risky.operation", {})
bus.drain()  # Wait for message delivery
# Output: Error on topic 'risky.operation': Something went wrong!

PubSubAggregator - Aggregating Multiple PubSub Instances

from splurge_pub_sub import PubSubAggregator, PubSub, Message

# Create PubSub instances from different packages/modules
pack_b_bus = PubSub()
pack_c_bus = PubSub()

# Create composite to aggregate messages from both
aggregator = PubSubAggregator(pubsubs=[pack_b_bus, pack_c_bus])

# Subscribe once to receive events from any managed PubSub
def unified_handler(msg: Message) -> None:
    print(f"Received from {msg.topic}: {msg.data}")

aggregator.subscribe("user.created", unified_handler, correlation_id="*")

# Publish from different PubSub instances
pack_b_bus.publish("user.created", {"id": 1, "source": "pack-b"})
pack_c_bus.publish("user.created", {"id": 2, "source": "pack-c"})

# Drain all buses to ensure messages are forwarded
pack_b_bus.drain()
pack_c_bus.drain()
aggregator.drain()

# Messages from both PubSub instances are received by composite subscribers

PubSubSolo - Scoped Singleton for Multi-Package Scenarios

from splurge_pub_sub import PubSubSolo, PubSubAggregator

# Each package gets its own singleton instance
bus_a = PubSubSolo.get_instance(scope="package_a")
bus_b = PubSubSolo.get_instance(scope="package_b")

# Aggregate multiple package singletons
aggregator = PubSubAggregator(pubsubs=[bus_a, bus_b])

# Subscribe to events from all packages
@aggregator.on("*")
def handle_all_events(msg: Message) -> None:
    print(f"Received: {msg.topic} - {msg.data}")

# Publish from different packages
bus_a.publish("package_a.event", {"id": 1})
bus_b.publish("package_b.event", {"id": 2})
aggregator.drain()

Context Manager

with PubSub() as bus:
    bus.subscribe("topic", callback)
    bus.publish("topic", data)
    bus.drain()  # Wait for delivery if needed
    # Cleanup happens automatically

Documentation

• README-DETAILS.md - Comprehensive developer's guide with features, examples, and API overview
• API-REFERENCE.md - Complete API reference with all classes, methods, and error types
• CLI-REFERENCE.md - Command-line interface documentation
• CHANGELOG.md - Version history and release notes

Requirements

• Python 3.10 or later
• No external dependencies

License

MIT License - see LICENSE file for details

Author

Jim Schilling

Support

For issues, questions, or contributions, visit the GitHub repository.