From c37a51c2ebccf273521524e168b765b3757ea8a3 Mon Sep 17 00:00:00 2001 From: "Mike A." Date: Mon, 14 Jul 2025 22:10:14 +0200 Subject: [PATCH] refactor(reports): improve `Serializable` base class --- findmy/reports/anisette.py | 81 ++++++++++++++++++++++++++++++-------- findmy/util/abc.py | 35 +++++++++++++--- findmy/util/files.py | 34 ++++++++++++++++ 3 files changed, 128 insertions(+), 22 deletions(-) create mode 100644 findmy/util/files.py diff --git a/findmy/reports/anisette.py b/findmy/reports/anisette.py index 5beee2b..3ef9346 100644 --- a/findmy/reports/anisette.py +++ b/findmy/reports/anisette.py @@ -10,17 +10,49 @@ from abc import ABC, abstractmethod from datetime import datetime, timezone from io import BytesIO from pathlib import Path -from typing import BinaryIO +from typing import BinaryIO, Literal, TypedDict, Union from anisette import Anisette, AnisetteHeaders from typing_extensions import override from findmy.util.abc import Closable, Serializable +from findmy.util.files import read_data_json, save_and_return_json from findmy.util.http import HttpSession logger = logging.getLogger(__name__) +class RemoteAnisetteMapping(TypedDict): + """JSON mapping representing state of a remote Anisette provider.""" + + type: Literal["aniRemote"] + url: str + + +class LocalAnisetteMapping(TypedDict): + """JSON mapping representing state of a local Anisette provider.""" + + type: Literal["aniLocal"] + prov_data: str + + +AnisetteMapping = Union[RemoteAnisetteMapping, LocalAnisetteMapping] + + +def get_provider_from_mapping( + mapping: AnisetteMapping, + *, + libs_path: str | Path | None = None, +) -> RemoteAnisetteProvider | LocalAnisetteProvider: + """Get the correct Anisette provider instance from saved JSON data.""" + if mapping["type"] == "aniRemote": + return RemoteAnisetteProvider.from_json(mapping) + if mapping["type"] == "aniLocal": + return LocalAnisetteProvider.from_json(mapping, libs_path=libs_path) + msg = f"Unknown anisette type: {mapping['type']}" + raise ValueError(msg) + + class BaseAnisetteProvider(Closable, Serializable, ABC): """ Abstract base class for Anisette providers. @@ -173,20 +205,25 @@ class RemoteAnisetteProvider(BaseAnisetteProvider): self._anisette_data_expires_at: float = 0 @override - def serialize(self) -> dict: + def to_json(self, dst: str | Path | None = None, /) -> RemoteAnisetteMapping: """See `BaseAnisetteProvider.serialize`.""" - return { - "type": "aniRemote", - "url": self._server_url, - } + return save_and_return_json( + { + "type": "aniRemote", + "url": self._server_url, + }, + dst, + ) @classmethod @override - def deserialize(cls, data: dict) -> RemoteAnisetteProvider: + def from_json(cls, val: str | Path | RemoteAnisetteMapping) -> RemoteAnisetteProvider: """See `BaseAnisetteProvider.deserialize`.""" - assert data["type"] == "aniRemote" + val = read_data_json(val) - server_url = data["url"] + assert val["type"] == "aniRemote" + + server_url = val["url"] return cls(server_url) @@ -276,24 +313,34 @@ class LocalAnisetteProvider(BaseAnisetteProvider): ) @override - def serialize(self) -> dict: + def to_json(self, dst: str | Path | None = None, /) -> LocalAnisetteMapping: """See `BaseAnisetteProvider.serialize`.""" with BytesIO() as buf: self._ani.save_provisioning(buf) prov_data = base64.b64encode(buf.getvalue()).decode("utf-8") - return { - "type": "aniLocal", - "prov_data": prov_data, - } + return save_and_return_json( + { + "type": "aniLocal", + "prov_data": prov_data, + }, + dst, + ) @classmethod @override - def deserialize(cls, data: dict, libs_path: str | Path | None = None) -> LocalAnisetteProvider: + def from_json( + cls, + val: str | Path | LocalAnisetteMapping, + *, + libs_path: str | Path | None = None, + ) -> LocalAnisetteProvider: """See `BaseAnisetteProvider.deserialize`.""" - assert data["type"] == "aniLocal" + val = read_data_json(val) - state_blob = BytesIO(base64.b64decode(data["prov_data"])) + assert val["type"] == "aniLocal" + + state_blob = BytesIO(base64.b64decode(val["prov_data"])) return cls(state_blob=state_blob, libs_path=libs_path) diff --git a/findmy/util/abc.py b/findmy/util/abc.py index 101e11a..7b77d3e 100644 --- a/findmy/util/abc.py +++ b/findmy/util/abc.py @@ -5,6 +5,10 @@ from __future__ import annotations import asyncio import logging from abc import ABC, abstractmethod +from typing import TYPE_CHECKING, Generic, Self, TypeVar + +if TYPE_CHECKING: + from pathlib import Path logging.getLogger(__name__) @@ -38,16 +42,37 @@ class Closable(ABC): pass -class Serializable(ABC): +T = TypeVar("T", bound=dict) + + +class Serializable(Generic[T], ABC): """ABC for serializable classes.""" @abstractmethod - def serialize(self) -> dict: - """Serialize the object to a JSON-serializable dictionary.""" + def to_json(self, dst: str | Path | None = None, /) -> T: + """ + Export the current state of the object as a JSON-serializable dictionary. + + If an argument is provided, the output will also be written to that file. + + The output of this method is guaranteed to be JSON-serializable, and passing + the return value of this function as an argument to `Serializable.from_json` + will always result in an exact copy of the internal state as it was when exported. + + You are encouraged to save and load object states to and from disk whenever possible, + to prevent unnecessary API calls or otherwise unexpected behavior. + """ raise NotImplementedError @classmethod @abstractmethod - def deserialize(cls, data: dict) -> Serializable: - """Deserialize the object from a JSON-serializable dictionary.""" + def from_json(cls, val: str | Path | T, /) -> Self: + """ + Restore state from a previous `Closable.to_json` export. + + If given a str or Path, it must point to a json file from `Serializable.to_json`. + Otherwise, it should be the Mapping itself. + + See `Serializable.to_json` for more information. + """ raise NotImplementedError diff --git a/findmy/util/files.py b/findmy/util/files.py new file mode 100644 index 0000000..e58bfd9 --- /dev/null +++ b/findmy/util/files.py @@ -0,0 +1,34 @@ +"""Utilities to simplify reading and writing data from and to files.""" + +from __future__ import annotations + +import json +from collections.abc import Mapping +from pathlib import Path +from typing import TypeVar, cast + +T = TypeVar("T", bound=Mapping) + + +def save_and_return_json(data: T, dst: str | Path | None) -> T: + """Save and return a JSON-serializable data structure.""" + if dst is None: + return data + + if isinstance(dst, str): + dst = Path(dst) + + dst.write_text(json.dumps(data, indent=4)) + + return data + + +def read_data_json(val: str | Path | T) -> T: + """Read JSON data from a file if a path is passed, or return the argument itself.""" + if isinstance(val, str): + val = Path(val) + + if isinstance(val, Path): + val = cast("T", json.loads(val.read_text())) + + return val