API Reference

from yokedcache import YokedCache
from yokedcache.config import CacheConfig

cache = YokedCache(config: CacheConfig)

# From CacheConfig
cache = YokedCache(CacheConfig(...))

# From environment variables
cache = YokedCache.from_env()

# From YAML file
cache = YokedCache.from_yaml("cache.yaml")

value = await cache.get(key: str) -> Any | None

user = await cache.get("user:42")
if user is None:
    user = await db.fetch_user(42)

await cache.set(
    key: str,
    value: Any,
    ttl: int | None = None,
    tags: list[str] | set[str] | None = None,
    serialization: SerializationMethod | None = None,
) -> bool

await cache.set("user:42", {"name": "Alice"}, ttl=300, tags=["users"])

await cache.delete(key: str) -> bool

exists = await cache.exists(key: str) -> bool

remaining = await cache.ttl(key: str) -> int | None

await cache.expire(key: str, ttl: int) -> bool

value = await cache.get_or_set(
    key: str,
    factory: Callable[[], Awaitable[Any]],
    ttl: int | None = None,
    tags: list[str] | None = None,
) -> Any

user = await cache.get_or_set(
    key=f"user:{user_id}",
    factory=lambda: db.fetch_user(user_id),
    ttl=300,
    tags=["users"],
)

results = await cache.get_many(keys: list[str]) -> dict[str, Any | None]

results = await cache.get_many(["user:1", "user:2", "user:3"])
# {"user:1": {...}, "user:2": {...}, "user:3": None}

await cache.set_many(
    mapping: dict[str, Any],
    ttl: int | None = None,
    tags: list[str] | None = None,
) -> bool

await cache.set_many({"user:1": u1, "user:2": u2}, ttl=300, tags=["users"])

await cache.delete_many(keys: list[str]) -> int

await cache.invalidate_tags(tags: list[str]) -> int

await cache.invalidate_tags(["users", "table:users"])

await cache.invalidate_pattern(pattern: str) -> int

await cache.invalidate_pattern("user:*")
await cache.invalidate_pattern("session:temp:*")

await cache.flush_all() -> bool

results = await cache.fuzzy_search(
    query: str,
    threshold: int = 80,          # similarity score 0–100
    max_results: int = 100,
    tags: set[str] | None = None,
) -> list[FuzzySearchResult]

results = await cache.fuzzy_search("alice", threshold=75, tags={"users"})
for r in results:
    print(r.key, r.score, r.value)

keys = await cache.get_keys_by_pattern(
    pattern: str,
    limit: int = 1000,
) -> list[str]

keys = await cache.get_all_keys(limit: int = 1000) -> list[str]

meta = await cache.get_meta(key: str) -> dict | None

is_healthy = await cache.health() -> bool

info = await cache.detailed_health_check() -> dict

{
    "status": "healthy",          # or "unhealthy"
    "backend_type": "redis",
    "redis_connected": True,
    "connection_pool": {
        "available": 48,
        "in_use": 2,
        "max": 50,
    },
    "circuit_breaker": {
        "state": "closed",
        "failure_count": 0,
    },
    "hit_rate": 0.87,
    "uptime_seconds": 3600,
}

stats = await cache.get_stats() -> CacheStats

cache.get_sync(key)
cache.set_sync(key, value, ttl, tags)
cache.delete_sync(key)
cache.exists_sync(key)
cache.get_many_sync(keys)
cache.set_many_sync(mapping, ttl, tags)
cache.delete_many_sync(keys)
cache.invalidate_tags_sync(tags)
cache.invalidate_pattern_sync(pattern)
cache.flush_all_sync()

from yokedcache.config import CacheConfig, TableCacheConfig

config = CacheConfig(
    tables={
        "users": TableCacheConfig(
            ttl=3600,
            tags={"user_data"},
            serialization_method=SerializationMethod.JSON,
            enable_fuzzy=True,
            fuzzy_threshold=85,
            enable_compression=False,
        ),
    }
)

from yokedcache import cached

@cached(
    cache: YokedCache,
    ttl: int | None = None,
    tags: list[str] | None = None,
    cache_key_prefix: str | None = None,
    serialization: SerializationMethod | None = None,
    single_flight: bool = False,
    serve_stale_on_error: bool = False,
    stale_ttl: int = 0,
)

@cached(cache=cache, ttl=300, tags=["users"])
async def get_user(user_id: int) -> dict:
    ...

# Access the underlying unwrapped function
raw = get_user.__wrapped__

# Inspect the generated key without calling the function
key = get_user.cache_key(user_id=42)

from yokedcache import cached_dependency

cached_dep = cached_dependency(
    dependency: Callable,
    cache: YokedCache,
    ttl: int = 300,
    table_name: str | None = None,
    tags: list[str] | None = None,
)

cached_get_db = cached_dependency(
    get_db,
    cache=cache,
    ttl=300,
    table_name="users",
)

from yokedcache.models import SerializationMethod

SerializationMethod.JSON     # default, portable, handles common Python types
SerializationMethod.PICKLE   # any Python object; requires trusted storage
SerializationMethod.MSGPACK  # binary, compact; requires pip install msgpack

from yokedcache.exceptions import (
    CacheError,              # base exception
    CacheConnectionError,    # backend connection failure
    CacheSerializationError, # serialization/deserialization failure
    CacheKeyError,           # invalid key format
    ConfigValidationError,   # invalid CacheConfig
)