Why cachekit?

Simple to use, works out of the box.

from cachekit import cache

@cache
def expensive_function():
    return fetch_data()

That's it. You get:

Quick Start

Installation

pip install cachekit

Or with uv (recommended):

uv add cachekit

Setup (choose a backend)

cachekit exposes one decorator API over a pluggable backend abstraction. Pick the backend that fits your infrastructure — they're peers behind the same @cache API:

# Run Redis locally or use your existing infrastructure
export REDIS_URL="redis://localhost:6379"

from cachekit import cache

@cache  # Auto-detects backend (defaults to Redis at localhost)
def expensive_api_call(user_id: int):
    return fetch_user_data(user_id)

More Backends

import os
from cachekit import cache

# Set your CachekitIO API key
# export CACHEKIT_API_KEY="your-api-key"  # pragma: allowlist secret

@cache.io()  # Uses CachekitIO SaaS backend — no Redis to manage
def expensive_api_call(user_id: int):
    return fetch_user_data(user_id)

from cachekit import cache
from cachekit.backends.memcached import MemcachedBackend

# pip install cachekit[memcached]

backend = MemcachedBackend()  # Defaults to 127.0.0.1:11211

@cache(backend=backend)
def expensive_api_call(user_id: int):
    return fetch_user_data(user_id)

CachekitIO Cloud (Beta) Managed caching with zero infrastructure. L1+L2 caching, circuit breaker, and automatic failover — no Redis to manage. cachekit.io is in closed beta — request access to get started.

Intent-Based Optimization

cachekit provides preset configurations for different use cases:

# Speed-critical: trading, gaming, real-time
@cache.minimal
def get_price(symbol: str):
    return fetch_price(symbol)

# Reliability-critical: payments, APIs
@cache.production
def process_payment(amount):
    return payment_gateway.charge(amount)

# Security-critical: PII, medical, financial (needs a key: master_key= or CACHEKIT_MASTER_KEY)
@cache.secure(master_key=secret_key)
def get_user_profile(user_id: int):
    return db.fetch_user(user_id)

🔒 @cache.secure forces integrity_checking=True — passing integrity_checking=False raises ConfigurationError at decoration, including as an override next to config=DecoratorConfig.secure(...). @cache.secure also rejects config=; the RORO form is @cache(config=DecoratorConfig.secure(...)).

Default TTL follows the cross-SDK intent-preset spec (@cache.io 3600 s) — the same numbers as cachekit-rs and cachekit-ts. ttl= overrides it; ttl=None is the explicit never-expire opt-in (details).

L1 SWR (within-TTL background refresh) runs only in L1-only mode (backend=None) — with a backend configured it has no effect. @cache.io additionally ships past-TTL SWR via stale_ttl (docs).

@cache.io() mirrors @cache.production (full reliability + observability) but routes to the managed CachekitIO SaaS backend instead of Redis. @cache.local() is a separate in-process path backed by ObjectCache (raw object references, entry-count LRU, no serialization) — the reliability and encryption features listed above do not apply to it.

# Development: verbose logging, integrity checks on, Prometheus off
@cache.dev
def debug_expensive_call():
    return complex_computation()

# Testing: deterministic, all protections off (no circuit breaker, no backpressure)
@cache.test
def test_cached_function():
    return fixed_test_value()

Architecture

┌─────────────────────────────────────────────────────────────┐
│                        Application                          │
├─────────────────────────────────────────────────────────────┤
│                     @cache Decorator                        │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────────────┐  │
│  │   Circuit   │  │  Adaptive   │  │    Distributed      │  │
│  │   Breaker   │  │  Timeouts   │  │      Locking        │  │
│  └─────────────┘  └─────────────┘  └─────────────────────┘  │
├─────────────────────────────────────────────────────────────┤
│  L1 Cache (In-Memory)  │  L2 Cache (Pluggable Backend)     │
│       ~50ns            │  Redis / CachekitIO / File /      │
│                        │  Memcached    ~2-50ms             │
├─────────────────────────────────────────────────────────────┤
│                    Rust Core (PyO3)                         │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────────────┐  │
│  │    LZ4      │  │  xxHash3    │  │    AES-256-GCM      │  │
│  │ Compression │  │  Checksums  │  │    Encryption       │  │
│  └─────────────┘  └─────────────┘  └─────────────────────┘  │
└─────────────────────────────────────────────────────────────┘

Features

Production Hardened

• Circuit breaker with graceful degradation
• Connection pooling with thread affinity (+28% throughput)
• Distributed locking prevents cache stampedes
• Pluggable backend abstraction (Redis, CachekitIO, File, Memcached, custom)
• Untrusted-decode bounds: nesting depth and header-declared allocation are capped on every cache read (a forged entry is a bounded cache miss), verified against the protocol's shared decode-bounds.json vectors

Smart Serialization

from cachekit.serializers import OrjsonSerializer, ArrowSerializer

# Fast JSON for API responses
@cache.production(serializer=OrjsonSerializer())
def get_api_response(endpoint: str):
    return {"status": "success", "data": fetch_api(endpoint)}

# Zero-copy DataFrames for large datasets
@cache(serializer=ArrowSerializer())
def get_large_dataset(date: str):
    return pd.read_csv(f"data/{date}.csv")

import tempfile
from cachekit.backends.file import FileBackend, FileBackendConfig
from cachekit.serializers import ArrowSerializer

calls = 0

@cache.secure(
    master_key=secret_key,
    serializer=ArrowSerializer(),
    backend=FileBackend(FileBackendConfig(cache_dir=tempfile.mkdtemp())),
)
def get_patient_data(hospital_id: int):
    global calls
    calls += 1
    return pd.DataFrame({"hospital_id": [hospital_id], "patients": [42]})

get_patient_data(7)
get_patient_data(7)  # second call is served from the encrypted cache
assert calls == 1

Integrity Checking

Performance Impact (benchmark-proven):

Security

Key rotation: keep a retiring key readable with CACHEKIT_PREVIOUS_MASTER_KEYS (comma-separated hex, max 3 decrypt-only keys) while new writes use CACHEKIT_MASTER_KEY. A one-deploy key swap is not zero-miss; follow the key rotation runbook, including its Before You Rotate checks. CK-framed entries are selected by exact key fingerprint — never trial decryption; Interop-mode entries carry no CK frame and attempt keyring keys sequentially instead.

cachekit employs comprehensive security tooling:

• Dependency Security: cargo-deny for license compliance + cargo-audit for RustSec scanning
• Formal Verification: Kani proves correctness of compression, checksums, encryption
• Runtime Analysis: Miri + sanitizers for memory safety
• Fuzzing: Coverage-guided testing with >80% code coverage
• Zero CVEs: Continuous vulnerability scanning

make security-install  # Install security tools (one-time)
make security-fast     # Run fast checks (< 3 min)

Monitoring & Observability

• Per-function statistics - cache_info() on every decorated function, modelled on functools.lru_cache
• Prometheus metrics - Recorded by default (your app owns exposition)
• Structured logging - Context-aware, per-operation fields
• Health checks - Comprehensive status endpoints

Every decorated function exposes cache_info(), returning a CacheInfo named tuple with hit/miss counts, the L1/L2 split, and average backend latency:

@cache()
def get_score(x):
    return x ** 2

get_score(2)
get_score(2)  # served from cache

info = get_score.cache_info()
# CacheInfo has 9 fields: hits, misses, l1_hits, l2_hits, maxsize,
# currsize, l2_avg_latency_ms, last_operation_at, session_id
assert info.l1_hits + info.l2_hits == info.hits  # every hit is L1 or L2

maxsize and currsize are always None (the cache lives in an external store, not a bounded in-process dict); they exist only for lru_cache API parity. See the API Reference for the full field reference and a sample stats endpoint, and the Prometheus Metrics guide for metric names and exposition setup.

from concurrent.futures import ThreadPoolExecutor

@cache()
def expensive_func(x):
    return x ** 2

# All threads share same stats
with ThreadPoolExecutor(max_workers=10) as executor:
    results = list(executor.map(expensive_func, range(100)))

info = expensive_func.cache_info()
# CacheInfo(hits=..., misses=..., l1_hits=..., l2_hits=...,
#           maxsize=None, currsize=None, l2_avg_latency_ms=...,
#           last_operation_at=..., session_id=...)

Documentation

Start Here

Feature Deep Dives

Configuration

Environment Variables

# Redis Connection (priority: CACHEKIT_REDIS_URL > REDIS_URL)
CACHEKIT_REDIS_URL="redis://localhost:6379"  # Primary (preferred)
REDIS_URL="redis://localhost:6379"           # Fallback

# CachekitIO SaaS Backend (closed beta — request access at cachekit.io)
CACHEKIT_API_KEY="your-api-key"             # For @cache.io() — or pass api_key= directly  # pragma: allowlist secret
CACHEKIT_API_URL="https://api.cachekit.io"  # Default SaaS endpoint

# Memcached Backend (optional: pip install cachekit[memcached])
CACHEKIT_MEMCACHED_SERVERS='["mc1:11211", "mc2:11211"]'  # Default: 127.0.0.1:11211
CACHEKIT_MEMCACHED_CONNECT_TIMEOUT=2.0                   # Default: 2.0 seconds
CACHEKIT_MEMCACHED_TIMEOUT=1.0                            # Default: 1.0 seconds
CACHEKIT_MEMCACHED_KEY_PREFIX="myapp:"                    # Default: "" (none)

# Optional Configuration
CACHEKIT_MAX_VALUE_SIZE=104857600
CACHEKIT_ARROW_COMPRESSION=zstd

Development

git clone https://github.com/cachekit-io/cachekit-py.git
cd cachekit-py
uv sync && make install
make quick-check  # format + lint + critical tests

See CONTRIBUTING.md for full development guidelines.

Requirements

License

MIT License - see LICENSE for details.