zzstoatzz.io
This repository has no description
make cast_to generic
type[T] (T bound to BaseModel) with overloads on _parse and the four
transport methods: cast_to given returns T, omitted returns Any for the
raw-json escape hatch. resource annotations are now checked against
their cast_to instead of laundered through Any. an empty body where
cast_to was given now raises SembleError instead of silently returning
None — that's a broken api contract, not a type to accommodate
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
author
zzstoatzz
co-author
Claude Fable 5
date
(Jun 10, 2026, 11:39 PM -0500)
commit
755ed98d
755ed98debcd34057d7ec84da319bacca83accd8
parent
291e6e22
291e6e22eb98bbb1a0af1ea48fceadafac8fdd04
+87
-22
+64
-22
src/semble/_client.py
+64
-22
src/semble/_client.py
···
1
1
from types import TracebackType
2
-
from typing import Any
2
+
from typing import Any, TypeVar, overload
3
3
4
4
import httpx2 as httpx
5
-
from pydantic import SecretStr
5
+
from pydantic import BaseModel, SecretStr
6
6
7
-
from semble._exceptions import status_error
7
+
from semble._exceptions import SembleError, status_error
8
8
from semble.resources.actors import Actors, AsyncActors
9
9
from semble.resources.cards import AsyncCards, Cards
10
10
from semble.resources.collections import AsyncCollections, Collections
···
15
15
from semble.resources.search import AsyncSearch, Search
16
16
from semble.settings import SembleSettings
17
17
18
+
T = TypeVar("T", bound=BaseModel)
19
+
20
+
21
+
@overload
22
+
def _parse(response: httpx.Response, cast_to: type[T]) -> T: ...
23
+
@overload
24
+
def _parse(response: httpx.Response, cast_to: None) -> Any: ...
25
+
def _parse(response: httpx.Response, cast_to: type[T] | None) -> Any:
26
+
if not response.is_success:
27
+
raise status_error(response)
28
+
if not response.content:
29
+
if cast_to is not None:
30
+
raise SembleError(
31
+
f"expected a json body from {response.request.url}, got an empty response"
32
+
)
33
+
return None
34
+
data = response.json()
35
+
if cast_to is None:
36
+
return data
37
+
return cast_to.model_validate(data)
38
+
18
39
19
40
class _BaseClient:
20
41
def __init__(
···
41
62
if self.api_key is not None:
42
63
headers["x-api-key"] = self.api_key.get_secret_value()
43
64
return headers
44
-
45
-
@staticmethod
46
-
def _parse(response: httpx.Response, cast_to: Any) -> Any:
47
-
if not response.is_success:
48
-
raise status_error(response)
49
-
if not response.content:
50
-
return None
51
-
data = response.json()
52
-
if cast_to is None:
53
-
return data
54
-
return cast_to.model_validate(data)
55
65
56
66
57
67
class Semble(_BaseClient):
···
88
98
self.notifications = Notifications(self)
89
99
self.search = Search(self)
90
100
101
+
@overload
102
+
def get(
103
+
self, nsid: str, params: dict[str, Any] | None = None, *, cast_to: type[T]
104
+
) -> T: ...
105
+
@overload
106
+
def get(
107
+
self, nsid: str, params: dict[str, Any] | None = None, *, cast_to: None = None
108
+
) -> Any: ...
91
109
def get(
92
110
self,
93
111
nsid: str,
94
112
params: dict[str, Any] | None = None,
95
113
*,
96
-
cast_to: Any = None,
114
+
cast_to: type[T] | None = None,
97
115
) -> Any:
98
116
"""GET an xrpc query by nsid. escape hatch for unwrapped endpoints."""
99
117
response = self._http.get(
100
118
self._url(nsid), params=params, headers=self._headers()
101
119
)
102
-
return self._parse(response, cast_to)
120
+
return _parse(response, cast_to)
103
121
122
+
@overload
123
+
def post(
124
+
self, nsid: str, json: dict[str, Any] | None = None, *, cast_to: type[T]
125
+
) -> T: ...
126
+
@overload
127
+
def post(
128
+
self, nsid: str, json: dict[str, Any] | None = None, *, cast_to: None = None
129
+
) -> Any: ...
104
130
def post(
105
131
self,
106
132
nsid: str,
107
133
json: dict[str, Any] | None = None,
108
134
*,
109
-
cast_to: Any = None,
135
+
cast_to: type[T] | None = None,
110
136
) -> Any:
111
137
"""POST an xrpc procedure by nsid. escape hatch for unwrapped endpoints."""
112
138
response = self._http.post(self._url(nsid), json=json, headers=self._headers())
113
-
return self._parse(response, cast_to)
139
+
return _parse(response, cast_to)
114
140
115
141
def close(self) -> None:
116
142
if self._owns_http:
···
162
188
self.notifications = AsyncNotifications(self)
163
189
self.search = AsyncSearch(self)
164
190
191
+
@overload
192
+
async def get(
193
+
self, nsid: str, params: dict[str, Any] | None = None, *, cast_to: type[T]
194
+
) -> T: ...
195
+
@overload
196
+
async def get(
197
+
self, nsid: str, params: dict[str, Any] | None = None, *, cast_to: None = None
198
+
) -> Any: ...
165
199
async def get(
166
200
self,
167
201
nsid: str,
168
202
params: dict[str, Any] | None = None,
169
203
*,
170
-
cast_to: Any = None,
204
+
cast_to: type[T] | None = None,
171
205
) -> Any:
172
206
"""GET an xrpc query by nsid. escape hatch for unwrapped endpoints."""
173
207
response = await self._http.get(
174
208
self._url(nsid), params=params, headers=self._headers()
175
209
)
176
-
return self._parse(response, cast_to)
210
+
return _parse(response, cast_to)
177
211
212
+
@overload
213
+
async def post(
214
+
self, nsid: str, json: dict[str, Any] | None = None, *, cast_to: type[T]
215
+
) -> T: ...
216
+
@overload
217
+
async def post(
218
+
self, nsid: str, json: dict[str, Any] | None = None, *, cast_to: None = None
219
+
) -> Any: ...
178
220
async def post(
179
221
self,
180
222
nsid: str,
181
223
json: dict[str, Any] | None = None,
182
224
*,
183
-
cast_to: Any = None,
225
+
cast_to: type[T] | None = None,
184
226
) -> Any:
185
227
"""POST an xrpc procedure by nsid. escape hatch for unwrapped endpoints."""
186
228
response = await self._http.post(
187
229
self._url(nsid), json=json, headers=self._headers()
188
230
)
189
-
return self._parse(response, cast_to)
231
+
return _parse(response, cast_to)
190
232
191
233
async def close(self) -> None:
192
234
if self._owns_http:
+23
tests/test_client.py
+23
tests/test_client.py
···
9
9
NotFoundError,
10
10
RateLimitError,
11
11
Semble,
12
+
SembleError,
12
13
ServerError,
13
14
)
15
+
from semble.types import CountResponse
14
16
from tests.conftest import AsyncClientFactory, SyncClientFactory
17
+
18
+
19
+
def empty_response_client() -> Semble:
20
+
def handler(request: httpx.Request) -> httpx.Response:
21
+
return httpx.Response(200)
22
+
23
+
return Semble(
24
+
api_key="sk_test",
25
+
http_client=httpx.Client(transport=httpx.MockTransport(handler)),
26
+
)
27
+
28
+
29
+
def test_empty_body_without_cast_to_returns_none() -> None:
30
+
assert empty_response_client().post("network.cosmik.card.updateNote", {}) is None
31
+
32
+
33
+
def test_empty_body_with_cast_to_raises() -> None:
34
+
with pytest.raises(SembleError, match="expected a json body"):
35
+
empty_response_client().get(
36
+
"network.cosmik.notification.getUnreadCount", cast_to=CountResponse
37
+
)
15
38
16
39
17
40
def test_api_key_header(sync_client: SyncClientFactory) -> None: