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

Core JSON-RPC Protocol

Name: tessl/pypi-json-rpc
Author: tessl

Complete implementation of JSON-RPC 1.0 and 2.0 protocols providing automatic version detection, request parsing, response generation, and comprehensive error handling. The protocol layer is transport-agnostic, allowing integration with any communication mechanism.

Capabilities

Request Processing

The main entry point for processing JSON-RPC requests, automatically detecting protocol version and handling both single and batch requests.

class JSONRPCResponseManager:
    @classmethod
    def handle(cls, request_str: str, dispatcher: dict, context: dict = None):
        """
        Handle JSON-RPC request string and return response.
        
        Parameters:
        - request_str: JSON string containing the request
        - dispatcher: Dictionary mapping method names to callable functions
        - context: Optional context dictionary passed to methods that request it
        
        Returns:
        Response object (JSONRPC10Response, JSONRPC20Response, JSONRPC20BatchResponse, or None for notifications)
        """
    
    @classmethod
    def handle_request(cls, request, dispatcher: dict, context: dict = None):
        """
        Handle parsed request object directly.
        
        Parameters:
        - request: Parsed request object (JSONRPC10Request, JSONRPC20Request, or JSONRPC20BatchRequest)
        - dispatcher: Dictionary mapping method names to callable functions  
        - context: Optional context dictionary passed to methods that request it
        
        Returns:
        Response object or None for notifications
        """

Protocol Version Detection

Automatic detection between JSON-RPC 1.0 and 2.0 based on request format, with fallback handling for invalid requests.

class JSONRPCRequest:
    @classmethod
    def from_json(cls, json_str: str):
        """
        Parse JSON string and create appropriate request object.
        
        Parameters:
        - json_str: JSON string to parse
        
        Returns:
        JSONRPC10Request or JSONRPC20Request/JSONRPC20BatchRequest based on format
        
        Raises:
        JSONRPCInvalidRequestException: If request format is invalid
        """
    
    @classmethod
    def from_data(cls, data: dict):
        """
        Create request object from parsed data.
        
        Parameters:
        - data: Dictionary or list containing request data
        
        Returns:
        JSONRPC10Request or JSONRPC20Request/JSONRPC20BatchRequest
        """

Response Class Mapping

Internal mapping system that selects appropriate response classes based on protocol version.

# Response class mapping used internally
RESPONSE_CLASS_MAP = {
    "1.0": JSONRPC10Response,
    "2.0": JSONRPC20Response,
}

Usage Examples

Basic Request Handling

from jsonrpc import JSONRPCResponseManager, dispatcher

# Register methods
@dispatcher.add_method
def ping():
    return "pong"

@dispatcher.add_method  
def add(a, b):
    return a + b

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

# Handle JSON-RPC 1.0 request (no "jsonrpc" field)
request = '{"method": "ping", "params": [], "id": 1}'
response = JSONRPCResponseManager.handle(request, dispatcher)
print(response.json)
# {"result": "pong", "id": 1}

Batch Request Processing

# Handle batch request (JSON-RPC 2.0 only)
batch_request = '''[
    {"jsonrpc": "2.0", "method": "add", "params": [1, 2], "id": "1"},
    {"jsonrpc": "2.0", "method": "ping", "id": "2"},
    {"jsonrpc": "2.0", "method": "add", "params": [5, 7], "id": "3"}
]'''

response = JSONRPCResponseManager.handle(batch_request, dispatcher)
print(response.json)
# [
#   {"jsonrpc": "2.0", "result": 3, "id": "1"},
#   {"jsonrpc": "2.0", "result": "pong", "id": "2"}, 
#   {"jsonrpc": "2.0", "result": 12, "id": "3"}
# ]

Context Injection

from jsonrpc import Dispatcher

# Create dispatcher with context-aware method
dispatcher = Dispatcher()

@dispatcher.add_method(context_arg="ctx")
def get_user_info(user_id, ctx):
    # Context can contain request information, authentication, etc.
    request_id = ctx.get("request", {}).get("_id")
    return {"user_id": user_id, "processed_by_request": request_id}

# Handle request with context
context = {"user": "admin", "timestamp": "2023-01-01"}
request = '{"jsonrpc": "2.0", "method": "get_user_info", "params": [123], "id": 1}'
response = JSONRPCResponseManager.handle(request, dispatcher, context)

Error Handling

from jsonrpc.exceptions import JSONRPCDispatchException, JSONRPCMethodNotFound

@dispatcher.add_method
def divide(a, b):
    if b == 0:
        raise JSONRPCDispatchException(
            code=-32602, 
            message="Division by zero", 
            data={"dividend": a, "divisor": b}
        )
    return a / b

# Error responses are automatically formatted
request = '{"jsonrpc": "2.0", "method": "divide", "params": [10, 0], "id": 1}'
response = JSONRPCResponseManager.handle(request, dispatcher)
print(response.json)
# {"jsonrpc": "2.0", "error": {"code": -32602, "message": "Division by zero", "data": {"dividend": 10, "divisor": 0}}, "id": 1}

# Method not found
request = '{"jsonrpc": "2.0", "method": "nonexistent", "id": 1}'
response = JSONRPCResponseManager.handle(request, dispatcher)
print(response.json)
# {"jsonrpc": "2.0", "error": {"code": -32601, "message": "Method not found"}, "id": 1}

Install with Tessl CLI