okami.mom/albedo

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

Aly Raffaufrefactor(python): separate capture, state, and value codecs448d7dc

main
branchesmainall refs →
4.3 KiB116 linesraw
1"""Values crossing a kernel connection; process and reference ownership stay with callers."""
2
3from __future__ import annotations
4
5import base64
6import dataclasses
7import importlib
8from typing import Any
9
10import albedo_api
11
12
13class InvalidValue(ValueError):
14    """A peer supplied an invalid encoded value."""
15
16
17class Unencodable(Exception):
18    """The value cannot cross the connection; it stays a live remote reference."""
19
20
21def encode(value: object, depth: int = 0) -> object:
22    """One value as connection-crossable JSON; Unencodable means a live reference.
23
24    Dataclasses and list subclasses carry their class so the owner rebuilds the
25    real object and remote results read exactly like local ones.
26    """
27    if depth > 32:
28        raise Unencodable
29    if value is None or isinstance(value, (bool, int, float, str)):
30        return value
31    if isinstance(value, (bytes, bytearray, memoryview)):
32        return {"__bytes__": base64.b64encode(bytes(value)).decode("ascii")}
33    if isinstance(value, dict) and type(value) is dict:
34        return {str(key): encode(item, depth + 1) for key, item in value.items()}
35    if isinstance(value, albedo_api.Record):
36        cls = type(value)
37        return {
38            "__record__": cls.__module__ + "." + cls.__qualname__,
39            "fields": {
40                str(key): encode(item, depth + 1) for key, item in value.items()
41            },
42        }
43    if isinstance(value, (list, tuple)):
44        try:
45            if type(value) is not list:
46                cls = type(value)
47                return {
48                    "__list__": cls.__module__ + "." + cls.__qualname__,
49                    "items": [encode(item, depth + 1) for item in value],
50                }
51        except Unencodable:
52            raise
53        except (TypeError, ValueError):
54            raise Unencodable from None
55        return [encode(item, depth + 1) for item in value]
56    if dataclasses.is_dataclass(value) and not isinstance(value, type):
57        cls = type(value)
58        try:
59            return {
60                "__class__": cls.__module__ + "." + cls.__qualname__,
61                "fields": {
62                    field.name: encode(getattr(value, field.name), depth + 1)
63                    for field in dataclasses.fields(value)
64                },
65            }
66        except (TypeError, ValueError, AttributeError):
67            raise Unencodable from None
68    raise Unencodable
69
70
71def _load_class(marker: str) -> type:
72    """The class a wire marker names; both bundles are identical, so it exists."""
73    module, _, qualified = marker.rpartition(".")
74    obj: Any = importlib.import_module(module)
75    for part in qualified.split("."):
76        obj = getattr(obj, part)
77    return obj
78
79
80def decode(value: Any) -> Any:
81    """Rebuild real objects from wire values, so remote results read like local ones."""
82    if isinstance(value, dict):
83        if set(value) == {"__bytes__"} and isinstance(value.get("__bytes__"), str):
84            try:
85                return base64.b64decode(value["__bytes__"], validate=True)
86            except ValueError as error:
87                raise InvalidValue("invalid remote byte encoding") from error
88        if set(value) == {"__class__", "fields"} and isinstance(
89            value.get("__class__"), str
90        ):
91            try:
92                return _load_class(value["__class__"])(
93                    **{key: decode(item) for key, item in value["fields"].items()}
94                )
95            except Exception:
96                return {key: decode(item) for key, item in value["fields"].items()}
97        if set(value) == {"__record__", "fields"} and isinstance(
98            value.get("__record__"), str
99        ):
100            fields = {key: decode(item) for key, item in value["fields"].items()}
101            try:
102                return _load_class(value["__record__"])(fields)
103            except Exception:
104                return fields
105        if set(value) == {"__list__", "items"} and isinstance(
106            value.get("__list__"), str
107        ):
108            items = [decode(item) for item in value["items"]]
109            try:
110                return _load_class(value["__list__"])(items)
111            except Exception:
112                return items
113        return {key: decode(item) for key, item in value.items()}
114    if isinstance(value, list):
115        return [decode(item) for item in value]
116    return value