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

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, LLMInfoT 

11from langgate.core.logging import get_logger 

12from langgate.core.models import 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(BaseRegistryClient[LLMInfoT], Generic[LLMInfoT]): 

41 """ 

42 Base HTTP client for the Model Registry API. 

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

44 

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

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

47 stored in this instance is always used for requests. 

48 

49 Type Parameters: 

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

51 """ 

52 

53 __orig_bases__: tuple 

54 model_info_cls: type[LLMInfoT] 

55 

56 def __init__( 

57 self, 

58 base_url: str, 

59 api_key: SecretStr | None = None, 

60 cache_ttl: timedelta | None = None, 

61 model_info_cls: type[LLMInfoT] | None = None, 

62 http_client: httpx.AsyncClient | None = None, 

63 ): 

64 """Initialize the client. 

65 Args: 

66 base_url: The base URL of the registry service 

67 api_key: Registry server API key for authentication 

68 cache_ttl: Cache time-to-live 

69 model_info_cls: Override for model info class 

70 http_client: Optional HTTP client for making requests 

71 """ 

72 super().__init__(cache_ttl=cache_ttl) 

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

74 self.api_key = api_key 

75 self._http_client = http_client 

76 

77 # Set model_info_cls if provided, otherwise it is inferred from the class 

78 if model_info_cls is not None: 

79 self.model_info_cls = model_info_cls 

80 

81 logger.debug( 

82 "initialized_base_http_registry_client", 

83 base_url=self.base_url, 

84 api_key=self.api_key, 

85 model_info_cls=self.model_info_cls, 

86 ) 

87 

88 def __init_subclass__(cls, **kwargs): 

89 """Set up model class when this class is subclassed.""" 

90 super().__init_subclass__(**kwargs) 

91 

92 # Extract the model class from generic parameters 

93 if not hasattr(cls, "model_info_cls"): 

94 cls.model_info_cls = cls._get_model_info_class() 

95 

96 @classmethod 

97 def _get_model_info_class(cls) -> type[LLMInfoT]: 

98 """Extract the model class from generic type parameters.""" 

99 return get_args(cls.__orig_bases__[0])[0] 

100 

101 @asynccontextmanager 

102 async def _get_client_for_request(self): 

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

104 if self._http_client: 

105 yield self._http_client 

106 else: 

107 async with httpx.AsyncClient() as temp_client: 

108 yield temp_client 

109 

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

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

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

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

114 if self.api_key: 

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

116 

117 async with self._get_client_for_request() as client: 

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

119 return response 

120 

121 async def _fetch_model_info(self, model_id: str) -> LLMInfoT: 

122 """Fetch model info from the source via HTTP.""" 

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

124 response.raise_for_status() 

125 return self.model_info_cls.model_validate(response.json()) 

126 

127 async def _fetch_all_models(self) -> list[LLMInfoT]: 

128 """Fetch all models from the source via HTTP.""" 

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

130 response.raise_for_status() 

131 return [self.model_info_cls.model_validate(model) for model in response.json()] 

132 

133 

134class HTTPRegistryClient(BaseHTTPRegistryClient[LLMInfo]): 

135 """HTTP client singleton for the Model Registry API using the default LLMInfo schema.""" 

136 

137 _instance = None 

138 

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

140 if cls._instance is None: 

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

142 logger.debug("creating_http_registry_client_singleton") 

143 return cls._instance 

144 

145 def __init__( 

146 self, 

147 base_url: str, 

148 api_key: SecretStr | None = None, 

149 cache_ttl: timedelta | None = None, 

150 http_client: httpx.AsyncClient | None = None, 

151 ): 

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

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

154 self._initialized = True 

155 logger.debug("initialized_http_registry_client_singleton")