Repository navigation
Expand file tree
/
Copy pathwebhooks.py
More file actions
64 lines (47 loc) · 1.85 KB
/
Copy pathwebhooks.py
File metadata and controls
64 lines (47 loc) · 1.85 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
"""Webhook CRUD."""
from __future__ import annotations
from collections.abc import Sequence
from typing import TYPE_CHECKING
from stemsplit_python.models.webhooks import Webhook, WebhookListResponse
if TYPE_CHECKING:
from stemsplit_python._transport import Transport
class WebhooksResource:
"""``POST /webhooks`` / ``GET /webhooks`` / ``DELETE /webhooks/{id}``."""
def __init__(self, transport: Transport) -> None:
self._transport = transport
def create(
self,
*,
url: str,
events: Sequence[str] | None = None,
idempotency_key: str | None = None,
) -> Webhook:
"""Register a webhook subscription.
The returned :class:`Webhook` carries a ``secret`` field — capture it
now, the API never returns it again.
"""
body: dict[str, object] = {"url": url}
if events is not None:
body["events"] = list(events)
data = self._transport.request(
"POST",
"/webhooks",
json_body=body,
idempotency_key=idempotency_key,
retry_safe=True,
)
return Webhook.model_validate(data)
def list(self) -> list[Webhook]:
"""List every webhook subscription on the account.
Returns the webhooks directly (the wire shape wraps them in
``{"webhooks": [...]}``; we unwrap so callers can iterate).
"""
data = self._transport.request("GET", "/webhooks")
return WebhookListResponse.model_validate(data).webhooks
def delete(self, webhook_id: str) -> bool:
"""Delete a webhook subscription. Returns ``True`` on success."""
data = self._transport.request("DELETE", f"/webhooks/{webhook_id}")
if isinstance(data, dict):
return bool(data.get("deleted", True))
return True
__all__ = ["WebhooksResource"]