Packages
snakepit
0.6.3
0.13.0
0.12.0
0.11.1
0.11.0
0.10.1
0.10.0
0.9.1
0.9.0
0.8.9
0.8.8
0.8.7
0.8.6
0.8.5
0.8.4
0.8.3
0.8.2
0.8.1
0.8.0
0.7.7
0.7.6
0.7.5
0.7.4
0.7.3
0.7.2
0.7.1
0.7.0
0.6.11
0.6.10
0.6.9
0.6.8
0.6.7
0.6.6
0.6.5
0.6.4
0.6.3
0.6.2
0.6.1
0.6.0
0.5.1
0.5.0
0.4.3
0.4.2
0.4.1
0.4.0
0.3.3
0.3.2
0.3.1
0.3.0
0.2.1
0.2.0
0.1.2
0.1.1
0.1.0
High-performance pooler and session manager for external language integrations. Supports Python, Node.js, Ruby, and more with gRPC streaming, session management, and production-ready process cleanup.
Current section
Files
Jump to
Current section
Files
priv/python/snakepit_bridge/adapters/showcase/handlers/basic_ops.py
"""Basic operations handler for showcase adapter."""
import time
from typing import Dict, Any
from datetime import datetime
from ..tool import Tool
class BasicOpsHandler:
"""Handler for basic operations like ping, echo, and error demonstrations."""
def get_tools(self) -> Dict[str, Tool]:
"""Return all tools provided by this handler."""
return {
"ping": Tool(self.ping),
"echo": Tool(self.echo),
"add": Tool(self.add),
"error_demo": Tool(self.error_demo),
"adapter_info": Tool(self.adapter_info),
}
def ping(self, ctx, message: str = "pong") -> Dict[str, str]:
"""Simple ping operation."""
return {"message": message, "timestamp": str(time.time())}
def echo(self, ctx, **kwargs) -> Dict[str, Any]:
"""Echo back all provided arguments."""
return {"echoed": kwargs}
def add(self, ctx, a: float, b: float) -> float:
"""Add two numbers together."""
return a + b
def error_demo(self, ctx, error_type: str = "generic") -> None:
"""Demonstrate error handling with different error types."""
if error_type == "value":
raise ValueError("This is a demonstration ValueError")
elif error_type == "runtime":
raise RuntimeError("This is a demonstration RuntimeError")
else:
raise Exception("This is a generic exception")
def adapter_info(self, ctx) -> Dict[str, Any]:
"""Return information about the adapter capabilities."""
return {
"adapter_name": "ShowcaseAdapter",
"version": "2.0.0", # Updated version for refactored adapter
"capabilities": [
"binary_serialization",
"streaming",
"ml_workflows",
"session_state_via_elixir"
],
"handlers": [
"BasicOpsHandler",
"SessionOpsHandler",
"BinaryOpsHandler",
"StreamingOpsHandler",
"ConcurrentOpsHandler",
"MLWorkflowHandler"
]
}