Packages
snakepit
0.4.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/enhanced.py
"""
Enhanced bridge adapter that will evolve through the stages.
"""
from typing import Any, Dict, Optional
import logging
logger = logging.getLogger(__name__)
class EnhancedBridge:
"""
Enhanced bridge adapter with session support.
This is a minimal implementation for Stage 0.
It will be significantly expanded in subsequent stages.
"""
def __init__(self):
self.session_context = None
self.initialized = False
def set_session_context(self, session_context):
"""Set the session context for this adapter instance."""
self.session_context = session_context
if hasattr(session_context, 'session_id'):
logger.info(f"Session context set: {session_context.session_id}")
async def initialize(self):
"""Initialize the adapter with any necessary setup."""
# Future: Register built-in tools
# Future: Set up framework integrations
self.initialized = True
if self.session_context and hasattr(self.session_context, 'session_id'):
logger.info(f"Adapter initialized for session: {self.session_context.session_id}")
else:
logger.info("Adapter initialized")
async def cleanup(self):
"""Clean up adapter resources."""
if self.session_context:
logger.info(f"Cleaning up adapter for session: {self.session_context.session_id}")
self.initialized = False
def get_info(self) -> Dict[str, Any]:
"""Get adapter information."""
return {
"adapter": "EnhancedBridge",
"version": "0.1.0",
"stage": 0,
"initialized": self.initialized,
"session_id": self.session_context.session_id if self.session_context else None,
"features": {
"variables": False, # Stage 1
"tools": False, # Stage 2
"streaming": False, # Stage 3
"optimization": False # Stage 4
}
}