Action Dispatch

A powerful and flexible Python library for multi-dimensional action dispatching. Route function calls dynamically based on context dimensions like user roles, environments, API versions, or any custom attributes.

Features

Multi-dimensional Routing - Dispatch based on multiple context attributes simultaneously Dynamic Handler Registration - Register handlers using decorators or programmatically Flexible Context Matching - Support for exact matches and fallback strategies

Installation

Requirements: Python 3.9+

pip install action-dispatch

Quick Start

from action_dispatch import ActionDispatcher

# Create dispatcher with dimensions
dispatcher = ActionDispatcher(['role', 'environment'])

# Register handlers using decorators
@dispatcher.handler("create_user", role="admin")
def admin_create_user(params):
    return f"Admin creating user: {params.get('username')}"

@dispatcher.handler("create_user", role="manager")
def manager_create_user(params):
    return f"Manager creating user: {params.get('username')}"

# Define context class
class RequestContext:
    def __init__(self, role, environment):
        self.role = role
        self.environment = environment

# Dispatch actions based on context
admin_context = RequestContext("admin", "production")
result = dispatcher.dispatch(admin_context, "create_user", username="john")
print(result)  # "Admin creating user: john"

Advanced Usage

Multi-dimensional Routing

dispatcher = ActionDispatcher(['role', 'environment', 'feature_flag'])

@dispatcher.handler("process_payment",
                   role="admin",
                   environment="production",
                   feature_flag="new_payment_system")
def new_payment_handler(params):
    return "Processing with new payment system"

@dispatcher.handler("process_payment",
                   role="admin",
                   environment="production")
def default_payment_handler(params):
    return "Processing with default system"

Global Handlers

# Global handlers work across all contexts
@dispatcher.global_handler("health_check")
def health_check(params):
    return {"status": "healthy", "timestamp": time.time()}

# Global handlers have priority over scoped handlers
result = dispatcher.dispatch(any_context, "health_check")

Programmatic Registration

def custom_handler(params):
    return "Custom response"

# Register without decorators
dispatcher.register("custom_action", custom_handler, role="user")

# Register global handler
dispatcher.register_global("system_status", lambda p: "OK")

Error Handling

from action_dispatch import HandlerNotFoundError, InvalidDimensionError

try:
    result = dispatcher.dispatch(context, "unknown_action")
except HandlerNotFoundError as e:
    print(f"No handler found for action: {e.action}")
    print(f"Context rules: {e.rules}")

try:
    dispatcher.register("action", handler, invalid_dimension="value")
except InvalidDimensionError as e:
    print(f"Invalid dimension: {e.dimension}")
    print(f"Available dimensions: {e.available_dimensions}")

Caching for Performance

Enable LRU caching to improve performance when you have many dimensions or high-frequency lookups with repeated action + scope combinations.

# Enable cache on initialization
dispatcher = ActionDispatcher(
    dimensions=['region', 'platform', 'version', 'tier'],
    enable_cache=True,
    cache_maxsize=512
)

@dispatcher.handler("get_data", region="asia", platform="mobile")
def get_data_asia_mobile(params):
    return "data for asia mobile"

# First lookup - cache miss
result = dispatcher.get_handler("get_data", region="asia", platform="mobile")

# Second lookup - cache hit (faster)
result = dispatcher.get_handler("get_data", region="asia", platform="mobile")

# Check cache statistics
info = dispatcher.cache_info()
print(f"Cache hits: {info['hits']}, misses: {info['misses']}")
# Output: Cache hits: 1, misses: 1

Runtime Cache Control

# Create dispatcher without cache
dispatcher = ActionDispatcher(dimensions=['platform'])

# Enable cache later
dispatcher.enable_cache(maxsize=256)
print(dispatcher.is_cache_enabled)  # True

# Clear cache manually (useful after bulk handler registration)
dispatcher.clear_cache()

# Disable cache when no longer needed
dispatcher.disable_cache()
print(dispatcher.is_cache_enabled)  # False

When to Use Caching

Scenario	Recommendation
2-5 dimensions	Cache optional
5-10 dimensions	Consider enabling cache
10+ dimensions	Recommend enabling cache
High QPS (>10K)	Recommend enabling cache
Dynamic handler registration	Cache auto-invalidates on registration

Real-World Examples

Web API with Role-Based Access Control

from action_dispatch import ActionDispatcher

# Set up dispatcher for API routing
api_dispatcher = ActionDispatcher(['role', 'api_version'])

@api_dispatcher.handler("get_users", role="admin", api_version="v2")
def get_users_admin_v2(params):
    return {
        "users": get_all_users(),
        "total": count_all_users(),
        "permissions": ["read", "write", "delete"]
    }

@api_dispatcher.handler("get_users", role="user", api_version="v2")
def get_users_regular_v2(params):
    return {
        "users": get_user_own_data(params['context_object']),
        "permissions": ["read"]
    }

# API endpoint
def api_get_users(request):
    try:
        result = api_dispatcher.dispatch(
            request.user,
            "get_users",
            request_id=request.id
        )
        return JsonResponse(result)
    except HandlerNotFoundError:
        return JsonResponse({"error": "Forbidden"}, status=403)

Microservices Environment Routing

service_dispatcher = ActionDispatcher(['environment', 'service_version'])

@service_dispatcher.handler("process_order",
                           environment="production",
                           service_version="v2")
def process_order_prod_v2(params):
    # Use production database and new algorithm
    return production_order_processor.process(params['order_data'])

@service_dispatcher.handler("process_order",
                           environment="staging")
def process_order_staging(params):
    # Use staging database with verbose logging
    return staging_order_processor.process(params['order_data'])

Plugin System

plugin_dispatcher = ActionDispatcher(['plugin_type', 'version'])

# Plugins can register their handlers
@plugin_dispatcher.handler("transform_data",
                          plugin_type="image_processor",
                          version="2.0")
def image_transform_v2(params):
    return enhanced_image_transform(params['data'])

# Dynamic plugin loading
for plugin in load_plugins():
    plugin.register_handlers(plugin_dispatcher)

# Process with appropriate plugin
result = plugin_dispatcher.dispatch(context, "transform_data", data=input_data)

API Reference

ActionDispatcher

__init__(dimensions=None, enable_cache=False, cache_maxsize=256)

Create a new dispatcher with optional dimensions and caching.

• dimensions (list, optional): List of dimension names for routing
• enable_cache (bool, optional): Enable LRU cache for handler lookups (default: False)
• cache_maxsize (int, optional): Maximum size of the LRU cache (default: 256)

@handler(action, **kwargs)

Decorator to register a handler for specific action and dimensions.

• action (str): Action name
• **kwargs: Dimension values for routing

@global_handler(action)

Decorator to register a global handler that works across all contexts.

• action (str): Action name

register(action, handler, **kwargs)

Programmatically register a handler.

• action (str): Action name
• handler (callable): Handler function
• **kwargs: Dimension values for routing

dispatch(context_object, action_name, **kwargs)

Dispatch an action based on context.

• context_object: Object with dimension attributes
• action_name (str): Action to dispatch
• **kwargs: Additional parameters passed to handler

Cache Methods

enable_cache(maxsize=None)

Enable LRU cache for handler lookups at runtime.

• maxsize (int, optional): Maximum cache size (uses existing value if None)

disable_cache()

Disable cache and clear cached data.

clear_cache()

Clear all cached handler lookups.

cache_info()

Get cache statistics. Returns None if cache is disabled.

Returns a dict with:

• hits: Number of cache hits
• misses: Number of cache misses
• maxsize: Maximum cache size
• currsize: Current number of cached items

is_cache_enabled

Property that returns True if cache is currently enabled.

Exceptions

• ActionDispatchError: Base exception class
• InvalidDimensionError: Raised for invalid dimension parameters
• HandlerNotFoundError: Raised when no handler is found
• InvalidActionError: Raised for invalid action names

Development

Setting up Development Environment

# Clone the repository
git clone https://github.com/eowl/action-dispatch.git
cd action-dispatch

# Create virtual environment
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

# Install development dependencies
pip install -e ".[dev]"

# Install pre-commit hooks
pre-commit install

Running Tests

# Run all tests
python -m unittest discover tests -v

# Run with coverage
coverage run -m unittest discover tests
coverage report
coverage html

# Run specific test file
python -m unittest tests.test_action_dispatcher -v

# Run specific test class
python -m unittest tests.test_action_dispatcher.TestActionDispatcher -v

Code Quality

# Format code
black action_dispatch tests

# Lint code
flake8 action_dispatch tests

# Type checking
mypy action_dispatch

Contributing

We welcome contributions! Please see our Contributing Guide for details.

1. Fork the repository
2. Create a feature branch (git checkout -b feature/amazing-feature)
3. Make your changes
4. Add tests for your changes
5. Ensure all tests pass (python -m pytest)
6. Commit your changes (git commit -m 'Add amazing feature')
7. Push to the branch (git push origin feature/amazing-feature)
8. Open a Pull Request

License

This project is licensed under the MIT License - see the LICENSE file for details.

Changelog

See CHANGELOG.md for a list of changes and releases.

Acknowledgments

• Inspired by the need for flexible routing in complex applications
• Built with modern Python best practices
• Thoroughly tested and documented