Packages
snakepit
0.7.2
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/telemetry/backends/grpc.py
"""gRPC telemetry backend.
This backend sends telemetry events via the gRPC telemetry stream to Elixir.
"""
from __future__ import annotations
from typing import Any, Dict, Optional, TYPE_CHECKING
from .base import TelemetryBackend
if TYPE_CHECKING:
from ..stream import TelemetryStream
class GrpcBackend(TelemetryBackend):
"""Telemetry backend that uses gRPC telemetry stream.
This backend delegates to a TelemetryStream instance, which manages
the actual gRPC stream communication with Elixir.
Args:
stream: TelemetryStream instance to use for emission
"""
def __init__(self, stream: TelemetryStream) -> None:
"""Initialize the gRPC backend.
Args:
stream: TelemetryStream instance to use
"""
self._stream = stream
def emit(
self,
event_name: str,
measurements: Dict[str, Any],
metadata: Optional[Dict[str, Any]] = None,
correlation_id: Optional[str] = None,
) -> None:
"""Emit a telemetry event via gRPC stream.
Args:
event_name: Event name in dotted notation
measurements: Numeric measurements
metadata: Contextual metadata
correlation_id: Optional correlation ID
"""
self._stream.emit(event_name, measurements, metadata, correlation_id)
def close(self) -> None:
"""Close the gRPC stream gracefully."""
self._stream.close()
@property
def stream(self) -> TelemetryStream:
"""Get the underlying TelemetryStream instance.
Returns:
The TelemetryStream instance
"""
return self._stream