JSON-RPC transport implementation for Python supporting both 1.0 and 2.0 protocols with Django and Flask backends
—
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.
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 mappingMultiple 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")
"""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
"""Pre-created dispatcher instance available for immediate use.
# Global dispatcher instance
dispatcher: Dispatcherfrom 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}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}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}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}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}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"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