Packages
snakepit
0.6.1
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/session_ops.py
"""Session operations handler for showcase adapter."""
import time
import os
from datetime import datetime
from typing import Dict, Any
from ..tool import Tool
class SessionOpsHandler:
"""Handler for session management operations.
Note: In a production system, state would be stored via SessionContext
variables. For this showcase, we use a simple in-memory approach to
demonstrate the concepts.
"""
# Temporary in-memory storage for demo purposes
# In production, use ctx.register_variable() or external storage
_session_data = {}
def get_tools(self) -> Dict[str, Tool]:
"""Return all tools provided by this handler."""
return {
"init_session": Tool(self.init_session),
"cleanup": Tool(self.cleanup_session),
"set_counter": Tool(self.set_counter),
"get_counter": Tool(self.get_counter),
"increment_counter": Tool(self.increment_counter),
"get_worker_info": Tool(self.get_worker_info)
}
def init_session(self, ctx, **kwargs) -> Dict[str, Any]:
"""Initialize session state."""
session_id = ctx.session_id
# Store initial state
self._session_data[session_id] = {
'session_start_time': time.time(),
'command_count': 0,
'counter': 0
}
return {
"timestamp": datetime.now().isoformat(),
"worker_pid": os.getpid(),
"session_id": session_id
}
def cleanup_session(self, ctx) -> Dict[str, Any]:
"""Clean up session, demonstrating state retrieval."""
session_id = ctx.session_id
session_data = self._session_data.get(session_id, {})
start_time = session_data.get('session_start_time', time.time())
duration_ms = (time.time() - start_time) * 1000
command_count = session_data.get('command_count', 0)
# Cleanup
if session_id in self._session_data:
del self._session_data[session_id]
return {
"duration_ms": round(duration_ms, 2),
"command_count": command_count
}
def set_counter(self, ctx, value: int) -> Dict[str, Any]:
"""Set counter value."""
session_id = ctx.session_id
if session_id not in self._session_data:
self._session_data[session_id] = {'command_count': 0, 'counter': 0}
self._session_data[session_id]['counter'] = value
self._session_data[session_id]['command_count'] += 1
return {"value": value}
def get_counter(self, ctx) -> Dict[str, Any]:
"""Get counter value."""
session_id = ctx.session_id
session_data = self._session_data.get(session_id, {'command_count': 0, 'counter': 0})
# Increment command count
if session_id in self._session_data:
self._session_data[session_id]['command_count'] += 1
return {"value": session_data.get('counter', 0)}
def increment_counter(self, ctx) -> Dict[str, Any]:
"""Demonstrate counter increment."""
session_id = ctx.session_id
if session_id not in self._session_data:
self._session_data[session_id] = {'command_count': 0, 'counter': 0}
# Increment both counters
self._session_data[session_id]['counter'] += 1
self._session_data[session_id]['command_count'] += 1
return {"value": self._session_data[session_id]['counter']}
def get_worker_info(self, ctx, call_number: int) -> Dict[str, Any]:
"""Get worker info and update command count."""
session_id = ctx.session_id
if session_id in self._session_data:
self._session_data[session_id]['command_count'] += 1
return {
"worker_pid": str(os.getpid()),
"call_number": call_number,
"session_id": session_id
}