Spaces:
Running
Running
File size: 2,463 Bytes
0231daa |
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 |
"""
Dependency injection for FastAPI routes.
This module provides dependency functions that can be injected into
route handlers, ensuring consistent access to shared resources.
"""
from typing import Optional
from fastapi import Depends, HTTPException, status
from src.config.settings import Settings, get_settings
from src.core.manager import ModelManager
from src.core.cache import EmbeddingCache
# Global instances (initialized at startup)
_model_manager: Optional[ModelManager] = None
_embedding_cache: Optional[EmbeddingCache] = None
def set_model_manager(manager: ModelManager) -> None:
"""
Set the global model manager instance.
Called during application startup.
Args:
manager: ModelManager instance
"""
global _model_manager
_model_manager = manager
def set_embedding_cache(cache: EmbeddingCache) -> None:
"""
Set the global embedding cache instance.
Called during application startup if caching is enabled.
Args:
cache: EmbeddingCache instance
"""
global _embedding_cache
_embedding_cache = cache
def get_model_manager() -> ModelManager:
"""
Get the model manager instance.
This is a dependency function that can be used in route handlers.
Returns:
ModelManager instance
Raises:
HTTPException: If model manager is not initialized
Example:
@router.post("/embed")
async def embed(
request: EmbedRequest,
manager: ModelManager = Depends(get_model_manager)
):
model = manager.get_model(request.model_id)
...
"""
if _model_manager is None:
raise HTTPException(
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
detail="Server is not ready. Model manager not initialized.",
)
return _model_manager
def get_embedding_cache() -> Optional[EmbeddingCache]:
"""
Get the embedding cache instance (if enabled).
Returns:
EmbeddingCache instance or None if caching is disabled
"""
return _embedding_cache
def get_cache_if_enabled(
settings: Settings = Depends(get_settings),
) -> Optional[EmbeddingCache]:
"""
Get cache only if caching is enabled in settings.
Args:
settings: Application settings
Returns:
EmbeddingCache instance if enabled, None otherwise
"""
if settings.ENABLE_CACHE:
return _embedding_cache
return None
|