litewave-cache-lib 0.1.3

A lightweight Python library for Redis-backed tenant configuration storage

litewave-cache-lib

A lightweight Python library providing two complementary packages for Redis-backed tenant configuration storage:

Package	Purpose
tenant_mgr_cache	Read-only cache reader — look up tenant secrets already stored in Redis
cache_manager	Full cache manager — read from Redis and optionally auto-fetch from the tenant-manager API on a cache miss

---

Installation

pip install git+https://github.com/aiorch/litewave-cache-lib

Or install from source:

git clone https://github.com/aiorch/litewave-cache-lib.git
cd litewave-cache-lib
pip install .

Requirements: Python >= 3.10, redis >= 5.0.0, httpx >= 0.25.0, python-dotenv >= 1.0.0

---

tenant_mgr_cache

A lightweight, read-only cache reader for tenant secrets stored in Redis. Provides three simple functions covering all lookup patterns.

Features

• Connection-pooled Redis client with automatic retry on failure (5-minute cooldown)
• Fetch values by raw Redis key, or by (tenant_id, secret_name) pair via get_secret
• Attribute-level extraction from stored JSON objects via get_attribute_value_by_key
• Automatic JSON deserialization of stored values
• Standardised tenant key format: tenant:{tenant_id}:secret:{secret_name}
• Zero-boilerplate logging setup

Configuration

Settings are read from environment variables at import time (CacheSettings is instantiated on import). Always call load_dotenv() before importing from tenant_mgr_cache.

Environment Variable	Required	Default	Description
REDIS_HOST	yes	—	Redis server hostname
REDIS_PORT	yes	—	Redis server port
REDIS_PASSWORD	yes	—	Redis password
REDIS_TENANT_CONFIG_DB	no	0	Redis database index for tenant config
REDIS_MAX_CONNECTIONS	no	50	Connection pool size
REDIS_SOCKET_CONNECT_TIMEOUT	no	2	Socket connect timeout (seconds)
REDIS_SOCKET_TIMEOUT	no	2	Socket read/write timeout (seconds)
LOG_LEVEL	no	INFO	Logging level (DEBUG, INFO, WARNING, …)

Usage

Important: load_dotenv() must be called before importing tenant_mgr_cache because environment variables are resolved at module load time.

from dotenv import load_dotenv

load_dotenv(".env", override=True)

from tenant_mgr_cache import (
    get_attribute_value_by_key,
    get_secret,
    get_value_by_key,
)

Case 1 — fetch by fully-formed Redis key

Pass a complete, pre-built Redis key directly:

key = "tenant:15:secret:satoken"
value = get_value_by_key(key)
if value is not None:
    print(f"{key} → {value}")
else:
    print("Key not found or Redis unavailable.")

Case 2 — fetch by tenant ID and secret name

The canonical key tenant:{tenant_id}:secret:{secret_name} is built internally:

value = get_secret("15", "satoken")
if value is not None:
    print(f"secret → {value}")
else:
    print("Not found.")

JSON objects stored in Redis are deserialized automatically; plain strings are returned as-is.

Case 3 — fetch a single attribute from a stored JSON object

Builds the key, fetches the JSON object, and returns the value of the requested attribute:

value = get_attribute_value_by_key("15", "satoken", "api_key")
if value is not None:
    print(f"api_key → {value}")
else:
    print("Attribute not found.")

API Reference

get_value_by_key(key: str) -> Optional[Any]

Fetches the value stored at key and JSON-decodes it where possible. Returns None when the key does not exist, Redis is unavailable, or an error occurs.

get_secret(tenant_id: str, secret_name: str) -> Optional[Any]

Builds the canonical key tenant:{tenant_id}:secret:{secret_name} and returns the stored value. JSON is deserialized automatically. Returns None if the key does not exist or Redis is unavailable.

get_attribute_value_by_key(tenant_id: str, secret_name: str, attribute_name: str) -> Optional[Any]

Builds the canonical key, fetches the stored JSON object, and returns data.get(attribute_name). Falls back to getattr for non-dict objects. Returns None if the key is missing, the value is not dict-like, or the attribute is absent.

get_redis_client() -> Optional[redis.Redis]

Returns the shared, connection-pooled Redis client. Returns None if the connection is unavailable. A failed connection is retried automatically after a 5-minute cooldown; subsequent calls reuse the existing client.

prepare_key(tenant_id: str, secret_name: str) -> str

Builds the canonical Redis key:

from tenant_mgr_cache.cache_client import prepare_key

key = prepare_key("15", "satoken")
print(key)  # "tenant:15:secret:satoken"

settings — CacheSettings

A dataclass instance holding all resolved configuration values:

from tenant_mgr_cache import settings

print(settings.REDIS_HOST)
print(settings.REDIS_PORT)
print(settings.REDIS_TENANT_CONFIG_DB)

logger — logging.Logger

A pre-configured logger (litewave_cache) that respects LOG_LEVEL:

from tenant_mgr_cache import logger

logger.info("Custom message from application code")

---

cache_manager

A higher-level cache manager that wraps a RedisClient and an optional TenantService. When self_managed_cache is enabled, a cache miss automatically fetches the secret from the tenant-manager HTTP API and populates Redis (default TTL: 48 hours).

Features

• Reads from Redis first; falls back to tenant-manager API on a cache miss (when enabled)
• Auto-populates Redis after a successful tenant-manager fetch (48-hour TTL by default)
• Fully configurable via constructor arguments or environment variables
• Low-level RedisClient and TenantService classes are also exported for direct use

Configuration

Redis connection (RedisClient)

Environment Variable	Required	Default	Description
REDIS_HOST	yes	—	Redis server hostname
REDIS_PORT	yes	—	Redis server port
REDIS_PASSWORD	no	None	Redis password
REDIS_TENANT_CONFIG_DB	no	0	Redis database index
REDIS_MAX_CONNECTIONS	no	50	Connection pool size
REDIS_SOCKET_CONNECT_TIMEOUT	no	2.0	Socket connect timeout (seconds)
REDIS_SOCKET_TIMEOUT	no	2.0	Socket read/write timeout (seconds)

Cache manager behaviour

Environment Variable	Required	Default	Description
SELF_MANAGED_CACHE	no	false	Set to true to enable auto-fetch from tenant-manager

Tenant-manager API (TenantService)

Only required when self_managed_cache is True:

Environment Variable	Required	Description
TENANT_MANAGER_BASE_URL	yes	Base URL of the tenant-manager API
SEED_TENANT_TOKEN	yes	Bearer token used to authenticate requests
SEED_TENANT_ID	yes	Tenant ID sent as the x-tenant-id header

Usage

from dotenv import load_dotenv
from cache_manager import CacheManager

load_dotenv()

manager = CacheManager(self_managed_cache=True, ttl=10 * 60)
data = manager.get_secret_data("15", "azureConfiguration")

if data is None:
    print("Secret not found or tenant-manager unavailable")
else:
    print(f"Secret: {data}")

Resolution order inside get_secret_data

1. Redis cache — returns immediately on a hit.
2. Tenant-manager API — called only when self_managed_cache=True; the response is stored in Redis with the configured TTL before being returned.
3. Returns None when both sources are unavailable or the secret does not exist.

API Reference

CacheManager(redis_client?, tenant_service?, self_managed_cache?, ttl?)

Parameter	Type	Default	Description
redis_client	RedisClient	auto-constructed from env	Pre-built Redis client
tenant_service	TenantService	auto-constructed when self_managed_cache=True	Pre-built tenant service
self_managed_cache	bool	SELF_MANAGED_CACHE env var	Enable auto-fetch on cache miss
ttl	int (seconds)	172800 (48 h)	TTL applied when writing to Redis

CacheManager.get_secret_data(tenant_id: str, secret_name: str) -> Optional[Any]

Retrieves the secret for the given tenant_id / secret_name pair following the resolution order described above.

RedisClient — low-level Redis wrapper

Can be used independently when direct Redis access is needed:

from cache_manager import RedisClient

client = RedisClient()                               # all settings from env vars
client = RedisClient(host="localhost", port=6379, db=1)  # override specific params

client.set("my:key", {"foo": "bar"}, ttl=3600)      # JSON-serialized, 1-hour TTL
data = client.get("my:key")                          # {"foo": "bar"}

client.set("config:flag", "enabled")
flag = client.get("config:flag")                     # "enabled"

missing = client.get("does:not:exist")               # None

TenantService — HTTP client for the tenant-manager API

Can be used independently when direct API access is needed:

from cache_manager import TenantService

service = TenantService(
    base_url="https://tenant-manager.internal",
    seed_token="my-token",
    seed_tenant_id="1",
)
secret = service.get_secret("15", "azureConfiguration")

---

Development

Setup

pip install -r requirements.txt

Running Tests

pytest

With coverage:

pytest --cov=tenant_mgr_cache --cov-report=term-missing

---

License

MIT License. See LICENSE for details.