Skip to content

Cache

FileCache

ocx_mirror_sdk.cache.FileCache 

FileCache(domain: str, *, max_age: int = 3600, root: Path | None = None)

TTL-based blob cache with domain separation.

PARAMETER DESCRIPTION
domain

Namespace subdirectory (e.g. "github").

TYPE: str

max_age

Maximum age of cache entries in seconds. 0 disables caching (every get returns None, fetch always calls the loader).

TYPE: int DEFAULT: 3600

root

Root cache directory. Defaults to the value set by :func:configure, or ~/.cache/ocx-mirror-sdk/.

TYPE: Path | None DEFAULT: None

Source code in src/ocx_mirror_sdk/cache.py 
81
82
83
84
85
86
87
88
89
90
def __init__(
    self,
    domain: str,
    *,
    max_age: int = 3600,
    root: Path | None = None,
) -> None:
    self._domain = domain
    self._explicit_root = root
    self._max_age = max_age

get 

get(key: str) -> bytes | None

Return cached bytes for key, or None on miss / expiry.

RAISES DESCRIPTION
CacheError

The on-disk entry exists but cannot be read.
Source code in src/ocx_mirror_sdk/cache.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
def get(self, key: str) -> bytes | None:
    """Return cached bytes for *key*, or ``None`` on miss / expiry.

    Raises:
        CacheError: The on-disk entry exists but cannot be read.
    """
    if self._max_age <= 0:
        return None
    path = self._path(key)
    if not path.exists():
        return None
    try:
        age = time.time() - path.stat().st_mtime
        if age >= self._max_age:
            return None
        return path.read_bytes()
    except OSError as e:
        raise CacheError("failed to read cache entry", path=path) from e

put 

put(key: str, data: bytes) -> None

Store data under key. No-op when caching is disabled.

RAISES DESCRIPTION
CacheError

Writing the entry failed (disk full, permission).
Source code in src/ocx_mirror_sdk/cache.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def put(self, key: str, data: bytes) -> None:
    """Store *data* under *key*. No-op when caching is disabled.

    Raises:
        CacheError: Writing the entry failed (disk full, permission).
    """
    if self._max_age <= 0:
        return
    path = self._path(key)
    try:
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_bytes(data)
    except OSError as e:
        raise CacheError("failed to write cache entry", path=path) from e

fetch 

fetch(key: str, loader: Callable[[], bytes]) -> bytes

Return cached bytes or call loader, cache the result, and return it.

Source code in src/ocx_mirror_sdk/cache.py 
134
135
136
137
138
139
140
141
def fetch(self, key: str, loader: Callable[[], bytes]) -> bytes:
    """Return cached bytes or call *loader*, cache the result, and return it."""
    cached = self.get(key)
    if cached is not None:
        return cached
    data = loader()
    self.put(key, data)
    return data

get_json 

get_json(key: str) -> Any | None

Return the deserialized JSON object for key, or None on miss.

RAISES DESCRIPTION
CacheError

The on-disk entry exists but is not valid JSON.
Source code in src/ocx_mirror_sdk/cache.py 
145
146
147
148
149
150
151
152
153
154
155
156
157
def get_json(self, key: str) -> Any | None:
    """Return the deserialized JSON object for *key*, or ``None`` on miss.

    Raises:
        CacheError: The on-disk entry exists but is not valid JSON.
    """
    raw = self.get(key)
    if raw is None:
        return None
    try:
        return json.loads(raw)
    except json.JSONDecodeError as e:
        raise CacheError("cache entry is not valid JSON", path=self._path(key)) from e

put_json 

put_json(key: str, obj: Any) -> None

Serialize obj as JSON and store it under key.

Source code in src/ocx_mirror_sdk/cache.py 
159
160
161
def put_json(self, key: str, obj: Any) -> None:
    """Serialize *obj* as JSON and store it under *key*."""
    self.put(key, json.dumps(obj).encode())

fetch_json 

fetch_json(key: str, loader: Callable[[], Any]) -> Any

Return cached JSON or call loader, cache the result, and return it.

Unlike :meth:fetch, the loader returns a JSON-serializable object (not raw bytes) and the return value is the deserialized object.

Source code in src/ocx_mirror_sdk/cache.py 
163
164
165
166
167
168
169
170
171
172
173
174
def fetch_json(self, key: str, loader: Callable[[], Any]) -> Any:
    """Return cached JSON or call *loader*, cache the result, and return it.

    Unlike :meth:`fetch`, the *loader* returns a JSON-serializable object
    (not raw bytes) and the return value is the deserialized object.
    """
    cached = self.get_json(key)
    if cached is not None:
        return cached
    obj = loader()
    self.put_json(key, obj)
    return obj

configure

ocx_mirror_sdk.cache.configure 

configure(*, cache_root: Path | None = None) -> None

Override SDK-wide cache settings.

PARAMETER DESCRIPTION
cache_root

New default cache root directory. Use None to reset to ~/.cache/ocx-mirror-sdk/. Cache instances pick up the change on their next IO call.

TYPE: Path | None DEFAULT: None

Example

from pathlib import Path from ocx_mirror_sdk import configure configure(cache_root=Path("/tmp/my-cache"))

Source code in src/ocx_mirror_sdk/cache.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def configure(*, cache_root: Path | None = None) -> None:
    """Override SDK-wide cache settings.

    Args:
        cache_root: New default cache root directory. Use ``None`` to
            reset to ``~/.cache/ocx-mirror-sdk/``. Cache instances pick up
            the change on their next IO call.

    Example:
        >>> from pathlib import Path
        >>> from ocx_mirror_sdk import configure
        >>> configure(cache_root=Path("/tmp/my-cache"))
    """
    global _cache_root_override
    _cache_root_override = cache_root