Last active
March 6, 2024 08:43
-
-
Save quaat/47148e04d985de27c5b1302387f5bb34 to your computer and use it in GitHub Desktop.
Example of how to abstract the storage backend for a caching service in Python, using the principles of protocols from type hinting for runtime type checking.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from dataclasses import dataclass, field | |
from typing import Any, Dict, Protocol, runtime_checkable | |
from redis import Redis | |
@runtime_checkable | |
class ICacheBackend(Protocol): | |
""" | |
Defines an interface for cache backends. Any class that implements this protocol | |
must provide `set`, `get`, and `aclose` methods. This allows for flexibility in | |
the cache implementation, as different storage mechanisms can be used as long as | |
they conform to this interface. | |
""" | |
def set(self, key: str, value: Any) -> None: | |
... | |
def get(self, key: str) -> Any: | |
... | |
def close(self) -> None: | |
... | |
@dataclass | |
class NameService: | |
""" | |
NameService class: Uses a cache backend (that implements the ICacheBackend protocol) | |
to store and retrieve names. It abstracts the underlying cache mechanism from the user. | |
""" | |
cache: ICacheBackend = field( | |
default_factory=dict, | |
metadata={"help": "Storage backend implementing the ICacheBackend interface"}) | |
def __post_init__(self): | |
if not isinstance(self.cache, ICacheBackend): | |
raise TypeError("Input cache must implement the ICacheBackend interface") | |
def store_name(self, name: str): | |
self.cache.set("name", name) | |
def get_name(self) -> str: | |
return self.cache.get("name") | |
@dataclass | |
class LocalStorage: | |
""" | |
LocalStorage class: An implementation of the ICacheBackend protocol using a Python | |
dictionary for local storage. It demonstrates how a class can conform to the protocol, | |
providing specific implementations of set and get methods. Notice the encoding to | |
UTF-8 in the set method, implying that the storage handles binary data, which could | |
be useful for compatibility with other caching backends like Redis. | |
""" | |
memory: Dict[str, str] | |
def set(self, key, name) -> None: | |
self.memory[key] = name.encode("utf-8") | |
def get(self, key) -> Any: | |
return self.memory[key] | |
def close(self) -> None: | |
pass | |
# Usage example | |
# NameService instantiation with LocalStorage: Here, NameService is | |
# instantiated with LocalStorage as its cache backend. Note that the | |
# NameService does not need to know the specifics of the cache backend. | |
_name = "Luke Skywalker" | |
service = NameService(cache=LocalStorage(memory={})) | |
service.store_name(_name) | |
name = service.get_name().decode("utf-8") | |
print(f"From local dict: {name}. Cache type: {type(service.cache)}") | |
# Storing and retrieving with Redis: | |
# Similar to the local storage example, this demonstrates storing and | |
# retrieving data with Redis. The decode step is necessary as Redis | |
# stores data in binary format. | |
service = NameService(cache=Redis(host="redis", port="6379")) | |
service.store_name(_name) | |
name = service.get_name().decode("utf-8") | |
print(f"From redis: {name}. Cache type: {type(service.cache)}") |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Updated the
NameService
to check if the cache complies with theICacheBackend
interface.