litewave-cache-lib 0.1.2

A lightweight Python library for Redis-backed tenant configuration storage

litewave-cache-lib

A lightweight Python library that provides a connection-pooled, fault-tolerant Redis client with automatic retry logic and JSON-aware key lookup helpers for tenant configuration storage.

---

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
• Automatic JSON deserialization of stored values
• Attribute-level extraction from stored JSON objects via get_attribute_value_by_key
• Standardised tenant key format: tenant:{tenant_id}:secret:{secret_name}
• Fully configurable via environment variables
• Zero-boilerplate logging setup

---

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
• python-dotenv >= 1.0.0

---

Configuration

All settings are read from environment variables at import time. REDIS_HOST, REDIS_PORT, and REDIS_PASSWORD are required — the library raises a ValueError at startup if any of them are missing.

Environment Variable	Default	Description
REDIS_HOST	(required)	Redis server hostname
REDIS_PORT	(required)	Redis server port
REDIS_PASSWORD	(required)	Redis password
REDIS_DB	0	Default Redis database index
REDIS_COORDINATION_DB	1	Redis DB reserved for coordination workloads
REDIS_TENANT_CONFIG_DB	2	Redis DB used for tenant configuration
REDIS_MAX_CONNECTIONS	50	Connection pool size
REDIS_SOCKET_CONNECT_TIMEOUT	2	Socket connect timeout in seconds
REDIS_SOCKET_TIMEOUT	2	Socket read/write timeout in seconds
LOG_LEVEL	INFO	Logging level (DEBUG, INFO, WARNING, …)

A .env file is supported via python-dotenv. Load it before importing the library:

from dotenv import load_dotenv
load_dotenv(".env", override=True)

from tenant_mgr_cache import get_redis_client

---

Usage

Get the Redis client

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.

from tenant_mgr_cache import get_redis_client

client = get_redis_client()
if client:
    client.set("tenant:15:secret:satoken", "abc123")
    print(client.get("tenant:15:secret:satoken"))  # "abc123"

---

get_secret — fetch by tenant ID and secret name

Builds the canonical key tenant:{tenant_id}:secret:{secret_name} internally and returns the stored value. JSON values are deserialized automatically; plain strings are returned as-is.

from tenant_mgr_cache import get_secret

# JSON object stored → returned as dict
config = get_secret("15", "db_config")
if isinstance(config, dict):
    print(config["host"])   # e.g. "db.prod.internal"
    print(config["port"])   # e.g. 5432

# Plain string stored → returned as-is
token = get_secret("15", "satoken")
print(token)                # e.g. "abc123"

# Key does not exist → None
missing = get_secret("15", "nonexistent")
print(missing)              # None

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, Redis is unavailable, or any error occurs.

---

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

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

---

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

Builds the canonical key for tenant_id + secret_name, 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.

---

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

Builds the canonical Redis key for a tenant secret:

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. Import it to inspect or override settings programmatically:

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 the LOG_LEVEL environment variable. Import it to emit log messages consistent with the library's format:

from tenant_mgr_cache import logger

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

---

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.