okami.mom/albedo

git clone https://git.nkp.pet/okami.mom/albedo

Aly Raffauffix(python): enforce ruff and ty checks4ad84fd

main
branchesmainall refs →
2.7 KiB89 linesraw
1from __future__ import annotations
2
3import builtins
4from typing import cast
5from albedo_api import Host, PythonApi, Record
6
7
8class Hook(Record):
9    id: str
10    session: str
11    name: str
12    enabled: bool
13    url: str
14    signatureHeader: str
15    signaturePrefix: str
16    revision: int
17
18
19host: Host
20
21
22class Webhooks:
23    async def list(self) -> builtins.list[Hook]:
24        """Hooks targeting this session. Agent management requires human opt-in."""
25        return [
26            Hook(item)
27            for item in cast(builtins.list[dict], await host("webhooks.list", {}))
28        ]
29
30    async def delivery(self, id: str) -> Record:
31        """Read one accepted payload for this session, even without management permission."""
32        return Record(cast(dict, await host("webhooks.delivery", {"id": id})))
33
34    async def create(self, name: str, secret: str | None = None) -> Record:
35        """Create a signed endpoint; returns a generated secret when omitted."""
36        value = cast(
37            dict, await host("webhooks.create", {"name": name, "secret": secret})
38        )
39        return Record({"hook": Hook(value["hook"]), "secret": value["secret"]})
40
41    async def rotate(
42        self, id: str, *, revision: int, secret: str | None = None
43    ) -> Record:
44        value = cast(
45            dict,
46            await host(
47                "webhooks.rotate", {"id": id, "revision": revision, "secret": secret}
48            ),
49        )
50        return Record({"hook": Hook(value["hook"]), "secret": value["secret"]})
51
52    async def configure(
53        self, id: str, *, revision: int, header: str, prefix: str = "sha256="
54    ) -> Hook:
55        return Hook(
56            cast(
57                dict,
58                await host(
59                    "webhooks.configure",
60                    {
61                        "id": id,
62                        "revision": revision,
63                        "header": header,
64                        "prefix": prefix,
65                    },
66                ),
67            )
68        )
69
70    async def enable(self, id: str, *, revision: int) -> Hook:
71        return Hook(
72            cast(dict, await host("webhooks.enable", {"id": id, "revision": revision}))
73        )
74
75    async def disable(self, id: str, *, revision: int) -> Hook:
76        return Hook(
77            cast(dict, await host("webhooks.disable", {"id": id, "revision": revision}))
78        )
79
80    async def delete(self, id: str, *, revision: int) -> Hook:
81        return Hook(
82            cast(dict, await host("webhooks.delete", {"id": id, "revision": revision}))
83        )
84
85
86def setup(api: PythonApi) -> dict[str, object]:
87    global host
88    host = api.host
89    return {"webhooks": Webhooks(), "WebhookError": api.HostError}