Spaces:
Sleeping
Sleeping
File size: 3,859 Bytes
27cde0c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 | """Dependency injection utilities for API routes."""
from typing import Annotated, Generator
from fastapi import Depends, HTTPException, status
from app.config import Settings, get_settings
from app.core.env import WebScraperEnv
from app.memory.manager import MemoryManager
from app.models.router import SmartModelRouter
from app.tools.registry import MCPToolRegistry
# Settings dependency
SettingsDep = Annotated[Settings, Depends(get_settings)]
# Store for active environments
_active_environments: dict[str, WebScraperEnv] = {}
def get_environment(episode_id: str) -> WebScraperEnv:
"""Get an active environment by episode ID."""
if episode_id not in _active_environments:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Episode {episode_id} not found",
)
return _active_environments[episode_id]
def create_environment(episode_id: str, settings: Settings) -> WebScraperEnv:
"""Create a new environment for an episode."""
if episode_id in _active_environments:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail=f"Episode {episode_id} already exists",
)
env = WebScraperEnv(episode_id=episode_id, settings=settings)
_active_environments[episode_id] = env
return env
def remove_environment(episode_id: str) -> bool:
"""Remove an environment from active storage."""
if episode_id in _active_environments:
del _active_environments[episode_id]
return True
return False
def list_environments() -> list[str]:
"""List all active episode IDs."""
return list(_active_environments.keys())
class DependencyContainer:
"""Container for dependency injection across the application."""
def __init__(self) -> None:
self._memory_manager: MemoryManager | None = None
self._model_router: SmartModelRouter | None = None
self._tool_registry: MCPToolRegistry | None = None
def set_memory_manager(self, manager: MemoryManager) -> None:
"""Set the memory manager instance."""
self._memory_manager = manager
def set_model_router(self, router: SmartModelRouter) -> None:
"""Set the model router instance."""
self._model_router = router
def set_tool_registry(self, registry: MCPToolRegistry) -> None:
"""Set the tool registry instance."""
self._tool_registry = registry
def get_memory_manager(self) -> MemoryManager:
"""Get the memory manager instance."""
if self._memory_manager is None:
raise RuntimeError("Memory manager not initialized")
return self._memory_manager
def get_model_router(self) -> SmartModelRouter:
"""Get the model router instance."""
if self._model_router is None:
raise RuntimeError("Model router not initialized")
return self._model_router
def get_tool_registry(self) -> MCPToolRegistry:
"""Get the tool registry instance."""
if self._tool_registry is None:
raise RuntimeError("Tool registry not initialized")
return self._tool_registry
# Global container instance
container = DependencyContainer()
def get_memory_manager() -> MemoryManager:
"""Dependency for memory manager."""
return container.get_memory_manager()
def get_model_router() -> SmartModelRouter:
"""Dependency for model router."""
return container.get_model_router()
def get_tool_registry() -> MCPToolRegistry:
"""Dependency for tool registry."""
return container.get_tool_registry()
# Type aliases for dependency injection
MemoryManagerDep = Annotated[MemoryManager, Depends(get_memory_manager)]
ModelRouterDep = Annotated[SmartModelRouter, Depends(get_model_router)]
ToolRegistryDep = Annotated[MCPToolRegistry, Depends(get_tool_registry)]
|