Phase 3 (Part 1): API Infrastructure - FastAPI, Database, Redis, gRPC Client
Completed Tasks (T027-T032): - ✅ FastAPI application with structured logging, CORS, global error handlers - ✅ Pydantic Settings for environment configuration - ✅ SQLAlchemy async engine with session management - ✅ Alembic migration environment setup - ✅ Redis async client with connection pooling - ✅ gRPC SDK Bridge client (placeholder - awaiting protobuf generation) Next: JWT utilities, middleware, database models 🤖 Generated with Claude Code
This commit is contained in:
Geutebruck API Developer
126
src/api/clients/redis_client.py
Normal file
126
src/api/clients/redis_client.py
Normal file
@@ -0,0 +1,126 @@
|
||||
"""
|
||||
Redis client with connection pooling
|
||||
"""
|
||||
import redis.asyncio as redis
|
||||
from typing import Optional, Any
|
||||
import json
|
||||
import structlog
|
||||
from config import settings
|
||||
|
||||
logger = structlog.get_logger()
|
||||
|
||||
class RedisClient:
|
||||
"""Async Redis client wrapper"""
|
||||
|
||||
def __init__(self):
|
||||
self._pool: Optional[redis.ConnectionPool] = None
|
||||
self._client: Optional[redis.Redis] = None
|
||||
|
||||
async def connect(self):
|
||||
"""Initialize Redis connection pool"""
|
||||
try:
|
||||
logger.info("redis_connecting", host=settings.REDIS_HOST, port=settings.REDIS_PORT)
|
||||
|
||||
self._pool = redis.ConnectionPool.from_url(
|
||||
settings.redis_url,
|
||||
max_connections=settings.REDIS_MAX_CONNECTIONS,
|
||||
decode_responses=True,
|
||||
)
|
||||
|
||||
self._client = redis.Redis(connection_pool=self._pool)
|
||||
|
||||
# Test connection
|
||||
await self._client.ping()
|
||||
|
||||
logger.info("redis_connected")
|
||||
except Exception as e:
|
||||
logger.error("redis_connection_failed", error=str(e))
|
||||
raise
|
||||
|
||||
async def close(self):
|
||||
"""Close Redis connections"""
|
||||
try:
|
||||
if self._client:
|
||||
await self._client.close()
|
||||
if self._pool:
|
||||
await self._pool.disconnect()
|
||||
logger.info("redis_closed")
|
||||
except Exception as e:
|
||||
logger.error("redis_close_failed", error=str(e))
|
||||
|
||||
async def get(self, key: str) -> Optional[str]:
|
||||
"""Get value by key"""
|
||||
if not self._client:
|
||||
raise RuntimeError("Redis client not connected")
|
||||
return await self._client.get(key)
|
||||
|
||||
async def set(self, key: str, value: Any, expire: Optional[int] = None) -> bool:
|
||||
"""Set value with optional expiration (seconds)"""
|
||||
if not self._client:
|
||||
raise RuntimeError("Redis client not connected")
|
||||
return await self._client.set(key, value, ex=expire)
|
||||
|
||||
async def delete(self, key: str) -> int:
|
||||
"""Delete key"""
|
||||
if not self._client:
|
||||
raise RuntimeError("Redis client not connected")
|
||||
return await self._client.delete(key)
|
||||
|
||||
async def exists(self, key: str) -> bool:
|
||||
"""Check if key exists"""
|
||||
if not self._client:
|
||||
raise RuntimeError("Redis client not connected")
|
||||
return await self._client.exists(key) > 0
|
||||
|
||||
async def get_json(self, key: str) -> Optional[dict]:
|
||||
"""Get JSON value"""
|
||||
value = await self.get(key)
|
||||
if value:
|
||||
return json.loads(value)
|
||||
return None
|
||||
|
||||
async def set_json(self, key: str, value: dict, expire: Optional[int] = None) -> bool:
|
||||
"""Set JSON value"""
|
||||
return await self.set(key, json.dumps(value), expire)
|
||||
|
||||
async def get_many(self, keys: list[str]) -> list[Optional[str]]:
|
||||
"""Get multiple values"""
|
||||
if not self._client:
|
||||
raise RuntimeError("Redis client not connected")
|
||||
return await self._client.mget(keys)
|
||||
|
||||
async def set_many(self, mapping: dict[str, Any]) -> bool:
|
||||
"""Set multiple key-value pairs"""
|
||||
if not self._client:
|
||||
raise RuntimeError("Redis client not connected")
|
||||
return await self._client.mset(mapping)
|
||||
|
||||
async def incr(self, key: str, amount: int = 1) -> int:
|
||||
"""Increment value"""
|
||||
if not self._client:
|
||||
raise RuntimeError("Redis client not connected")
|
||||
return await self._client.incrby(key, amount)
|
||||
|
||||
async def expire(self, key: str, seconds: int) -> bool:
|
||||
"""Set expiration on key"""
|
||||
if not self._client:
|
||||
raise RuntimeError("Redis client not connected")
|
||||
return await self._client.expire(key, seconds)
|
||||
|
||||
async def ttl(self, key: str) -> int:
|
||||
"""Get time to live for key"""
|
||||
if not self._client:
|
||||
raise RuntimeError("Redis client not connected")
|
||||
return await self._client.ttl(key)
|
||||
|
||||
# Global Redis client instance
|
||||
redis_client = RedisClient()
|
||||
|
||||
# Convenience functions
|
||||
async def init_redis():
|
||||
"""Initialize Redis connection (call on startup)"""
|
||||
await redis_client.connect()
|
||||
|
||||
async def close_redis():
|
||||
"""Close Redis connection (call on shutdown)"""
|
||||
await redis_client.close()
|
||||
222
src/api/clients/sdk_bridge_client.py
Normal file
222
src/api/clients/sdk_bridge_client.py
Normal file
@@ -0,0 +1,222 @@
|
||||
"""
|
||||
gRPC client for SDK Bridge communication
|
||||
"""
|
||||
import grpc
|
||||
from typing import Optional, List
|
||||
import structlog
|
||||
from config import settings
|
||||
|
||||
# TODO: Import generated protobuf classes after running protoc
|
||||
# from protos import camera_pb2, camera_pb2_grpc
|
||||
# from protos import monitor_pb2, monitor_pb2_grpc
|
||||
# from protos import crossswitch_pb2, crossswitch_pb2_grpc
|
||||
|
||||
logger = structlog.get_logger()
|
||||
|
||||
class SDKBridgeClient:
|
||||
"""gRPC client for communicating with SDK Bridge"""
|
||||
|
||||
def __init__(self):
|
||||
self._channel: Optional[grpc.aio.Channel] = None
|
||||
# TODO: Initialize stubs after protobuf generation
|
||||
# self._camera_stub = None
|
||||
# self._monitor_stub = None
|
||||
# self._crossswitch_stub = None
|
||||
|
||||
async def connect(self):
|
||||
"""Initialize gRPC channel to SDK Bridge"""
|
||||
try:
|
||||
logger.info("sdk_bridge_connecting", url=settings.sdk_bridge_url)
|
||||
|
||||
# Create async gRPC channel
|
||||
self._channel = grpc.aio.insecure_channel(
|
||||
settings.sdk_bridge_url,
|
||||
options=[
|
||||
('grpc.max_send_message_length', 50 * 1024 * 1024), # 50MB
|
||||
('grpc.max_receive_message_length', 50 * 1024 * 1024), # 50MB
|
||||
('grpc.keepalive_time_ms', 30000), # 30 seconds
|
||||
('grpc.keepalive_timeout_ms', 10000), # 10 seconds
|
||||
]
|
||||
)
|
||||
|
||||
# TODO: Initialize service stubs after protobuf generation
|
||||
# self._camera_stub = camera_pb2_grpc.CameraServiceStub(self._channel)
|
||||
# self._monitor_stub = monitor_pb2_grpc.MonitorServiceStub(self._channel)
|
||||
# self._crossswitch_stub = crossswitch_pb2_grpc.CrossSwitchServiceStub(self._channel)
|
||||
|
||||
# Test connection with health check
|
||||
# await self.health_check()
|
||||
|
||||
logger.info("sdk_bridge_connected")
|
||||
except Exception as e:
|
||||
logger.error("sdk_bridge_connection_failed", error=str(e))
|
||||
raise
|
||||
|
||||
async def close(self):
|
||||
"""Close gRPC channel"""
|
||||
try:
|
||||
if self._channel:
|
||||
await self._channel.close()
|
||||
logger.info("sdk_bridge_closed")
|
||||
except Exception as e:
|
||||
logger.error("sdk_bridge_close_failed", error=str(e))
|
||||
|
||||
async def health_check(self) -> dict:
|
||||
"""Check SDK Bridge health"""
|
||||
try:
|
||||
logger.debug("sdk_bridge_health_check")
|
||||
# TODO: Implement after protobuf generation
|
||||
# request = crossswitch_pb2.Empty()
|
||||
# response = await self._crossswitch_stub.HealthCheck(request, timeout=5.0)
|
||||
# return {
|
||||
# "is_healthy": response.is_healthy,
|
||||
# "sdk_status": response.sdk_status,
|
||||
# "geviserver_host": response.geviserver_host
|
||||
# }
|
||||
return {"is_healthy": True, "sdk_status": "connected", "geviserver_host": "localhost"}
|
||||
except grpc.RpcError as e:
|
||||
logger.error("sdk_bridge_health_check_failed", error=str(e))
|
||||
return {"is_healthy": False, "sdk_status": "error", "error": str(e)}
|
||||
|
||||
async def list_cameras(self) -> List[dict]:
|
||||
"""List all cameras from GeViServer"""
|
||||
try:
|
||||
logger.debug("sdk_bridge_list_cameras")
|
||||
# TODO: Implement after protobuf generation
|
||||
# request = camera_pb2.ListCamerasRequest()
|
||||
# response = await self._camera_stub.ListCameras(request, timeout=10.0)
|
||||
# return [
|
||||
# {
|
||||
# "id": camera.id,
|
||||
# "name": camera.name,
|
||||
# "description": camera.description,
|
||||
# "has_ptz": camera.has_ptz,
|
||||
# "has_video_sensor": camera.has_video_sensor,
|
||||
# "status": camera.status
|
||||
# }
|
||||
# for camera in response.cameras
|
||||
# ]
|
||||
return [] # Placeholder
|
||||
except grpc.RpcError as e:
|
||||
logger.error("sdk_bridge_list_cameras_failed", error=str(e))
|
||||
raise
|
||||
|
||||
async def get_camera(self, camera_id: int) -> Optional[dict]:
|
||||
"""Get camera details"""
|
||||
try:
|
||||
logger.debug("sdk_bridge_get_camera", camera_id=camera_id)
|
||||
# TODO: Implement after protobuf generation
|
||||
# request = camera_pb2.GetCameraRequest(camera_id=camera_id)
|
||||
# response = await self._camera_stub.GetCamera(request, timeout=5.0)
|
||||
# return {
|
||||
# "id": response.id,
|
||||
# "name": response.name,
|
||||
# "description": response.description,
|
||||
# "has_ptz": response.has_ptz,
|
||||
# "has_video_sensor": response.has_video_sensor,
|
||||
# "status": response.status
|
||||
# }
|
||||
return None # Placeholder
|
||||
except grpc.RpcError as e:
|
||||
if e.code() == grpc.StatusCode.NOT_FOUND:
|
||||
return None
|
||||
logger.error("sdk_bridge_get_camera_failed", camera_id=camera_id, error=str(e))
|
||||
raise
|
||||
|
||||
async def list_monitors(self) -> List[dict]:
|
||||
"""List all monitors from GeViServer"""
|
||||
try:
|
||||
logger.debug("sdk_bridge_list_monitors")
|
||||
# TODO: Implement after protobuf generation
|
||||
# request = monitor_pb2.ListMonitorsRequest()
|
||||
# response = await self._monitor_stub.ListMonitors(request, timeout=10.0)
|
||||
# return [
|
||||
# {
|
||||
# "id": monitor.id,
|
||||
# "name": monitor.name,
|
||||
# "description": monitor.description,
|
||||
# "is_active": monitor.is_active,
|
||||
# "current_camera_id": monitor.current_camera_id,
|
||||
# "status": monitor.status
|
||||
# }
|
||||
# for monitor in response.monitors
|
||||
# ]
|
||||
return [] # Placeholder
|
||||
except grpc.RpcError as e:
|
||||
logger.error("sdk_bridge_list_monitors_failed", error=str(e))
|
||||
raise
|
||||
|
||||
async def execute_crossswitch(self, camera_id: int, monitor_id: int, mode: int = 0) -> dict:
|
||||
"""Execute cross-switch operation"""
|
||||
try:
|
||||
logger.info("sdk_bridge_crossswitch", camera_id=camera_id, monitor_id=monitor_id, mode=mode)
|
||||
# TODO: Implement after protobuf generation
|
||||
# request = crossswitch_pb2.CrossSwitchRequest(
|
||||
# camera_id=camera_id,
|
||||
# monitor_id=monitor_id,
|
||||
# mode=mode
|
||||
# )
|
||||
# response = await self._crossswitch_stub.ExecuteCrossSwitch(request, timeout=10.0)
|
||||
# return {
|
||||
# "success": response.success,
|
||||
# "message": response.message,
|
||||
# "camera_id": response.camera_id,
|
||||
# "monitor_id": response.monitor_id
|
||||
# }
|
||||
return {"success": True, "message": "Placeholder", "camera_id": camera_id, "monitor_id": monitor_id}
|
||||
except grpc.RpcError as e:
|
||||
logger.error("sdk_bridge_crossswitch_failed", error=str(e))
|
||||
raise
|
||||
|
||||
async def clear_monitor(self, monitor_id: int) -> dict:
|
||||
"""Clear monitor (stop video)"""
|
||||
try:
|
||||
logger.info("sdk_bridge_clear_monitor", monitor_id=monitor_id)
|
||||
# TODO: Implement after protobuf generation
|
||||
# request = crossswitch_pb2.ClearMonitorRequest(monitor_id=monitor_id)
|
||||
# response = await self._crossswitch_stub.ClearMonitor(request, timeout=10.0)
|
||||
# return {
|
||||
# "success": response.success,
|
||||
# "message": response.message,
|
||||
# "monitor_id": response.monitor_id
|
||||
# }
|
||||
return {"success": True, "message": "Placeholder", "monitor_id": monitor_id}
|
||||
except grpc.RpcError as e:
|
||||
logger.error("sdk_bridge_clear_monitor_failed", error=str(e))
|
||||
raise
|
||||
|
||||
async def get_routing_state(self) -> dict:
|
||||
"""Get current routing state"""
|
||||
try:
|
||||
logger.debug("sdk_bridge_get_routing_state")
|
||||
# TODO: Implement after protobuf generation
|
||||
# request = crossswitch_pb2.GetRoutingStateRequest()
|
||||
# response = await self._crossswitch_stub.GetRoutingState(request, timeout=10.0)
|
||||
# return {
|
||||
# "routes": [
|
||||
# {
|
||||
# "camera_id": route.camera_id,
|
||||
# "monitor_id": route.monitor_id,
|
||||
# "camera_name": route.camera_name,
|
||||
# "monitor_name": route.monitor_name
|
||||
# }
|
||||
# for route in response.routes
|
||||
# ],
|
||||
# "total_routes": response.total_routes
|
||||
# }
|
||||
return {"routes": [], "total_routes": 0} # Placeholder
|
||||
except grpc.RpcError as e:
|
||||
logger.error("sdk_bridge_get_routing_state_failed", error=str(e))
|
||||
raise
|
||||
|
||||
# Global SDK Bridge client instance
|
||||
sdk_bridge_client = SDKBridgeClient()
|
||||
|
||||
# Convenience functions
|
||||
async def init_sdk_bridge():
|
||||
"""Initialize SDK Bridge connection (call on startup)"""
|
||||
await sdk_bridge_client.connect()
|
||||
|
||||
async def close_sdk_bridge():
|
||||
"""Close SDK Bridge connection (call on shutdown)"""
|
||||
await sdk_bridge_client.close()
|
||||
Reference in New Issue
Block a user