Coverage for src/lexigram/admin/data/adapters/api_adapter.py: 0%

1"""REST API data source adapter for Lexigram Admin.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING, Any, Generic, TypeVar 

6 

7try: 

8 import httpx 

9 

10 HAS_HTTPX = True 

11except ImportError: 

12 HAS_HTTPX = False 

13 

14from lexigram.admin.data.data_source import IDataSource, QueryResult 

15from lexigram.di.decorators import inject 

16 

17if TYPE_CHECKING: 

18 from lexigram.admin.data.query import QuerySpec 

19 

20T = TypeVar("T") 

21 

22 

23@inject 

24class APIDataSource(IDataSource[T], Generic[T]): 

25 """Data source adapter for REST API endpoints. 

26 

27 This adapter communicates with a backend API using HTTP requests, 

28 translating the Query object into URL parameters. 

29 """ 

30 

31 def __init__( 

32 self, 

33 base_url: str, 

34 *, 

35 client: httpx.AsyncClient | None = None, 

36 timeout: float = 10.0, 

37 headers: dict[str, str] | None = None, 

38 ) -> None: 

39 """Initialize with API configuration. 

40 

41 Args: 

42 base_url: The base URL of the API endpoint. 

43 client: Optional httpx.AsyncClient instance. 

44 timeout: Default timeout for requests in seconds. 

45 headers: Optional default headers for requests. 

46 

47 Raises: 

48 ImportError: If 'httpx' is not installed. 

49 """ 

50 if not HAS_HTTPX: 

51 raise ImportError( 

52 "httpx is required for APIDataSource. Install with: pip install httpx", 

53 ) 

54 

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

56 self._client = client 

57 self.timeout = timeout 

58 self.headers = headers or {} 

59 

60 async def _get_client(self) -> httpx.AsyncClient: 

61 """Get or create an async HTTP client.""" 

62 if self._client is None: 

63 self._client = httpx.AsyncClient(timeout=self.timeout, headers=self.headers) 

64 return self._client 

65 

66 async def find_one(self, item_id: Any) -> T | None: 

67 """Fetch a single entity by its resource ID.""" 

68 client = await self._get_client() 

69 url = f"{self.base_url}/{item_id}" 

70 

71 response = await client.get(url) 

72 if response.status_code == 404: 

73 return None 

74 

75 response.raise_for_status() 

76 return response.json() 

77 

78 async def find_many(self, query: QuerySpec) -> QueryResult[T]: 

79 """Fetch multiple entities using query parameters.""" 

80 client = await self._get_client() 

81 params = self._transform_query(query) 

82 

83 response = await client.get(self.base_url, params=params) 

84 response.raise_for_status() 

85 

86 data = response.json() 

87 

88 # We assume the API returns a standard structure or a list 

89 if isinstance(data, list): 

90 items = data 

91 return QueryResult( 

92 items=items, 

93 total=len(items), 

94 page=query.page, 

95 per_page=query.per_page, 

96 ) 

97 

98 # If it's an object, we try to extract common pagination fields 

99 items = data.get("items", []) 

100 total = data.get("total", len(items)) 

101 

102 return QueryResult( 

103 items=items, 

104 total=total, 

105 page=data.get("page", query.page), 

106 per_page=data.get("per_page", query.per_page), 

107 has_next=data.get("has_next", False), 

108 has_prev=data.get("has_prev", False), 

109 cursor=data.get("cursor"), 

110 ) 

111 

112 async def count(self, query: QuerySpec) -> int: 

113 """Count matching entities (often requires a separate endpoint or HEAD).""" 

114 client = await self._get_client() 

115 params = self._transform_query(query) 

116 params["count_only"] = "true" 

117 

118 response = await client.get(f"{self.base_url}/count", params=params) 

119 if response.status_code == 404: 

120 # Fallback to find_many and extract total 

121 res = await self.find_many(query) 

122 return res.total 

123 

124 response.raise_for_status() 

125 return response.json().get("count", 0) 

126 

127 async def create(self, data: dict[str, Any]) -> T: 

128 """Create a new entity via POST.""" 

129 client = await self._get_client() 

130 response = await client.post(self.base_url, json=data) 

131 response.raise_for_status() 

132 return response.json() 

133 

134 async def update(self, item_id: Any, data: dict[str, Any]) -> T: 

135 """Update an entity via PATCH or PUT.""" 

136 client = await self._get_client() 

137 url = f"{self.base_url}/{item_id}" 

138 response = await client.patch(url, json=data) 

139 response.raise_for_status() 

140 return response.json() 

141 

142 async def delete(self, item_id: Any) -> bool: 

143 """Delete an entity via DELETE.""" 

144 client = await self._get_client() 

145 url = f"{self.base_url}/{item_id}" 

146 response = await client.delete(url) 

147 return response.status_code in (200, 204) 

148 

149 async def bulk_create(self, items: list[dict[str, Any]]) -> list[T]: 

150 """Bulk create via batch POST.""" 

151 client = await self._get_client() 

152 url = f"{self.base_url}/bulk" 

153 response = await client.post(url, json={"items": items}) 

154 response.raise_for_status() 

155 return response.json().get("items", []) 

156 

157 async def bulk_update(self, ids: list[Any], data: dict[str, Any]) -> int: 

158 """Bulk update via batch PATCH.""" 

159 client = await self._get_client() 

160 url = f"{self.base_url}/bulk" 

161 response = await client.patch(url, json={"ids": ids, "data": data}) 

162 response.raise_for_status() 

163 return response.json().get("updated_count", 0) 

164 

165 async def bulk_delete(self, ids: list[Any]) -> int: 

166 """Bulk delete via batch DELETE.""" 

167 client = await self._get_client() 

168 url = f"{self.base_url}/bulk" 

169 response = await client.request("DELETE", url, json={"ids": ids}) 

170 response.raise_for_status() 

171 return response.json().get("deleted_count", 0) 

172 

173 def _transform_query(self, query: QuerySpec) -> dict[str, Any]: 

174 """Translate Query object to URL parameters.""" 

175 params: dict[str, Any] = { 

176 "page": query.page, 

177 "per_page": query.per_page, 

178 } 

179 

180 if query.sort_by: 

181 params["sort_by"] = query.sort_by 

182 params["sort_order"] = query.sort_order 

183 

184 if query.search: 

185 params["search"] = query.search 

186 if query.search_fields: 

187 params["search_fields"] = ",".join(query.search_fields) 

188 

189 if query.select_fields: 

190 params["select"] = ",".join(query.select_fields) 

191 

192 if query.include: 

193 params["include"] = ",".join(query.include) 

194 

195 if query.cursor: 

196 params["cursor"] = query.cursor 

197 

198 # Transform filters 

199 for condition in query.filter_conditions: 

200 key = f"filter[{condition.field}][{condition.operator.value}]" 

201 params[key] = condition.value 

202 

203 return params