Skip to content

Cache Utilities

Helper functions for key normalization, TTL/expiry arithmetic, and serialization.

Key Utilities

normalize_key

normalize_key 

normalize_key(key: str) -> str

Normalize cache key to a safe form.

Source code in src/jinpy_utils/cache/utils.py 
14
15
16
17
18
19
20
def normalize_key(key: str) -> str:
    """Normalize cache key to a safe form."""
    key = key.strip()
    if not key:
        raise ValueError("Cache key cannot be empty")
    # Prevent accidental whitespace/control characters
    return "".join(ch for ch in key if ch.isprintable())

Time Utilities

now_seconds

now_seconds 

now_seconds() -> float

Current monotonic-ish epoch seconds.

Source code in src/jinpy_utils/cache/utils.py 
23
24
25
def now_seconds() -> float:
    """Current monotonic-ish epoch seconds."""
    return time.time()

compute_expiry

compute_expiry 

compute_expiry(ttl: float | None) -> float | None

Compute absolute expiry timestamp in seconds.

Source code in src/jinpy_utils/cache/utils.py 
28
29
30
31
32
33
34
35
def compute_expiry(ttl: float | None) -> float | None:
    """Compute absolute expiry timestamp in seconds."""
    if ttl is None:
        return None
    if ttl <= 0:
        # Immediate expiry
        return now_seconds() - 1
    return now_seconds() + ttl

remaining_ttl

remaining_ttl 

remaining_ttl(expiry: float | None) -> float | None

Compute remaining ttl from absolute expiry.

Source code in src/jinpy_utils/cache/utils.py 
38
39
40
41
42
43
def remaining_ttl(expiry: float | None) -> float | None:
    """Compute remaining ttl from absolute expiry."""
    if expiry is None:
        return None
    remaining = expiry - now_seconds()
    return remaining if remaining > 0 else 0.0

Serialization

default_serializer

default_serializer 

default_serializer(
    kind: SerializerType,
) -> tuple[Callable[[Any], bytes], Callable[[bytes], Any]]

Return serializer, deserializer functions based on kind.

Source code in src/jinpy_utils/cache/utils.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def default_serializer(
    kind: SerializerType,
) -> tuple[Callable[[Any], bytes], Callable[[bytes], Any]]:
    """Return serializer, deserializer functions based on kind."""
    if kind == "json":

        def ser(obj: Any) -> bytes:
            return json.dumps(
                obj, default=str, separators=(",", ":"), ensure_ascii=False
            ).encode("utf-8")

        def de(data: bytes) -> Any:
            return json.loads(data.decode("utf-8"))

        return ser, de

    if kind == "pickle":

        def ser(obj: Any) -> bytes:
            return pickle.dumps(obj, protocol=pickle.HIGHEST_PROTOCOL)

        def de(data: bytes) -> Any:
            return pickle.loads(data)

        return ser, de

    if kind == "str":

        def ser(obj: Any) -> bytes:
            return str(obj).encode("utf-8")

        def de(data: bytes) -> Any:
            return data.decode("utf-8")

        return ser, de

    if kind == "bytes":

        def ser(obj: Any) -> bytes:
            if isinstance(obj, (bytes | bytearray | memoryview)):
                return bytes(obj)
            raise TypeError("Expected bytes-like value")

        def de(data: bytes) -> Any:
            return data

        return ser, de

    raise ValueError(f"Unsupported serializer type: {kind}")

Examples

from jinpy_utils.cache.utils import normalize_key, compute_expiry, remaining_ttl

k = normalize_key("  user:1  ")
exp = compute_expiry(10)
print(remaining_ttl(exp))  # ~10.0 initially