pip install pypercache

git clone https://github.com/BrandonBahret/PyperCache.git
cd PyperCache
pip install .

from pypercache import Cache, RequestLogger
from pypercache.api_wrapper import ApiWrapper
from pypercache.models.apimodel import Alias, Columns, Lazy, Timestamp, apimodel
from pypercache.query import JsonInjester

Why this example is the right starting point

JSONPlaceholder is intentionally simple in the best possible way. There is one base URL, no authentication, and small JSON payloads with predictable structure. That makes it a good teaching surface for the core wrapper pattern without introducing concerns like OAuth, pagination cursors, or multi-origin routing.

The goal of the example is not to exercise every PyperCache feature. It is to show the smallest realistic client you can copy into a real project and then extend.

Start with models that match the payloads you already have

The wrapper begins with a few response models. Each one is decorated with @apimodel(validate=True), which means incoming JSON is hydrated into typed Python objects and checked against the annotated field types during construction.

Where the upstream API uses camelCase keys, the example maps them into snake_case attributes with Alias(...). That keeps the Python surface conventional without forcing you to rewrite the raw payload format.

from typing import Annotated
from pypercache.models.apimodel import Alias, apimodel

@apimodel(validate=True)
class Post:
    user_id: Annotated[int, Alias("userId")]
    id: int
    title: str
    body: str

@apimodel(validate=True)
class Company:
    name: str
    catch_phrase: Annotated[str, Alias("catchPhrase")]
    bs: str

Nested response objects work the same way. The User model contains an Address, which contains a Geo, and hydration recursively builds those child objects for you.

Wrap the API in thin endpoint methods

Once the models exist, the wrapper class stays small. Its constructor passes a single origin, a cache path, and a default expiry to ApiWrapper. After that, each endpoint is just a thin method around request().

class JSONPlaceholderClient(ApiWrapper):
    def __init__(self) -> None:
        super().__init__(
            origins={"default": "https://jsonplaceholder.typicode.com"},
            default_origin="default",
            cache_path="jsonplaceholder_cache.json",
            default_expiry=300,
            request_log_path="jsonplaceholder_requests.log",
        )

    def get_post(self, post_id: int) -> Post:
        return self.request("GET", f"/posts/{post_id}", expected="json", cast=Post)

    def list_post_comments(self, post_id: int) -> list[Comment]:
        return self.request(
            "GET",
            f"/posts/{post_id}/comments",
            expected="json",
            cast=list[Comment],
        )

This is the part worth internalizing: the endpoint methods describe intent, not plumbing. You name the HTTP method, path, expected response type, and the model to hydrate into. URL joining, JSON decoding, cache lookup, cache writeback, and object construction are handled underneath.

Let cached GETs make the second call boring

The companion app.py script demonstrates the runtime behavior. It first fetches post #1, then immediately fetches the same post again. Because the method is a GET with expected="json", PyperCache stores the first response and serves the second one from the cache while the entry is still fresh.

client = JSONPlaceholderClient(
    cache_path="jsonplaceholder_cache.json",
    request_log_path="jsonplaceholder_requests.log",
)

post = client.get_post(1)
cached_post = client.get_post(1)

author = client.get_user(post.user_id)
comments = client.list_post_comments(post.id)
todo = client.get_todo(1)

That sequence is the narrative center of the example. The first request proves the wrapper can fetch and hydrate a resource. The second request proves the cache is active. The follow-up calls show that once you have a typed object back, using fields like post.user_id to drive the next request feels like ordinary Python code rather than manual JSON plumbing.

Treat writes differently from reads

The example ends with a create_post() method. This is where the code deliberately opts out of the cache:

def create_post(self, *, user_id: int, title: str, body: str) -> CreatedPost:
    return self.request(
        "POST",
        "/posts",
        expected="json",
        json_body={
            "userId": user_id,
            "title": title,
            "body": body,
        },
        use_cache=False,
        cast=CreatedPost,
    )

This mirrors the documented caching rule for ApiWrapper.request(): cached responses are for read paths, not mutating calls. The explicit use_cache=False makes that policy visible in the method body, which is useful both as documentation for readers and as protection if the method grows later.

JSONPlaceholder itself does not persist the new record, so the point is not durable remote state. The point is to show how request encoding with json_body=... and typed response hydration still work cleanly on the write path.

What to copy into your own project

If you are starting a real client, the practical pattern is straightforward: model the responses you care about, subclass ApiWrapper, make each endpoint method one call to request(), and let the cache cover repeatable GET traffic. Add custom headers in get_session() only when the API requires them.

When the process exits, call client.close(). That closes the wrapper's cache cleanly. For SQLite-backed caches, it also flushes pending writes if manual flush mode was enabled.

Create a cache

Pass a file path. The extension determines which storage backend is used.

from pypercache import Cache

cache = Cache(filepath="app_cache.pkl")

Omit filepath and it defaults to api-cache.pkl in the working directory.

Store and read

cache.store("user:1", {"name": "Alice", "role": "admin"})
cache.store("user:2", {"name": "Bob", "role": "user"}, expiry=3600)

record = cache.get("user:1")
print(record.data)          # the raw payload
print(record.is_data_stale) # False — no expiry set
print(record.expiry)        # inf

get() always returns a CacheRecord. Access .data for the raw payload. It raises KeyError if the key doesn't exist.

TTL and freshness

The typical pattern is fetch-or-cache: check freshness, run the expensive work if stale, then read from cache.

key = "expensive-result"

if not cache.is_data_fresh(key):
    payload = run_expensive_work()
    cache.store(key, payload, expiry=300)

result = cache.get(key).data

is_data_fresh() returns False when the key doesn't exist or the record's TTL has elapsed. It never throws.

Updating a record

update() replaces the payload and refreshes the timestamp while keeping the original expiry and any cast metadata intact.

cache.update("settings", {"theme": "light", "page_size": 50})

Typed round-trips

Store a record with cast=MyModel and retrieve it as a fully hydrated object using get_object().

from pypercache.models.apimodel import apimodel

@apimodel(validate=True)
class User:
    id: int
    name: str

cache.store("user:1", {"id": 1, "name": "Ada"}, cast=User)

user = cache.get_object("user:1")
print(user.name)  # "Ada"

cache.close()

Querying the payload inline

Every CacheRecord exposes a JsonInjester via .query, so you can navigate the stored data without a separate extraction step.

cache.store("order:1", {
    "customer": {"name": "Sarah Johnson"},
    "items": [
        {"name": "Laptop", "price": 999.99, "category": "electronics"},
        {"name": "Mouse", "price": 59.99, "category": "electronics"},
    ],
})

q = cache.get("order:1").query
print(q.get("customer.name"))
print(q.get("items?name*"))
print(q.get("items?category=electronics"))

Clearing the cache

cache.completely_erase_cache()
print(cache.has("user:1"))  # False

Cache(filepath="cache.pkl")      # Pickle — default
Cache(filepath="cache.json")     # JSON — human-readable
Cache(filepath="cache.manifest") # Chunked — large stores
Cache(filepath="cache.db")       # SQLite — best write behavior

Quick comparison

Extension	Backend	Best for
.pkl	Pickle	Default choice. Simple, fast, Python-native serialization. Rewrites the whole file on save.
.json	JSON	Human-readable files you want to inspect or diff. Not great for large caches — full rewrite on every save.
.manifest	Chunked	Many keys or large payloads. Per-chunk rewrites instead of full-file rewrites. Backed by a directory, not a single file.
.db	SQLite	Frequent writes, growing caches, concurrent access. WAL mode with batched flushes — the most operationally robust option.

SQLite specifics

SQLite loads all rows into memory on open and uses WAL mode. For Cache, store() and update() flush immediately by default. Manual flush mode is opt-in; when enabled, writes stay in memory until flush() or close() runs. Prefer using it as a context manager:

from pypercache.storage import SQLiteStorage

with SQLiteStorage("cache.db") as store:
    ...

Or just call cache.close() before your process exits — that's what ApiWrapper.close() does under the hood.

Request logging is separate

RequestLogger writes JSONL audit records to its own file. It doesn't know about the cache backend and you don't need a cache to use it.

from pypercache import RequestLogger

log = RequestLogger("requests.log")
log.log(uri="/v1/items", status=200)

for entry in log.get_logs_from_last_seconds(60):
    print(entry.data["uri"], entry.data["status"])

Basic subclass pattern

Override get_session() to set headers or auth, then add one method per endpoint.

from typing import Annotated
from pypercache.api_wrapper import ApiWrapper
from pypercache.models.apimodel import Alias, apimodel

@apimodel(validate=True)
class Widget:
    id: int
    display_name: Annotated[str, Alias("displayName")]

class WidgetClient(ApiWrapper):
    def __init__(self) -> None:
        super().__init__(
            origins={"default": "https://api.example.com"},
            default_origin="default",
            cache_path="widget_cache.json",
            default_expiry=300,
        )

    def get_session(self):
        session = super().get_session()
        session.headers.update({"User-Agent": "widget-client/1.0"})
        return session

    def list_widgets(self) -> list[Widget]:
        return self.request("GET", "/widgets", expected="json", cast=list[Widget])

    def get_widget(self, widget_id: int) -> Widget:
        return self.request("GET", f"/widgets/{widget_id}", expected="json", cast=Widget)

    def create_widget(self, name: str) -> Widget:
        return self.request(
            "POST", "/widgets",
            expected="json",
            json_body={"name": name},
            use_cache=False,
            cast=Widget,
        )

Constructor

super().__init__(
    origins={"api": "https://api.example.com"},
    default_origin="api",
    cache_path="example.db",      # None disables caching
    default_expiry=300,
    request_log_path="reqs.log",  # None disables logging
    timeout=10,
)

Multi-origin APIs

When your API spans multiple hosts, add each as an origin and select it per request with origin=.

super().__init__(
    origins={
        "api": "https://api.example.com",
        "auth": "https://auth.example.com",
    },
    default_origin="api",
)

# In an endpoint method:
return self.request("GET", "/token", origin="auth", ...)

Caching rules

The behavior is intentionally simple: only GET requests with expected="auto" or expected="json" are cached. Stale entries are refetched, not served. Pass use_cache=False to skip both lookup and writeback — always do this for POST, PUT, and DELETE requests.

Error handling

from pypercache.api_wrapper import ApiHTTPError

try:
    client.get_widget(404)
except ApiHTTPError as exc:
    print(exc.status_code)
    print(exc.url)
    print(exc.body)

Lifecycle

client = WidgetClient()
try:
    widgets = client.list_widgets()
finally:
    client.close()  # closes cache + session

Manual fetch-or-cache

This is the core pattern — everything ApiWrapper.request() automates, written explicitly.

import requests
from pypercache import Cache, RequestLogger
from pypercache.models.apimodel import apimodel

@apimodel(validate=True)
class User:
    id: int
    name: str
    email: str

cache = Cache(filepath="users_cache.db")
log = RequestLogger(filepath="users_requests.log")

key = "user:1"
if not cache.is_data_fresh(key):
    response = requests.get("https://api.example.com/users/1", timeout=10)
    response.raise_for_status()
    payload = response.json()
    log.log(uri=response.url, status=response.status_code)
    cache.store(key, payload, expiry=300, cast=User)

user = cache.get_object(key)
print(user.name)
cache.close()

Navigating a loaded record

record = cache.get("user:1")
print(record.query.has("name"))
print(record.query.get("address.city", default_value="unknown"))

@apimodel vs @Cache.cached

Use @apimodel when you want aliases, timestamps, lazy fields, or column transforms. Use @Cache.cached when you only need lightweight class registration — your class already knows how to accept the cached payload.

@Cache.cached
class SearchResult:
    def __init__(self, hits=None, total=0, **kwargs):
        self.hits = hits or []
        self.total = total

cache.store("search:python", {"hits": [], "total": 0}, cast=SearchResult)
result = cache.get_object("search:python")

Using RequestLogger standalone

from pypercache import RequestLogger

log = RequestLogger("requests.log")
log.log(uri="/health", status=200)

for entry in log.get_logs_from_last_seconds(60):
    print(entry.data["uri"], entry.data["status"])

Basic usage

from pypercache.models.apimodel import apimodel

@apimodel
class Widget:
    id: int
    name: str

widget = Widget({"id": 1, "name": "Gear"})
print(widget.name)
print(Widget.from_dict({"id": 2, "name": "Bolt"}).id)
print(widget.as_dict())

Nested hydration

Annotated nested model types are hydrated automatically — no explicit Address(raw_dict) calls needed.

@apimodel
class Address:
    city: str

@apimodel
class Company:
    name: str
    address: Address

@apimodel
class User:
    id: int
    company: Company
    previous: list[Address]

u = User({
    "id": 1,
    "company": {"name": "ACME", "address": {"city": "Redmond"}},
    "previous": [{"city": "Phoenix"}, {"city": "Tempe"}],
})
print(u.company.address.city)  # "Redmond"
print(u.previous[0].city)     # "Phoenix"

Validation and strictness

@apimodel(validate=True)
class User:
    id: int
    name: str

validate=True checks hydrated values against type annotations. Add strict=True to also raise on missing annotated fields (instead of silently storing UNSET).

from pypercache.models.validation import ApiModelValidationError

@apimodel(validate=True, strict=True)
class StrictUser:
    id: int
    name: str

try:
    StrictUser({"id": 1})       # missing "name"
except ApiModelValidationError as exc:
    print(exc)

Alias — mapping raw keys

Use when the raw payload uses a key you don't want on the Python object.

from typing import Annotated
from pypercache.models.apimodel import Alias, apimodel

@apimodel(validate=True)
class Post:
    user_id: Annotated[int, Alias("userId")]
    title: str
    body: str

Timestamp — parsing date fields

from datetime import datetime
from pypercache.models.apimodel import Alias, Timestamp, apimodel

@apimodel(validate=True)
class AuditRecord:
    created_at: Annotated[datetime, Timestamp()]
    refreshed_at: Annotated[datetime, Alias("refreshedAt"), Timestamp(unit="ms")]

Timestamp() handles ISO 8601 strings, numeric Unix timestamps, millisecond timestamps (unit="ms"), and explicit format strings.

Columns — parallel arrays to rows

Some APIs return data as a dict of parallel arrays. Columns converts that into a list of row model instances.

from pypercache.models.apimodel import Columns, apimodel

@apimodel(validate=True)
class HourRow:
    time: int
    temperature_2m: float

@apimodel(validate=True)
class Forecast:
    hourly: Annotated[list[HourRow], Columns(required=("time", "temperature_2m"))]

forecast = Forecast({
    "hourly": {
        "time": [1713000000, 1713003600],
        "temperature_2m": [21.2, 22.5],
    }
})
print(forecast.hourly[0].temperature_2m)  # 21.2

Lazy — deferred hydration

Wrap a type in Lazy[T] and it won't be hydrated until first access. Useful for large or expensive nested fields.

from pypercache.models.apimodel import Lazy, apimodel

@apimodel(validate=True)
class User:
    id: int
    profile: Lazy[Profile]

Lazy composes with Alias, Timestamp, and Columns.

Standalone usage

from pypercache.query import JsonInjester

data = {
    "meta": {"total": 2},
    "users": [
        {"name": "Ada", "role": "admin"},
        {"name": "Linus", "role": "member"},
    ],
}

q = JsonInjester(data)
print(q.get("meta.total"))
print(q.get("users?role=admin"))
print(q.get("users?name*"))

Or access it directly from a CacheRecord:

q = cache.get("order:1").query
print(q.get("customer.name"))

Constructor options

root — move the starting cursor to a subtree:

q = JsonInjester({"meta": {"total": 5}}, root="meta")
print(q.get("total"))  # 5

default_tail — when a selector resolves to a dict, automatically follow one more selector before returning:

q = JsonInjester({"wrapper": {"value": 5}}, default_tail="value")
print(q.get("wrapper"))  # 5

When selectors resolve to nothing

from pypercache.query.json_injester import UNSET

value = q.get("missing.path")
print(value is UNSET)                              # True
print(q.get("missing.path", default_value=0))  # 0

Use has() when you just want a boolean — it's clearer than checking against None:

has_discount = q.has("discount")
discount = q.get("discount", default_value=0)

Good fit / bad fit

Reference

get() options

Option	Behavior
default_value	Return this when the path is missing or resolves to None. Without it, the result is UNSET.
select_first=True	Unwrap the first element from a list result. Returns UNSET if the list is empty.
cast=Type	Hydrate a dict result into a type or @apimodel class.

Full example

from pypercache.query import JsonInjester
from pypercache.query.json_injester import UNSET

data = {
    "meta": {"total": 3},
    "hits": [
        {"name": "Alice", "role": "staff", "score": 92},
        {"name": "Bob", "role": "guest", "score": 74},
        {"name": "Carol", "role": "staff", "score": 88},
    ],
}

q = JsonInjester(data)

print(q.get("meta.total"))                               # 3
print(q.get("hits?role=staff.name"))                      # ["Alice", "Carol"]
print(q.get("hits?name*"))                                # ["Alice", "Bob", "Carol"]
print(q.get("hits?role=staff", select_first=True))        # Alice's dict
print(q.get("hits?role=contractor", select_first=True) is UNSET)  # True

Limits

• Read-only — JsonInjester never mutates the payload.
• Integer indexing like "users.0.name" is not supported. Use select_first=True instead.
• A bare path on a list root is not supported. Use ?key* or ?key=value.
• Works on one in-memory payload at a time — no cross-record queries.

from pypercache import Cache, CacheRecord

Cache

Constructor

Default path: api-cache.pkl. Backend selected from the file extension.

Methods

store

Creates or overwrites a record. Pass cast=MyModel to store hydration metadata alongside the payload.

get

Returns the record. Raises KeyError if missing.

get_object

Hydrates the payload using the stored cast type. Raises KeyError if missing (and no default given), or AttributeError if no cast metadata was stored.

has

Checks existence only. Doesn't care whether the record is stale.

is_data_fresh

Returns True only when the key exists and the TTL has not elapsed. Never raises.

update

Replaces the payload, refreshes the timestamp, preserves expiry and cast metadata. Raises KeyError if the key doesn't exist.

completely_erase_cache

Deletes all records from the backend.

close

Closes the storage backend. This matters most for SQLite: it always closes the connection cleanly and flushes pending writes when manual flush mode is enabled. It is a no-op for other backends.

Cache.cached

Class decorator for lightweight registration. The decorated class can then be passed as cast=.

---

CacheRecord

Returned by cache.get(key).

Attributes

Attribute	Type	Description
data	any	The stored payload.
timestamp	float	Unix timestamp of the last write.
expiry	float	TTL in seconds (math.inf if none set).
cast_str	str | None	Stored cast metadata string.
cast	type | None	Lazily resolved Python type.
is_data_stale	bool	True when the record has expired.
query	JsonInjester	A JsonInjester instance over data.

Methods

update

Replaces the payload, refreshes the timestamp, and invalidates the cached query wrapper.

as_dict

Returns a serializable dict representation of the record.

from pypercache.api_wrapper import ApiHTTPError, ApiWrapper, ApiWrapperError, SSEEvent

Constructor

Methods

get_session

Override to centralize headers, auth, retries, or adapters.

request

• expected: "auto", "json", "text", or "bytes"
• Only GET requests with expected="auto" or "json" are cached
• Raises ApiHTTPError on HTTP 4xx/5xx

download_to

Downloads a bytes response and writes it to destination.

stream_sse

Parses a Server-Sent Events stream into SSEEvent objects. Does not reconnect automatically.

close

Closes the cache and, if the wrapper created the session, the session too.

---

ApiHTTPError

Raised on HTTP 4xx and 5xx responses.

Attribute	Description
status_code	HTTP status code as an integer.
url	The request URL.
body	The response body.

SSEEvent

Frozen dataclass with fields: event, data, id, retry.

from pypercache.models.apimodel import Alias, Columns, Lazy, Timestamp, apimodel
from pypercache.models.validation import ApiModelValidationError

Decorator

Can be used bare (@apimodel) or with keyword arguments (@apimodel(validate=True)).

What the decorator adds

• Registers the class in the shared class repository
• Injects a constructor that accepts one raw dict
• Adds from_dict(cls, data) classmethod
• Adds as_dict(self) instance method
• Handles eager and lazy field hydration from annotations

Field helpers

Alias

Read the field from a different raw key name.

Timestamp

Parses raw timestamp values into datetime. Handles ISO 8601 strings, numeric Unix timestamps, millisecond timestamps (unit="ms"), and explicit format strings.

Columns

Converts a dict-of-parallel-arrays payload into list[RowModel].

Lazy

Defers hydration of a field until first access. Composes with Alias, Timestamp, and Columns.

Validation

• validate=True — checks values against type annotations; raises ApiModelValidationError on mismatch
• strict=True — raises ApiModelValidationError when an annotated field is missing instead of storing UNSET

Notes

• as_dict() returns the underlying raw dict representation
• Assigning to a decorated field writes the converted value back into that raw dict
• Define models at module scope so they can be resolved reliably by the class repository

from pypercache import RequestLogger
from pypercache.storage import (
    ChunkedDictionary, ChunkedStorage, JSONStorage,
    PickleStorage, SQLiteStorage, StorageMechanism,
    get_storage_mechanism,
)

Backend selection

Returns the backend class for the given file extension.

Extension	Class
.pkl	PickleStorage
.json	JSONStorage
.manifest	ChunkedStorage
.db	SQLiteStorage

StorageMechanism

Abstract base class for custom backends. Implement these private methods:

• _impl__touch_store(filepath)
• _impl__load(filepath)
• _impl__save(cache_records_dict, filepath)
• _impl__update_record(key, data)
• _impl__erase_everything()

SQLiteStorage

Loads all rows into memory on open. Uses WAL mode. Flushes store() and update() writes immediately by default. Supports flush(), close(), enable_manual_flush_mode(), disable_manual_flush_mode(), and context manager usage. Manual flush mode is opt-in; when enabled, writes stay buffered until flush() or close().

---

RequestLogger

Default path: api_logfile.log. Writes JSONL records.

log

Appends one JSON object line to the log file.

get_logs_from_last_seconds

Returns matching records sorted oldest-first.

as_list

Returns raw record dicts.

LogRecord

Attribute	Description
timestamp	Unix timestamp of the log entry.
data	Dict with shape {"uri": "...", "status": 200}.