Skip to content

Instantly share code, notes, and snippets.

@quaat
Last active March 6, 2024 08:43
Show Gist options
  • Save quaat/47148e04d985de27c5b1302387f5bb34 to your computer and use it in GitHub Desktop.
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.
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)}")
@quaat
Copy link
Author

quaat commented Mar 6, 2024

Updated the NameService to check if the cache complies with the ICacheBackend interface.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment