codekingpro/portable-devtools

1from __future__ import annotations2 3from typing import Any, Protocol, runtime_checkable4 5 6class UntypedSerializerProtocol(Protocol):7    """Protocol for serialization and deserialization of objects."""8 9    def dumps(self, obj: Any) -> bytes: ...10 11    def loads(self, data: bytes) -> Any: ...12 13 14@runtime_checkable15class SerializerProtocol(Protocol):16    """Protocol for serialization and deserialization of objects.17 18    - `dumps_typed`: Serialize an object to a tuple `(type, bytes)`.19    - `loads_typed`: Deserialize an object from a tuple `(type, bytes)`.20 21    Valid implementations include the `pickle`, `json` and `orjson` modules.22    """23 24    def dumps_typed(self, obj: Any) -> tuple[str, bytes]: ...25 26    def loads_typed(self, data: tuple[str, bytes]) -> Any: ...27 28 29class SerializerCompat(SerializerProtocol):30    def __init__(self, serde: UntypedSerializerProtocol) -> None:31        self.serde = serde32 33    def dumps_typed(self, obj: Any) -> tuple[str, bytes]:34        return type(obj).__name__, self.serde.dumps(obj)35 36    def loads_typed(self, data: tuple[str, bytes]) -> Any:37        return self.serde.loads(data[1])38 39 40def maybe_add_typed_methods(41    serde: SerializerProtocol | UntypedSerializerProtocol,42) -> SerializerProtocol:43    """Wrap serde old serde implementations in a class with loads_typed and dumps_typed for backwards compatibility."""44 45    if not isinstance(serde, SerializerProtocol):46        return SerializerCompat(serde)47 48    return serde49 50 51class CipherProtocol(Protocol):52    """Protocol for encryption and decryption of data.53 54    - `encrypt`: Encrypt plaintext.55    - `decrypt`: Decrypt ciphertext.56    """57 58    def encrypt(self, plaintext: bytes) -> tuple[str, bytes]:59        """Encrypt plaintext. Returns a tuple `(cipher name, ciphertext)`."""60        ...61 62    def decrypt(self, ciphername: str, ciphertext: bytes) -> bytes:63        """Decrypt ciphertext. Returns the plaintext."""64        ...65