Coverage for packages/server/src/langgate/server/api/services/registry_api.py: 43%

1import structlog 

2from fastapi import HTTPException, status 

3 

4from langgate.registry import ModelRegistry 

5from langgate.registry.models import ImageModelInfo, LLMInfo 

6 

7logger = structlog.stdlib.get_logger(__name__) 

8 

9 

10class ModelRegistryAPI: 

11 """ 

12 API for accessing the model registry. 

13 This will be expanded when gateway becomes a microservice. 

14 """ 

15 

16 def __init__(self, registry: ModelRegistry | None = None): 

17 self.registry = registry or ModelRegistry() 

18 

19 async def get_llm_info(self, model_id: str) -> LLMInfo: 

20 """Get LLM information.""" 

21 try: 

22 return self.registry.get_llm_info(model_id) 

23 except ValueError as exc: 

24 if "not found" in str(exc): 

25 raise HTTPException( 

26 status_code=status.HTTP_404_NOT_FOUND, 

27 detail=f"Model {model_id} not found", 

28 ) from exc 

29 await logger.aexception("model_info_error", model_id=model_id) 

30 raise 

31 

32 async def list_llms(self) -> list[LLMInfo]: 

33 """List all available LLMs.""" 

34 return self.registry.list_llms() 

35 

36 async def get_image_model_info(self, model_id: str) -> ImageModelInfo: 

37 """Get image model information.""" 

38 try: 

39 return self.registry.get_image_model_info(model_id) 

40 except ValueError as exc: 

41 if "not found" in str(exc): 

42 raise HTTPException( 

43 status_code=status.HTTP_404_NOT_FOUND, 

44 detail=f"Model {model_id} not found", 

45 ) from exc 

46 await logger.aexception("model_info_error", model_id=model_id) 

47 raise 

48 

49 async def list_image_models(self) -> list[ImageModelInfo]: 

50 """List all available image models.""" 

51 return self.registry.list_image_models()