cache_core

Multi-backend cache abstraction layer for Python — async-first, with pluggable clients for Redis and in-memory storage.

Installation

# Core (in-memory client — no external dependencies)
pip install cache_core

# With Redis support
pip install "cache_core[redis]"

# Or install in development mode from source
pip install -e .

Quick Start

import asyncio
from cache_core import CacheBuilder, create_memory_cache_client

async def main():
    cache = CacheBuilder(create_memory_cache_client())

    # Simple key-value
    await cache.key("greeting").set("hello")
    print(await cache.key("greeting").get())  # "hello"

    # Complex types auto-serialized
    await cache.key("user:1").set({"name": "Alice", "roles": ["admin"]}, ttl=3600)
    user = await cache.key("user:1").get()
    print(user["name"])  # "Alice"

asyncio.run(main())

Usage Examples

Key-Value Operations

# Set and get
await cache.key("count").set(42)
count = await cache.key("count").get()          # 42 (int)

await cache.key("name").set("Alice")
name = await cache.key("name").get()            # "Alice" (str)

# TTL (time-to-live in seconds)
await cache.key("session").set("data", ttl=300)
print(await cache.key("session").ttl())         # ~300

# Existence and deletion
exists = await cache.key("count").exists()      # True
deleted = await cache.key("count").delete()     # True
deleted = await cache.key("nope").delete()      # False

# Atomic increment / decrement
await cache.key("visits").incr()                # 1
await cache.key("visits").incr(5)               # 6
await cache.key("visits").decr(2)               # 4

Complex Types — Automatic JSON Serialization

Primitives (str, int, float, bool, None) pass through as-is. Complex types (dict, list, set, tuple) are automatically JSON-encoded on write and decoded on read.

# Dicts
await cache.key("config").set({"theme": "dark", "items": [1, 2, 3]})
config = await cache.key("config").get()        # {"theme": "dark", "items": [1, 2, 3]}

# Lists
await cache.key("ids").set([10, 20, 30])
ids = await cache.key("ids").get()              # [10, 20, 30]

# Nested structures
await cache.key("nested").set({
    "users": [
        {"id": 1, "tags": ["python", "async"]},
        {"id": 2, "tags": ["rust"]},
    ]
})

Namespace Scoping

Prefix keys with namespaces to organize your cache and avoid collisions.

users = cache.ns("users")
posts = cache.ns("posts")

await users.key("123").set({"name": "Alice"})
await users.key("456").set({"name": "Bob"})
await posts.key("1").set({"title": "Hello World"})

# Keys: "users:123", "users:456", "posts:1"

alice = await cache.key("users:123").get()      # {"name": "Alice"}
alice = await users.key("123").get()            # same result

# Nested namespaces
blog = cache.ns("blog").ns("comments")
await blog.key("post:1").set("Great post!")      # key: "blog:comments:post:1"

Lists (Arrays)

Each element is automatically serialized/deserialized individually.

# Push to the right (append)
await cache.list("queue").push("task_a", "task_b", "task_c")
await cache.list("queue").push({"id": 4, "type": "email"})

# Push to the left (prepend)
await cache.list("queue").push_left("urgent")

# Pop from the right (LIFO)
item = await cache.list("queue").pop()          # {"id": 4, "type": "email"}

# Pop from the left (FIFO)
item = await cache.list("queue").pop_left()     # "urgent"

# Range and length
items = await cache.list("queue").range(0, -1)  # all items
size = await cache.list("queue").len()          # 3

Sets

Each member is automatically serialized. Duplicates are ignored (deduplicated by JSON value).

# Add members — duplicates are silently ignored
added = await cache.set_("tags").add(
    {"name": "python"}, {"name": "rust"}, {"name": "python"}
)
print(added)  # 2 (third was a duplicate)

# Retrieve all members
members = await cache.set_("tags").members()
# [{"name": "python"}, {"name": "rust"}]

# Membership check
has_python = await cache.set_("tags").contains({"name": "python"})  # True

# Remove members
removed = await cache.set_("tags").remove({"name": "rust"})         # 1

# Cardinality
size = await cache.set_("tags").size()          # 1

Hashes (Maps)

Store field-value pairs under a single key. Each value is auto-serialized.

# Bulk set fields
await cache.hash("users").set({
    "alice": '{"name":"Alice","age":30}',
    "bob": '{"name":"Bob","age":25}',
})

# Get all fields
all_users = await cache.hash("users").getall()
# {"alice": {"name": "Alice", "age": 30}, "bob": {"name": "Bob", "age": 25}}

# Single field operations
await cache.hash("users").field("alice").set('{"name":"Alice","age":31}')
user = await cache.hash("users").field("alice").get()
# {"name": "Alice", "age": 31}

exists = await cache.hash("users").field("bob").exists()   # True
deleted = await cache.hash("users").field("bob").delete()   # 1

# List fields
fields = await cache.hash("users").keys()      # ["alice"]
count = await cache.hash("users").len()         # 1

Sorted Sets

Members ordered by score.

# Add members with scores
await cache.sorted_set("leaderboard").add({
    "alice": 1500.0,
    "bob": 1200.0,
    "carol": 1800.0,
})

# Range by score (ascending)
top = await cache.sorted_set("leaderboard").range(0, -1)
# ["bob", "alice", "carol"]

# With scores
top_scored = await cache.sorted_set("leaderboard").range(0, -1, withscores=True)
# [("bob", 1200.0), ("alice", 1500.0), ("carol", 1800.0)]

# Remove and size
await cache.sorted_set("leaderboard").remove("bob")
size = await cache.sorted_set("leaderboard").size()   # 2

Pattern-Based Batch Operations

Delete or list keys matching a glob pattern.

# Seed some keys
await cache.key("session:a").set("x")
await cache.key("session:b").set("y")
await cache.key("cache:config").set("z")

# Count matching keys
count = await cache.pattern("session:*").count()   # 2

# List matching keys
keys = await cache.pattern("session:*").keys()     # ["session:a", "session:b"]

# Batch delete
deleted = await cache.pattern("session:*").delete()  # 2

Async Context Manager

from cache_core import CacheBuilder, create_memory_cache_client

async with CacheBuilder(create_memory_cache_client()) as cache:
    await cache.key("temp").set("value")
    # ... do work ...
# Automatically disconnects on exit

Complete Usage Example

import asyncio
from cache_core import CacheBuilder, create_memory_cache_client


async def main():
    # Create client and builder (in-memory for testing; swap for Redis in production)
    cache = CacheBuilder(create_memory_cache_client())

    # 1. Cache a user profile with TTL
    user = {"id": 1, "name": "Alice", "email": "alice@example.com", "roles": ["admin"]}
    await cache.key("user:1").set(user, ttl=3600)

    # 2. Maintain a task queue
    tasks = cache.list("queue:pending")
    await tasks.push(
        {"task_id": "t1", "action": "send_email", "to": "alice@example.com"},
        {"task_id": "t2", "action": "generate_report", "user_id": 1},
    )

    # 3. Track unique tags for a post
    post_tags = cache.set_("post:42:tags")
    await post_tags.add("python", "async", "caching")
    await post_tags.add("python")  # duplicate — ignored

    # 4. Store configuration as a hash
    config = cache.hash("app:config")
    await config.set({
        "theme": "dark",
        "language": "en",
        "features": '{"search":true,"chat":false}',
    })

    # 5. Leaderboard with scores
    lb = cache.sorted_set("game:scores")
    await lb.add({"alice": 9500, "bob": 8200, "carol": 9900})

    # --- Read back ---
    print("User:", await cache.key("user:1").get())
    # User: {'id': 1, 'name': 'Alice', 'email': 'alice@example.com', 'roles': ['admin']}

    print("Next task:", await tasks.pop())
    # Next task: {'task_id': 't2', 'action': 'generate_report', 'user_id': 1}

    print("Tags:", await post_tags.members())
    # Tags: ['python', 'async', 'caching']

    print("Config:", await config.getall())
    # Config: {'theme': 'dark', 'language': 'en', 'features': {'search': True, 'chat': False}}

    print("Leaderboard:", await lb.range(0, -1, withscores=True))
    # Leaderboard: [('bob', 8200.0), ('alice', 9500.0), ('carol', 9900.0)]

    # 6. Cleanup session keys
    print("Session keys cleaned:", await cache.pattern("session:*").delete())

    # 7. Flush everything
    await cache.flush()


if __name__ == "__main__":
    asyncio.run(main())

Redis Backend

import os
from cache_core import CacheBuilder, create_redis_cache_client

# From URL
client = create_redis_cache_client(url="redis://localhost:6379/0")

# From environment variables: REDIS_URL, REDIS_HOST, REDIS_PORT, REDIS_DB, REDIS_PASSWORD
client = create_redis_cache_client()

cache = CacheBuilder(client)
await cache.key("hello").set("world")

Custom Serializer

from cache_core import CacheBuilder, PickleSerializer, PassthroughSerializer, create_memory_cache_client

# Pickle — preserves full Python objects (Python-only)
cache = CacheBuilder(create_memory_cache_client(), serializer=PickleSerializer())

# Pass-through — no serialization, caller handles it
cache = CacheBuilder(create_memory_cache_client(), serializer=PassthroughSerializer())

API Reference

CacheBuilder

KeyOperation

ListOperation

SetOperation

HashOperation

HashFieldOperation

SortedSetOperation

PatternOperation