Skip to content

reference-caching.md

Latest commit

 

History

History
317 lines (229 loc) · 8.7 KB

reference-caching.md

File metadata and controls

317 lines (229 loc) · 8.7 KB
Raw

Home › Features › Reference Caching

Reference Caching with @cache.local()

Available since v0.6.0

TL;DR

Reference caching (@cache.local(), a Python-only, SDK-local preset) caches opaque, non-serializable objects — things that can't be converted to bytes. Perfect for SDK connections, ML models, database connections, and language runtime objects that must maintain identity.

from cachekit import cache

@cache.local()
def get_langfuse_client():
    """Returns same instance for identical args."""
    return Langfuse(project="my-project")

client1 = get_langfuse_client()
client2 = get_langfuse_client()
assert client1 is client2  # Same object by reference

When to Use

Use @cache.local() when:

Scenario Example Why
SDK connections Langfuse, httpx.Client, gRPC stubs Maintain session state, reuse TCP connection
ML models Loaded transformers, embedders In-memory weight matrices, can't be serialized
Database connections SQLAlchemy session, asyncpg connection Stateful, must be re-created per process
Language runtime objects Java objects via jpype, R objects via rpy2 Opaque to Python, cross-language marshalling complex

NOT for: Serializable data (dicts, dataframes, JSON). Use @cache for those.

Quick Start

from cachekit import cache

# Cache a connection pool (same instance reused)
@cache.local()
def get_database_client(host: str):
    return sqlalchemy.create_engine(f"postgresql://{host}/mydb")

# First call: creates client
db1 = get_database_client("localhost")

# Second call with same args: returns cached instance
db2 = get_database_client("localhost")

assert db1 is db2  # Guaranteed same object

Configure TTL and size:

@cache.local(ttl=600, max_entries=128)
def get_model(model_name: str):
    return transformers.AutoModel.from_pretrained(model_name)

Mutation Warning ⚠️

Critical: Cached objects are returned by reference. Mutations affect all callers.

from cachekit import cache

@cache.local()
def get_config_dict():
    return {"timeout": 30, "retries": 3}

config1 = get_config_dict()
config1["timeout"] = 999  # Mutation!

config2 = get_config_dict()
assert config2["timeout"] == 999  # 999, not 30 — same object!

Fix: Copy if you need to mutate:

import copy
from cachekit import cache

@cache.local()
def get_safe_config():
    return {"timeout": 30, "retries": 3}

config = copy.copy(get_safe_config())  # Shallow copy
config["timeout"] = 999  # Safe — original unchanged
assert get_safe_config()["timeout"] == 30

Identity Semantics

Same args = same object:

from cachekit import cache

@cache.local()
def get_client(api_key: str):
    return {"key": api_key, "session": object()}

client_a = get_client("key123")
client_b = get_client("key123")
assert client_a is client_b  # True (identity check)

client_c = get_client("key456")
assert client_a is not client_c  # Different args, different object

Feature for connection reuse:

  • Multiple callers within same process reuse same socket, session state, credentials
  • Reduces memory overhead, avoids re-authentication

Footgun for mutable data:

  • Mutations visible to all callers
  • Use @cache (which serializes) if you need isolation
  • Or manually copy on retrieval

Concurrent misses share one call

Calls that miss the same key at the same time (a cold key, or one past its ttl) run the function once, and every one of them gets that call's object. This holds across threads for a sync function and across tasks on one event loop for an async one. The same-object guarantee above therefore holds for a concurrent cold herd too, not only for calls made after the first one returns.

Failure, cancellation, invalidation, cache_info() counting and the caveats work as described in in-process single-flight.

Object Lifecycle

Cached objects are held strongly until eviction:

Function call with args A
  ↓
[Check cache]
  ├─ Hit: Return cached object (reference count unchanged)
  └─ Miss: Call function (once for every concurrent miss on the key), store result
  ↓
[Store in LRU cache (max_entries)]
  ↓
[LRU eviction when full]
  └─ Oldest/least-used entry removed
  ↓
[Object eligible for garbage collection if no other refs]

Strong reference guarantee: Cached object won't be garbage-collected until:

  1. Evicted from cache (LRU), OR
  2. Cache cleared via cache_clear(), OR
  3. Function invalidated via invalidate_cache()

Parameters

Parameter Type Default Description
ttl int 300 Time-to-live in seconds. Objects evicted after TTL expires.
max_entries int 256 Maximum cache entries. Oldest removed when full.
namespace str None Optional key namespace for multi-tenant isolation.
key callable None Custom key function. Default: (args, kwargs) hash.

Wrapper API

All cachekit decorators expose the same cache management interface:

from cachekit import InvalidationOutcome, cache

@cache.local(ttl=300)
def get_client(api_key: str):
    return {"key": api_key}

get_client("key1")  # miss
get_client("key1")  # hit

info = get_client.cache_info()
assert info.hits == 1
assert info.misses == 1
assert info.maxsize == 256
assert info.currsize == 1

# Remove specific entry: there is no L2, so the outcome is always NO_L2
assert get_client.invalidate_cache("key1") is InvalidationOutcome.NO_L2
assert get_client.cache_info().currsize == 0

get_client("key2")
get_client.cache_clear()  # Remove all entries
assert get_client.cache_info().currsize == 0

raw = get_client.__wrapped__("key3")  # Bypass cache
assert get_client.cache_info().currsize == 0  # Not cached

Async variant:

@cache.local()
async def get_async_client(api_key: str):
    return await create_client(api_key)

await get_async_client.ainvalidate_cache("my-api-key")

Comparison

Feature @cache.local() functools.lru_cache cachetools.TTLCache
In-process only ✅ ✅ ✅
Distributed ❌ ❌ ❌
TTL support ✅ ❌ ✅
Unhashable args (dicts, lists) ✅ ❌ ❌
Async functions ✅ ❌ ✅
Per-key invalidation ✅ ❌ ❌
Thread-safe ✅ ✅ ❌ (needs lock)
Hit/miss statistics ✅ ✅ ❌ (size only)

Why @cache.local() wins:

  • Accepts unhashable args (no need to convert to strings)
  • TTL + LRU (best of both worlds)
  • Async-native, not a decorator shim
  • Per-key invalidation without clearing entire cache
  • Thread-safe by default

Future: Lifecycle Callbacks (v0.7)

Planned (not yet available):

def on_evict(key, value):
    """Called when cache entry is evicted."""
    value.cleanup()  # Close connection, free memory, etc.

@cache.local(on_evict=on_evict)
def get_connection():
    return Database.connect()

For now, manually call cleanup when needed:

get_connection.cache_clear()  # All entries evicted, consider calling cleanup

Examples

ML model caching:

import torch
from cachekit import cache

@cache.local(ttl=3600, max_entries=4)
def load_model(model_name: str):
    """Load once, reuse across requests."""
    return torch.hub.load("pytorch/vision", model_name, pretrained=True)

# First inference: loads model (slow)
embeddings1 = load_model("resnet50")(image1)

# Second inference: reuses model (fast)
embeddings2 = load_model("resnet50")(image2)

Database session pool:

from sqlalchemy.orm import Session
from cachekit import cache

@cache.local()
def get_session(db_url: str) -> Session:
    """Reuse SQLAlchemy session per connection string."""
    engine = sqlalchemy.create_engine(db_url)
    return Session(engine)

# Multiple queries reuse same session
result1 = get_session("postgresql://...").query(User).all()
result2 = get_session("postgresql://...").query(Product).all()

Async HTTP client:

import httpx
from cachekit import cache

@cache.local()
async def get_http_client(api_key: str):
    """Reuse HTTP client for connection pooling."""
    return httpx.AsyncClient(
        headers={"Authorization": f"Bearer {api_key}"},
        http2=True
    )

# Both requests use same connection pool
response1 = await (await get_http_client("key123")).get("https://api.example.com/users")
response2 = await (await get_http_client("key123")).get("https://api.example.com/products")