okami.mom/albedo

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

dawnlinks: link workspaces into one project, sharing memory and work items2f43cb2

main
branchesmainall refs →
2.0 KiB71 linesraw
1from __future__ import annotations
2
3from albedo_api import Host, PythonApi, Record
4import builtins
5from typing import cast
6
7
8class WorkItem(Record):
9    """One ledger item: item.id and item["id"] both work."""
10
11    id: int
12    title: str
13    notes: str
14    status: str
15    parent: int | None
16    session: str | None
17    run: str | None
18    revision: int
19    workspace: str  # where it was filed: this workspace or a linked one
20
21
22host: Host
23
24
25class Work:
26    async def list(self, after: int = 0, limit: int = 50) -> builtins.list[WorkItem]:
27        return [
28            WorkItem(item)
29            for item in cast(
30                builtins.list[dict],
31                await host("work.list", {"after": after, "limit": limit}),
32            )
33        ]
34
35    async def get(self, id: int) -> WorkItem:
36        return WorkItem(cast(dict, await host("work.get", {"id": id})))
37
38    async def create(
39        self, title: str, notes: str = "", parent: int | None = None
40    ) -> WorkItem:
41        return WorkItem(
42            cast(
43                dict,
44                await host(
45                    "work.create", {"title": title, "notes": notes, "parent": parent}
46                ),
47            )
48        )
49
50    async def update(self, id: int, *, revision: int, **changes: object) -> WorkItem:
51        allowed = {"title", "notes", "status", "session", "run"}
52        if changes.keys() - allowed:
53            raise ValueError("unknown work fields: " + repr(changes.keys() - allowed))
54        return WorkItem(
55            cast(
56                dict,
57                await host("work.update", {"id": id, "revision": revision, **changes}),
58            )
59        )
60
61    async def delete(self, id: int, *, revision: int) -> WorkItem:
62        """Remove an item at the revision you last read; one with sub-items stays."""
63        return WorkItem(
64            cast(dict, await host("work.delete", {"id": id, "revision": revision}))
65        )
66
67
68def setup(api: PythonApi) -> dict[str, object]:
69    global host
70    host = api.host
71    return {"work": Work(), "WorkError": api.HostError}