CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-json-rpc

JSON-RPC transport implementation for Python supporting both 1.0 and 2.0 protocols with Django and Flask backends

Pending
Overview
Eval results
Files

dispatcher.mddocs/

Method Dispatching

Flexible method registration and dispatch system supporting functions, classes, objects, decorators, and context injection. The Dispatcher class provides a dictionary-like interface for mapping method names to callable functions with extensive customization options.

Capabilities

Core Dispatcher

Dictionary-like object that maps method names to callable functions, implementing the MutableMapping interface for standard dictionary operations.

class Dispatcher:
    def __init__(self, prototype = None):
        """
        Initialize dispatcher with optional prototype.
        
        Parameters:
        - prototype: Initial method mapping (dict or object with callable attributes)
        """
    
    # Dictionary interface
    def __getitem__(self, key: str):
        """Get method by name."""
    
    def __setitem__(self, key: str, value):
        """Set method by name."""
    
    def __delitem__(self, key: str):
        """Remove method by name."""
    
    def __len__(self) -> int:
        """Number of registered methods."""
    
    def __iter__(self):
        """Iterate over method names."""
    
    def __repr__(self) -> str:
        """String representation of method mapping."""
    
    # Properties
    method_map: dict  # Internal method storage
    context_arg_for_method: dict  # Context argument mapping

Method Registration

Multiple ways to register methods with flexible naming and context injection support.

def add_method(self, f = None, name: str = None, context_arg: str = None):
    """
    Add a method to the dispatcher.
    
    Parameters:
    - f: Callable to register
    - name: Method name (defaults to function name)
    - context_arg: Parameter name that will receive context data
    
    Returns:
    The original function (for decorator usage)
    
    Usage:
    - As method: dispatcher.add_method(func, "name")
    - As decorator: @dispatcher.add_method
    - With custom name: @dispatcher.add_method(name="custom")
    - With context: @dispatcher.add_method(context_arg="ctx")
    """

Bulk Registration

Register multiple methods from classes, objects, or dictionaries with optional prefixing.

def add_class(self, cls):
    """
    Add all public methods from a class.
    
    Parameters:
    - cls: Class to register methods from
    
    Methods are prefixed with lowercase class name + '.'
    """

def add_object(self, obj):
    """
    Add all public methods from an object instance.
    
    Parameters:
    - obj: Object instance to register methods from
    
    Methods are prefixed with lowercase class name + '.'
    """

def add_dict(self, dict: dict, prefix: str = ''):
    """
    Add methods from dictionary.
    
    Parameters:
    - dict: Dictionary of name -> callable mappings
    - prefix: Optional prefix for method names
    """

def build_method_map(self, prototype, prefix: str = ''):
    """
    Build method map from prototype object or dictionary.
    
    Parameters:
    - prototype: Object or dict to extract methods from
    - prefix: Optional prefix for method names
    """

Global Dispatcher Instance

Pre-created dispatcher instance available for immediate use.

# Global dispatcher instance
dispatcher: Dispatcher

Usage Examples

Basic Method Registration

from jsonrpc import dispatcher, JSONRPCResponseManager

# Method 1: Using decorator
@dispatcher.add_method
def add(a, b):
    """Add two numbers."""
    return a + b

@dispatcher.add_method
def multiply(x, y):
    """Multiply two numbers."""
    return x * y

# Method 2: Direct registration
def subtract(a, b):
    return a - b

dispatcher.add_method(subtract)

# Method 3: Dictionary-style assignment
dispatcher["divide"] = lambda a, b: a / b

# Test the methods
request = '{"jsonrpc": "2.0", "method": "add", "params": [5, 3], "id": 1}'
response = JSONRPCResponseManager.handle(request, dispatcher)
print(response.json)
# {"jsonrpc": "2.0", "result": 8, "id": 1}

Custom Method Names

from jsonrpc import Dispatcher

dispatcher = Dispatcher()

# Custom name with decorator
@dispatcher.add_method(name="math.add")
def addition(a, b):
    return a + b

# Custom name with method call
def multiplication(a, b):
    return a * b

dispatcher.add_method(multiplication, name="math.multiply")

# Dictionary-style with custom name
dispatcher["math.power"] = lambda base, exp: base ** exp

# Usage
request = '{"jsonrpc": "2.0", "method": "math.add", "params": [2, 3], "id": 1}'
response = JSONRPCResponseManager.handle(request, dispatcher)
print(response.json)
# {"jsonrpc": "2.0", "result": 5, "id": 1}

Context Injection

from jsonrpc import Dispatcher, JSONRPCResponseManager

dispatcher = Dispatcher()

# Method that receives context
@dispatcher.add_method(context_arg="context")
def get_request_info(context):
    """Return information about the current request."""
    request = context.get("request")
    return {
        "request_id": request._id if request else None,
        "method": request.method if request else None,
        "user": context.get("user", "anonymous")
    }

@dispatcher.add_method(context_arg="ctx")
def calculate_with_audit(a, b, operation, ctx):
    """Perform calculation with audit logging."""
    result = a + b if operation == "add" else a * b
    
    # Access context data
    user = ctx.get("user", "unknown")
    timestamp = ctx.get("timestamp")
    
    return {
        "result": result,
        "calculated_by": user,
        "timestamp": timestamp
    }

# Handle request with context
context = {
    "user": "alice",
    "timestamp": "2023-01-01T12:00:00Z",
    "session_id": "abc123"
}

request = '{"jsonrpc": "2.0", "method": "get_request_info", "id": 1}'
response = JSONRPCResponseManager.handle(request, dispatcher, context)
print(response.json)
# {"jsonrpc": "2.0", "result": {"request_id": 1, "method": "get_request_info", "user": "alice"}, "id": 1}

Class and Object Registration

from jsonrpc import Dispatcher

# Example service class
class MathService:
    def add(self, a, b):
        return a + b
    
    def subtract(self, a, b):
        return a - b
    
    def _private_method(self):
        # Private methods (starting with _) are not registered
        return "private"

class StringService:
    def __init__(self, prefix=""):
        self.prefix = prefix
    
    def upper(self, text):
        return (self.prefix + text).upper()
    
    def lower(self, text):
        return (self.prefix + text).lower()

dispatcher = Dispatcher()

# Register all public methods from class (creates new instance)
dispatcher.add_class(MathService)

# Register methods from specific object instance
string_service = StringService("Hello ")
dispatcher.add_object(string_service)

# Methods are prefixed with class name
print(list(dispatcher.method_map.keys()))
# ['mathservice.add', 'mathservice.subtract', 'stringservice.upper', 'stringservice.lower']

# Usage
request = '{"jsonrpc": "2.0", "method": "mathservice.add", "params": [10, 5], "id": 1}'
response = JSONRPCResponseManager.handle(request, dispatcher)
print(response.json)
# {"jsonrpc": "2.0", "result": 15, "id": 1}

Dictionary Registration

from jsonrpc import Dispatcher

# Create method dictionary
math_methods = {
    "add": lambda a, b: a + b,
    "multiply": lambda a, b: a * b,
    "power": lambda base, exp: base ** exp
}

string_methods = {
    "upper": str.upper,
    "lower": str.lower,
    "capitalize": str.capitalize
}

dispatcher = Dispatcher()

# Register with prefix
dispatcher.add_dict(math_methods, prefix="math")
dispatcher.add_dict(string_methods, prefix="string")

print(list(dispatcher.method_map.keys()))
# ['math.add', 'math.multiply', 'math.power', 'string.upper', 'string.lower', 'string.capitalize']

# Usage
request = '{"jsonrpc": "2.0", "method": "string.upper", "params": ["hello world"], "id": 1}'
response = JSONRPCResponseManager.handle(request, dispatcher)
print(response.json)
# {"jsonrpc": "2.0", "result": "HELLO WORLD", "id": 1}

Prototype Initialization

from jsonrpc import Dispatcher

# Initialize with function dictionary
initial_methods = {
    "ping": lambda: "pong",
    "echo": lambda msg: msg,
    "add": lambda a, b: a + b
}

dispatcher = Dispatcher(prototype=initial_methods)

# Initialize with object
class APIService:
    def status(self):
        return "running"
    
    def version(self):
        return "1.0.0"

api_dispatcher = Dispatcher(prototype=APIService())

print(list(api_dispatcher.method_map.keys()))
# ['status', 'version']

# Add more methods later
@api_dispatcher.add_method
def shutdown():
    return "shutting down"

Method Removal and Management

from jsonrpc import Dispatcher

dispatcher = Dispatcher()

# Add methods
dispatcher["add"] = lambda a, b: a + b
dispatcher["subtract"] = lambda a, b: a - b
dispatcher["multiply"] = lambda a, b: a * b

print(len(dispatcher))  # 3
print("add" in dispatcher)  # True

# Remove method
del dispatcher["subtract"]
print(len(dispatcher))  # 2

# Clear all methods
dispatcher.method_map.clear()
print(len(dispatcher))  # 0

# Check available methods
@dispatcher.add_method
def test():
    return "test"

print(list(dispatcher))  # ['test']
print(repr(dispatcher))  # {'test': <function test at 0x...>}

Install with Tessl CLI

npx tessl i tessl/pypi-json-rpc

docs

backends.md

core-jsonrpc.md

dispatcher.md

exceptions.md

index.md

requests-responses.md

tile.json