Packages
snakepit
0.3.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/cli/generic.py
#!/usr/bin/env python3
"""
Generic Snakepit Bridge CLI
Console script entry point for the generic bridge.
This provides the same functionality as generic_bridge_v2.py but
as an installed console script.
"""
import sys
import os
from ..adapters.generic import GenericCommandHandler
from ..core import ProtocolHandler, setup_graceful_shutdown, setup_broken_pipe_suppression
def main():
"""Main entry point for the generic bridge console script."""
# Suppress broken pipe errors globally
setup_broken_pipe_suppression()
if len(sys.argv) > 1 and sys.argv[1] == "--help":
print("Generic Snakepit Bridge V2 (Console Script)")
print("Usage: snakepit-generic-bridge [--mode pool-worker]")
print("")
print("This bridge provides an extensible architecture for creating custom adapters.")
print("Installed as a console script from the snakepit-bridge package.")
print("")
print("Default supported commands:")
handler = GenericCommandHandler()
for cmd in sorted(handler.get_supported_commands()):
print(f" {cmd}")
return
# Create protocol handler with generic command handler
command_handler = GenericCommandHandler()
protocol_handler = ProtocolHandler(command_handler)
# Set up graceful shutdown handling
setup_graceful_shutdown(protocol_handler)
# Start protocol handler
try:
protocol_handler.run()
except (KeyboardInterrupt, BrokenPipeError, IOError):
# Clean shutdown, suppress errors
os._exit(0)
except Exception as e:
from ..core import safe_print
safe_print(f"Bridge error: {e}")
sys.exit(1)
if __name__ == "__main__":
main()