Coverage for packages/client/src/langgate/client/http.py: 46%

1"""HTTP client for LangGate API.""" 

2 

3from contextlib import asynccontextmanager 

4from datetime import timedelta 

5from typing import Generic, get_args 

6 

7import httpx 

8from pydantic import SecretStr 

9 

10from langgate.client.protocol import BaseRegistryClient, ImageInfoT, LLMInfoT 

11from langgate.core.logging import get_logger 

12from langgate.core.models import ImageModelInfo, LLMInfo 

13 

14logger = get_logger(__name__) 

15 

16 

17def create_registry_http_client( 

18 base_url: str, 

19 api_key: SecretStr | None = None, 

20 timeout: float | httpx.Timeout | None = 10.0, 

21 **kwargs, 

22) -> httpx.AsyncClient: 

23 """ 

24 Creates and configures an httpx.AsyncClient for the Registry API. 

25 """ 

26 headers = kwargs.pop("headers", {}) 

27 if api_key: 

28 headers["X-API-Key"] = api_key.get_secret_value() 

29 

30 processed_base_url = base_url.rstrip("/") 

31 

32 return httpx.AsyncClient( 

33 base_url=processed_base_url, 

34 headers=headers, 

35 timeout=timeout, 

36 **kwargs, 

37 ) 

38 

39 

40class BaseHTTPRegistryClient( 

41 BaseRegistryClient[LLMInfoT, ImageInfoT], Generic[LLMInfoT, ImageInfoT] 

42): 

43 """ 

44 Base HTTP client for the Model Registry API. 

45 Supports LLMInfo-derived and ImageModelInfo-derived schemas for response parsing and httpx client injection. 

46 

47 Handles infrequent HTTP requests via temporary clients by default or uses an 

48 injected client as the request engine. Configuration (base_url, api_key) 

49 stored in this instance is always used for requests. 

50 

51 Type Parameters: 

52 LLMInfoT: The LLMInfo-derived model class for response parsing 

53 ImageInfoT: The ImageModelInfo-derived model class for response parsing 

54 """ 

55 

56 __orig_bases__: tuple 

57 llm_info_cls: type[LLMInfoT] 

58 image_info_cls: type[ImageInfoT] 

59 

60 def __init__( 

61 self, 

62 base_url: str, 

63 api_key: SecretStr | None = None, 

64 cache_ttl: timedelta | None = None, 

65 llm_info_cls: type[LLMInfoT] | None = None, 

66 image_info_cls: type[ImageInfoT] | None = None, 

67 http_client: httpx.AsyncClient | None = None, 

68 ): 

69 """Initialize the client. 

70 Args: 

71 base_url: The base URL of the registry service 

72 api_key: Registry server API key for authentication 

73 cache_ttl: Cache time-to-live 

74 llm_info_cls: Override for LLM info class 

75 image_info_cls: Override for image model info class 

76 http_client: Optional HTTP client for making requests 

77 """ 

78 super().__init__(cache_ttl=cache_ttl) 

79 self.base_url = base_url.rstrip("/") 

80 self.api_key = api_key 

81 self._http_client = http_client 

82 

83 # Set model info classes if provided, otherwise they are inferred from the class 

84 if llm_info_cls is not None: 

85 self.llm_info_cls = llm_info_cls 

86 if image_info_cls is not None: 

87 self.image_info_cls = image_info_cls 

88 

89 logger.debug( 

90 "initialized_base_http_registry_client", 

91 base_url=self.base_url, 

92 api_key=self.api_key, 

93 llm_info_cls=getattr(self, "llm_info_cls", None), 

94 image_info_cls=getattr(self, "image_info_cls", None), 

95 ) 

96 

97 def __init_subclass__(cls, **kwargs): 

98 """Set up model classes when this class is subclassed.""" 

99 super().__init_subclass__(**kwargs) 

100 

101 # Extract the model classes from generic parameters 

102 if not hasattr(cls, "llm_info_cls") or not hasattr(cls, "image_info_cls"): 

103 llm_cls, image_cls = cls._get_model_info_classes() 

104 if not hasattr(cls, "llm_info_cls"): 

105 cls.llm_info_cls = llm_cls 

106 if not hasattr(cls, "image_info_cls"): 

107 cls.image_info_cls = image_cls 

108 

109 @classmethod 

110 def _get_model_info_classes(cls) -> tuple[type[LLMInfoT], type[ImageInfoT]]: 

111 """Extract the model classes from generic type parameters.""" 

112 args = get_args(cls.__orig_bases__[0]) 

113 return args[0], args[1] 

114 

115 @asynccontextmanager 

116 async def _get_client_for_request(self): 

117 """Provides the httpx client to use (injected or temporary).""" 

118 if self._http_client: 

119 yield self._http_client 

120 else: 

121 async with httpx.AsyncClient() as temp_client: 

122 yield temp_client 

123 

124 async def _request(self, method: str, url_path: str, **kwargs) -> httpx.Response: 

125 """Makes an HTTP request using the appropriate client engine.""" 

126 url = f"{self.base_url}{url_path}" 

127 headers = kwargs.pop("headers", {}) 

128 if self.api_key: 

129 headers["X-API-Key"] = self.api_key.get_secret_value() 

130 

131 async with self._get_client_for_request() as client: 

132 response = await client.request(method, url, headers=headers, **kwargs) 

133 return response 

134 

135 async def _fetch_llm_info(self, model_id: str) -> LLMInfoT: 

136 """Fetch LLM info from the source via HTTP.""" 

137 response = await self._request("GET", f"/models/llms/{model_id}") 

138 response.raise_for_status() 

139 return self.llm_info_cls.model_validate(response.json()) 

140 

141 async def _fetch_image_model_info(self, model_id: str) -> ImageInfoT: 

142 """Fetch image model info from the source via HTTP.""" 

143 response = await self._request("GET", f"/models/images/{model_id}") 

144 response.raise_for_status() 

145 return self.image_info_cls.model_validate(response.json()) 

146 

147 async def _fetch_all_llms(self) -> list[LLMInfoT]: 

148 """Fetch all LLMs from the source via HTTP.""" 

149 response = await self._request("GET", "/models/llms") 

150 response.raise_for_status() 

151 return [self.llm_info_cls.model_validate(model) for model in response.json()] 

152 

153 async def _fetch_all_image_models(self) -> list[ImageInfoT]: 

154 """Fetch all image models from the source via HTTP.""" 

155 response = await self._request("GET", "/models/images") 

156 response.raise_for_status() 

157 return [self.image_info_cls.model_validate(model) for model in response.json()] 

158 

159 

160class HTTPRegistryClient(BaseHTTPRegistryClient[LLMInfo, ImageModelInfo]): 

161 """HTTP client singleton for the Model Registry API using the default schemas.""" 

162 

163 _instance = None 

164 

165 def __new__(cls, *args, **kwargs): 

166 if cls._instance is None: 

167 cls._instance = super().__new__(cls) 

168 logger.debug("creating_http_registry_client_singleton") 

169 return cls._instance 

170 

171 def __init__( 

172 self, 

173 base_url: str, 

174 api_key: SecretStr | None = None, 

175 cache_ttl: timedelta | None = None, 

176 http_client: httpx.AsyncClient | None = None, 

177 ): 

178 if not hasattr(self, "_initialized"): 

179 super().__init__(base_url, api_key, cache_ttl, http_client=http_client) 

180 self._initialized = True 

181 logger.debug("initialized_http_registry_client_singleton")