Coverage for src/bayernwerk_client/map/client.py: 98%

1from __future__ import annotations 

2 

3from collections.abc import Callable, Iterator 

4from pathlib import Path 

5from typing import Any, Self 

6 

7import httpx 

8 

9from bayernwerk_client.exceptions import ApiError, AuthenticationError 

10from bayernwerk_client.map.instances import iter_instance_items 

11from bayernwerk_client.tokens import TokenSet, TokenStore 

12 

13DEFAULT_BASE_URL = "https://icon-api.eon.com" 

14DEFAULT_TENANT = "BAG" 

15"""Tenant code for Bayernwerk (as sent by the SPA itself).""" 

16DEFAULT_LANG = "de" 

17 

18MAP_TOKEN_PATH = Path.home() / ".cache" / "bayernwerk-client" / "map-tokens.json" 

19"""Distinct from efix's token path - two services must not share a cache file.""" 

20 

21TokenRefresher = Callable[[TokenSet], TokenSet] 

22 

23 

24class MapClient: 

25 """REST client for icon-api.eon.com (the Mein.Auftragsportal backend). 

26 

27 Plain httpx underneath - no browser involved here. Obtain the initial 

28 `TokenSet` via `bayernwerk_client.map.auth.login_interactive` (or by loading 

29 one previously saved with `TokenStore`), and optionally pass 

30 `on_token_expired` so the client can transparently renew an expired 

31 access token instead of raising. In practice that callback should just 

32 re-run `login_interactive` - see the module docstring in `auth.py` for 

33 why the captured `refresh_token` can't be used for a silent httpx-only 

34 refresh here. 

35 """ 

36 

37 def __init__( 

38 self, 

39 tokens: TokenSet, 

40 *, 

41 base_url: str = DEFAULT_BASE_URL, 

42 tenant: str = DEFAULT_TENANT, 

43 lang: str = DEFAULT_LANG, 

44 on_token_expired: TokenRefresher | None = None, 

45 token_store: TokenStore | None = None, 

46 timeout: float = 30.0, 

47 ) -> None: 

48 self._tokens = tokens 

49 self._on_token_expired = on_token_expired 

50 self._token_store = token_store 

51 self._default_params = {"tenant": tenant, "lang": lang} 

52 self._http = httpx.Client(base_url=base_url, timeout=timeout) 

53 

54 @classmethod 

55 def from_token_store(cls, token_store: TokenStore, **kwargs: Any) -> MapClient: 

56 tokens = token_store.load() 

57 if tokens is None: 

58 raise AuthenticationError(f"No cached tokens found at {token_store.path}") 

59 return cls(tokens, token_store=token_store, **kwargs) 

60 

61 def close(self) -> None: 

62 self._http.close() 

63 

64 def __enter__(self) -> Self: 

65 return self 

66 

67 def __exit__(self, *exc_info: object) -> None: 

68 self.close() 

69 

70 @property 

71 def tokens(self) -> TokenSet: 

72 return self._tokens 

73 

74 def _renew(self) -> None: 

75 if self._on_token_expired is None: 

76 raise AuthenticationError("Access token expired/rejected and no on_token_expired callback was configured") 

77 self._tokens = self._on_token_expired(self._tokens) 

78 if self._token_store is not None: 

79 self._token_store.save(self._tokens) 

80 

81 def _send(self, method: str, path: str, **kwargs: Any) -> httpx.Response: 

82 if self._tokens.is_expired: 

83 self._renew() 

84 

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

86 headers["Authorization"] = f"Bearer {self._tokens.access_token}" 

87 params = {**self._default_params, **kwargs.pop("params", {})} 

88 response = self._http.request(method, path, headers=headers, params=params, **kwargs) 

89 

90 if response.status_code == 401 and self._on_token_expired is not None: 

91 self._renew() 

92 headers["Authorization"] = f"Bearer {self._tokens.access_token}" 

93 response = self._http.request(method, path, headers=headers, params=params, **kwargs) 

94 

95 if response.is_error: 

96 raise ApiError(response.status_code, response.text) 

97 return response 

98 

99 def request(self, method: str, path: str, **kwargs: Any) -> Any: 

100 response = self._send(method, path, **kwargs) 

101 if not response.content: 

102 return None 

103 return response.json() 

104 

105 def get(self, path: str, **kwargs: Any) -> Any: 

106 return self.request("GET", path, **kwargs) 

107 

108 # --- Orders ----------------------------------------------------------- 

109 def list_orders(self, **params: Any) -> Any: 

110 return self.get("/icon-installer-space-srv/api/orders", params=params) 

111 

112 def get_order(self, order_id: str) -> Any: 

113 return self.get(f"/icon-installer-space-srv/api/orders/{order_id}") 

114 

115 def get_order_details(self, order_id: str) -> Any: 

116 return self.get(f"/icon-installer-space-srv/api/orders/{order_id}/order-details") 

117 

118 def get_order_instances(self, order_id: str) -> Any: 

119 return self.get(f"/icon-installer-space-srv/api/orders/{order_id}/instances") 

120 

121 def iter_order_instance_items(self, order_id: str) -> Iterator[dict[str, Any]]: 

122 """`get_order_instances`, flattened via `iter_instance_items` - see 

123 `instances.py` for why this walks the tree schema-less rather than 

124 via a typed model.""" 

125 return iter_instance_items(self.get_order_instances(order_id)) 

126 

127 def get_order_documents(self, order_id: str) -> Any: 

128 return self.get(f"/icon-installer-space-srv/api/orders/{order_id}/documents") 

129 

130 def get_order_notes(self, order_id: str) -> Any: 

131 return self.get(f"/icon-installer-space-srv/api/orders/{order_id}/notes") 

132 

133 def download_order_document(self, order_id: str, document: dict[str, Any]) -> bytes: 

134 """Download the raw bytes of one entry from `get_order_documents(order_id)`. 

135 

136 `document` must be one of the dicts returned by `get_order_documents` - 

137 it needs `id` and `storageLocation` from there (the backend 400s 

138 without `storageLocation`, e.g. `AXON_IVY`). The response has no 

139 useful filename header; use `document["filename"]` for that. 

140 """ 

141 response = self._send( 

142 "GET", 

143 f"/icon-installer-space-srv/api/orders/{order_id}/documents/{document['id']}/download", 

144 params={"storageLocation": document["storageLocation"]}, 

145 ) 

146 return response.content 

147 

148 def sync_order_documents(self, order_id: str, folder: Path | str) -> list[str]: 

149 """Download whichever of the portal's documents for `order_id` aren't already 

150 present (matched by filename) in `folder`. Existing files are left untouched. 

151 

152 Returns the filenames that were newly downloaded. 

153 """ 

154 folder = Path(folder) 

155 folder.mkdir(parents=True, exist_ok=True) 

156 existing = {p.name for p in folder.iterdir() if p.is_file()} 

157 

158 downloaded = [] 

159 for document in self.get_order_documents(order_id): 

160 filename = document["filename"] 

161 if filename in existing: 

162 continue 

163 content = self.download_order_document(order_id, document) 

164 (folder / filename).write_bytes(content) 

165 downloaded.append(filename) 

166 return downloaded 

167 

168 # --- Installers / hardware master data ---------------------------------- 

169 def list_installers(self, **params: Any) -> Any: 

170 return self.get("/icon-installer-space-srv/api/installers", params=params) 

171 

172 def list_inverters(self, *, primary_energy_forms: list[str], **params: Any) -> Any: 

173 """`primary_energy_forms` is required by the backend (e.g. `["PV"]`).""" 

174 return self.get( 

175 "/icon-installer-space-srv/api/harnes/inverters", 

176 params={"primaryEnergyForms": primary_energy_forms, **params}, 

177 ) 

178 

179 def list_storages(self, **params: Any) -> Any: 

180 return self.get("/icon-installer-space-srv/api/harnes/storages", params=params) 

181 

182 # --- Product orders ----------------------------------------------------- 

183 def list_product_orders(self, **params: Any) -> Any: 

184 return self.get("/icon-installer-srv/api/product-orders", params=params)