splurge-pub-sub 2025.0.0

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
• 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
• Error Handling: Custom error handlers for failed callbacks
• 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
bus.publish("user.created", {"id": 123, "name": "Alice"})
# 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"})

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

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", {})  # Error handler called
# Output: Error on topic 'risky.operation': Something went wrong!

Context Manager

with PubSub() as bus:
    bus.subscribe("topic", callback)
    bus.publish("topic", data)
    # 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.