Key-Value Stores

Overview

The Key-Value (KV) Store module provides a powerful and extensible abstraction layer for interacting with key-value data stores like Redis. It is a fundamental building block used by many other Athomic modules, including:

• Caching: For storing the results of expensive operations.
• Rate Limiting: For tracking request counts.
• Feature Flags: For fetching flag configurations.
• Distributed Locking: For managing lock states.
• Sagas & Schedulers: For persisting state and scheduled tasks.

The architecture is designed to be highly flexible, featuring a protocol-based design, multiple backend providers, and a powerful wrapper system that allows for adding cross-cutting functionality declaratively.

---

Core Concepts

KVStoreProtocol

This is the contract that all KV store providers must implement. It defines a rich set of asynchronous operations, including not only basic CRUD (get, set, delete, exists) but also advanced commands for sorted sets (zadd, zrem, zpopbyscore), which are critical for implementing components like the task scheduler.

KVStoreFactory

This is the main entry point for creating a KV store client. The factory is responsible for: 1. Reading the configuration for a named connection. 2. Instantiating the correct base provider (e.g., RedisKVClient) using a registry. 3. Applying a chain of wrappers to the base provider, adding functionality like multi-tenancy and default TTLs. 4. Applying Fallback Topologies, orchestrating highly resilient multi-layer caches directly within the domain using secondary KV connections.

Wrapper (Decorator) Pattern

A key architectural feature of the KV Store module is its use of the Decorator pattern. You can "wrap" a base client with additional functionality defined in your configuration. This makes the system extremely flexible. For example, you can add multi-tenant key resolution to any KV store (Redis, in-memory, or a future DynamoDB provider) just by adding a wrapper to the configuration.

---

Available Providers

• RedisKVClient: The primary, production-ready provider for Redis. It uses the high-performance redis-py async client and supports the full KVStoreProtocol, including atomic sorted set operations implemented with Lua scripting for maximum safety.
• LocalKVClient: A simple, in-memory provider that simulates a KV store using Python dictionaries. It is perfect for fast, dependency-free unit and integration testing.

Available Wrappers

• KeyResolvingKVClient: This wrapper automatically prepends contextual prefixes to all keys before they hit the database. It uses the ContextKeyGenerator to include the namespace, tenant_id, and user_id (if configured), making multi-tenancy transparent.
• DefaultTTLKvClient: This wrapper automatically applies a default Time-To-Live (TTL) to all set operations if one is not explicitly provided in the method call. This is useful for ensuring that cache keys, for example, always expire.

---

Fallback Orchestration

The KV Store module completely encapsulates its own highly resilient multi-layer caching topologies via the FallbackKVProvider. If a primary provider (e.g., Redis) is unavailable or encounters a cache miss, the system can seamlessly shift reads and writes to a secondary provider (e.g., in-memory) without leaking domain complexity into the application's generic resilience layers.

Key Features

• Read Strategies: Define when to trigger the fallback chain. Options include on_error (only on network/connection failure) or on_miss_or_error (fallback if the primary fails OR the key is not found).
• Write Strategies: Define how writes propagate. write_around (writes only hit the primary provider) or write_through (writes hit the primary and propagate concurrently to all fallbacks).
• Self-Healing: If a value is retrieved from a fallback cache, a background task automatically writes it back to the primary, "healing" the cache layer for future requests.

---

Configuration

The true power of this module is visible in its configuration. You define a base provider, apply wrappers, and optionally set up fallback topologies entirely through configuration.

[default.database.kvstore.default_redis]
enabled = true
namespace = "cache"

  [default.database.kvstore.default_redis.provider]
  backend = "redis"
  uri = "redis://localhost:6379/0"

  [[default.database.kvstore.default_redis.wrappers]]
  name = "default_ttl"
  enabled = true
    [default.database.kvstore.default_redis.wrappers.config]
    default_ttl_seconds = 3600 

  [[default.database.kvstore.default_redis.wrappers]]
  name = "key_resolver"
  enabled = true

  # Configures Fallback Topologies directly within KVStore settings
  [default.database.kvstore.default_redis.fallback]
  enabled = true
  read_strategy = "on_miss_or_error"
  write_strategy = "write_around"
  enable_self_healing = true
  connections = ["local_memory_cache"]

---

API Reference

nala.athomic.database.kvstore.protocol.KVStoreProtocol

Bases: BaseServiceProtocol, Protocol

Defines the contract for an asynchronous Key-Value store provider.

This protocol specifies the standard interface for all key-value store implementations within the framework. It includes basic CRUD operations, sorted set commands, and lifecycle management. Any class that conforms to this protocol can be used as a KV store backend for features like caching, batch operate, rate limiting, and feature flags.

acquire_budget(key, reservation_id, cost, limit, ttl_seconds) async

Atomically reserves capacity under a shared, TTL'd budget.

Expired reservations (older than their own ttl_seconds) are reaped first. If the sum of cost plus all remaining non-expired reservations would exceed limit, no reservation is made and the call returns False. Otherwise a reservation of cost units is recorded under reservation_id, expiring automatically after ttl_seconds even if release_budget is never called.

Parameters:

Name	Type	Description	Default
key	str	The key identifying the shared budget.	required
reservation_id	str	A unique identifier for this reservation.	required
cost	int	The number of units to reserve.	required
limit	int	The total capacity of the shared budget.	required
ttl_seconds	int	How long the reservation remains valid if not explicitly released.	required

Returns:

Type	Description
bool	True if the reservation was admitted, False if it would exceed
bool	limit.

clear() async

Removes all keys from the current database.

close() async

Closes the connection to the backend service.

connect() async

Establishes the connection to the backend service.

delete(key) async

Deletes a key from the store.

Parameters:

Name	Type	Description	Default
key	str	The key to delete.	required

delete_many(keys) async

Deletes multiple keys in a single operation.

Parameters:

Name	Type	Description	Default
keys	List[str]	A list of keys to delete.	required

Returns:

Type	Description
int	The number of keys actually deleted.

exists(key) async

Checks if a key exists in the store.

Parameters:

Name	Type	Description	Default
key	str	The key to check.	required

Returns:

Type	Description
bool	True if the key exists, False otherwise.

get(key) async

Retrieves a value by its key.

Parameters:

Name	Type	Description	Default
key	str	The key of the item to retrieve.	required

Returns:

Type	Description
Optional[Any]	The deserialized value, or None if the key is not found.

get_final_client() async

Returns the underlying raw asynchronous client instance.

This provides an escape hatch to access provider-specific features not covered by the protocol (e.g., calling a unique Redis command).

Returns:

Type	Description
Any	The raw provider-specific async client instance (e.g., redis.asyncio.Redis).

get_sync_client()

Returns an underlying raw synchronous client instance, if applicable.

Returns:

Type	Description
Any	The raw provider-specific sync client instance (e.g., redis.Redis).

Raises:

Type	Description
NotImplementedError	If a synchronous client is not available or supported by the provider.

hdel(key, fields) async

Removes the specified fields from the hash stored at key.

hgetall(key) async

Returns all fields and values of the hash stored at key.

hset(key, field, value) async

Sets field in the hash stored at key to value.

increment(key, amount=1) async

Atomically increments the integer value of a key by a given amount.

If the key does not exist, it is set to 0 before performing the operation.

Parameters:

Name	Type	Description	Default
key	str	The key of the item to increment.	required
amount	int	The amount to increment by. Defaults to 1.	1

Returns:

Type	Description
int	The value of the key after the increment.

is_available() async

Checks if the provider is connected and ready to accept operations.

Returns:

Type	Description
bool	True if the service is ready, False otherwise.

release_budget(key, reservation_id) async

Releases a reservation previously made via acquire_budget.

No-op if the reservation does not exist (e.g. it already expired and was reaped).

Parameters:

Name	Type	Description	Default
key	str	The key identifying the shared budget.	required
reservation_id	str	The identifier of the reservation to release.	required

set(key, value, ttl=None, nx=False) async

Sets a key-value pair, with an optional TTL.

Parameters:

Name	Type	Description	Default
key	str	The key of the item to set.	required
value	Any	The value to store. It will be serialized by the provider.	required
ttl	Optional[int]	Optional time-to-live for the key in seconds.	None
nx	bool	If True, set the key only if it does not already exist.	False

Returns:

Type	Description
bool	True if the operation was successful.

set_many(mapping, ttl=None, nx=False) async

Sets multiple key-value pairs in a single operation.

Parameters:

Name	Type	Description	Default
mapping	Dict[str, Any]	A dictionary of key-value pairs to store.	required
ttl	Optional[int]	Optional TTL in seconds applied to all keys in this batch.	None
nx	bool	If True, only sets keys that do not exist (atomic per key).	False

Returns:

Type	Description
Dict[str, bool]	A dictionary mapping each key to a boolean indicating success.

zadd(key, mapping) async

Adds one or more members with scores to a sorted set.

Parameters:

Name	Type	Description	Default
key	str	The key of the sorted set.	required
mapping	Mapping[str, float]	A dictionary of member-score pairs to add.	required

zpopbyscore(key, max_score) async

Atomically removes and returns the member with the lowest score.

The member returned is the one with the lowest score that is less than or equal to the max_score.

Parameters:

Name	Type	Description	Default
key	str	The key of the sorted set.	required
max_score	float	The maximum score to consider for popping a member.	required

Returns:

Type	Description
Optional[str]	The member name as a string, or None if no qualifying members exist.

zrangebyscore(key, min_score, max_score) async

Returns members of a sorted set within a score range.

Retrieves all members in a sorted set with scores between min_score and max_score (inclusive).

Parameters:

Name	Type	Description	Default
key	str	The key of the sorted set.	required
min_score	float	The minimum score of the range.	required
max_score	float	The maximum score of the range.	required

Returns:

Type	Description
list[str]	A list of member names.

zrem(key, members) async

Removes one or more members from a sorted set.

Parameters:

Name	Type	Description	Default
key	str	The key of the sorted set.	required
members	list[str]	A list of members to remove.	required

Returns:

Type	Description
int	The number of members that were removed from the set.

nala.athomic.database.kvstore.factory.KVStoreFactory

Bases: FactoryProtocol

Factory responsible for instantiating and orchestrating KVStore clients.

create(settings) classmethod

Creates a KVStore client based on the provided settings.

Parameters:

Name	Type	Description	Default
settings	KVStoreSettings	The global configuration for the KVStore.	required

Returns:

Type	Description
KVStoreProtocol	An instance implementing the KVStoreProtocol, fully decorated.

Raises:

Type	Description
ValueError	If a provider or wrapper is not found in the registry.
TypeError	If a provider does not comply with the protocol.

nala.athomic.database.kvstore.providers.redis.client.RedisKVClient

Bases: BaseKVStore, CredentialResolve

A concrete KVStoreProvider implementation for Redis using the aioredis library.

This class manages the asynchronous connection lifecycle, handles low-level Redis commands (including atomic operations via Lua scripts), and securely resolves connection credentials.

Attributes:

Name	Type	Description
settings	RedisSettings	The Redis-specific configuration settings.
_client	Redis | None	The underlying asynchronous Redis client instance.

__init__(settings=None, serializer=None)

Initializes the RedisKVClient.

Parameters:

Name	Type	Description	Default
settings	Optional[KVStoreSettings]	The configuration for this KV store instance.	None
serializer	Optional[SerializerProtocol]	The serializer used for values.	None

get_final_client() async

Returns the raw asynchronous Redis client instance (Public Contract).

get_sync_client()

Creates and returns a NEW instance of the synchronous Redis client (Public Contract).

This instance is not managed by the provider's lifecycle.

Raises:

Type	Description
ValueError	If the Redis URI is not available for connection string construction.
RuntimeError	If the synchronous client creation fails.

nala.athomic.database.kvstore.wrappers.key_resolver_kv_client.KeyResolvingKVClient

Bases: WrapperBase

A decorator (Wrapper) for the KVStore client that dynamically resolves keys using a ContextKeyGenerator.

This wrapper applies a contextual prefix (e.g., namespace, tenant ID, user ID) to the logical key provided by the caller before forwarding the operation to the underlying KV store. This is essential for supporting multi-tenancy and avoiding key collisions in shared storage systems.

Attributes:

Name	Type	Description
resolver	ContextKeyGenerator	The utility used to generate context-aware keys.

__init__(client, settings=None, wrapper_settings=None, context_settings=None)

Initializes the KeyResolvingKVClient and sets up the ContextKeyGenerator.

Parameters:

Name	Type	Description	Default
client	KVStoreProtocol	The underlying KVStore client being wrapped.	required
settings	Optional[KVStoreSettings]	The global KVStore settings.	None
wrapper_settings	Optional[KeyResolvingWrapperSettings]	Specific settings for this wrapper.	None
context_settings	Optional[ContextSettings]	Optional settings for the key generator context.	None

delete(key) async

Deletes a key after resolving the contextual key.

delete_many(keys) async

Deletes multiple keys after resolving their contextual keys.

Parameters:

Name	Type	Description	Default
keys	List[str]	The list of logical keys to delete.	required

Returns:

Name	Type	Description
int	int	The number of keys successfully deleted.

exists(key) async

Checks for key existence after resolving the contextual key.

get(key) async

Retrieves a value after resolving the contextual key.

hdel(key, fields) async

Delegates the hdel operation after resolving the contextual key.

hgetall(key) async

Delegates the hgetall operation after resolving the contextual key.

hset(key, field, value) async

Delegates the hset operation after resolving the contextual key.

set(key, value, ttl=None, nx=False) async

Sets a value after resolving the contextual key.

set_many(mapping, ttl=None, nx=False) async

Sets multiple values after resolving their contextual keys.

It resolves all keys in the batch, forwards the operation to the client, and then maps the results back to the original keys so the caller remains agnostic of the prefixes.

Parameters:

Name	Type	Description	Default
mapping	Dict[str, Any]	A dictionary of logical key-value pairs.	required
ttl	Optional[int]	Time-To-Live in seconds.	None
nx	bool	If True, set only if the key does not exist.	False

Returns:

Type	Description
Dict[str, bool]	Dict[str, bool]: A map of logical keys to success status.

zadd(key, mapping) async

Delegates the zadd operation after resolving the contextual key.

zpopbyscore(key, max_score) async

Delegates the zpopbyscore operation after resolving the contextual key.

zrangebyscore(key, min_score, max_score) async

Delegates the zrangebyscore operation after resolving the contextual key.

zrem(key, members) async

Delegates the zrem operation after resolving the contextual key.

nala.athomic.database.kvstore.wrappers.default_ttl_kv_client.DefaultTTLKvClient

Bases: WrapperBase

A decorator (Wrapper) for the KVStore client that automatically applies a default Time-To-Live (TTL).

This class intercepts the set operation and injects a configured default TTL if no explicit TTL is provided by the caller. It ensures that keys are automatically expired, preventing cache overflow and maintaining consistency.

Attributes:

Name	Type	Description
default_ttl	Optional[int]	The default TTL in seconds extracted from configuration, or None.

__init__(client, settings=None, wrapper_settings=None)

Initializes the wrapper, resolving the configured default TTL value.

Parameters:

Name	Type	Description	Default
client	KVStoreProtocol	The underlying KVStore client being wrapped.	required
settings	Optional[KVStoreSettings]	Optional global KVStore settings.	None
wrapper_settings	Optional[DefaultTTLWrapperSettings]	Specific settings for this wrapper instance.	None

set(key, value, ttl=None, nx=False) async

Sets a key-value pair, applying the default TTL if no explicit TTL is given.

Parameters:

Name	Type	Description	Default
key	str	The key of the item to set.	required
value	Any	The value to store.	required
ttl	Optional[int]	Explicit time-to-live in seconds provided by the caller.	None
nx	bool	If True, set the key only if it does not already exist.	False

Returns:

Name	Type	Description
bool	bool	True if the operation was successful.

set_many(mapping, ttl=None, nx=False) async

Sets multiple key-value pairs, applying the default TTL if no explicit TTL is given.

Parameters:

Name	Type	Description	Default
mapping	Dict[str, Any]	The dictionary of keys and values.	required
ttl	Optional[int]	Explicit time-to-live in seconds for the batch.	None
nx	bool	If True, set keys only if they do not exist.	False

Returns:

Type	Description
Dict[str, bool]	Dict[str, bool]: The result map from the underlying client.

zadd(key, mapping) async

Delegates the zadd operation to the wrapped client.

zpopbyscore(key, max_score) async

Delegates the zpopbyscore operation to the wrapped client.

zrangebyscore(key, min_score, max_score) async

Delegates the zrangebyscore operation to the wrapped client.

zrem(key, members) async

Delegates the zrem operation to the wrapped client.

nala.athomic.database.kvstore.fallback.fallback_kv_provider.FallbackKVProvider

Bases: BaseService, KVStoreProtocol

A Key-Value (KV) provider that implements a resilient fallback and self-healing strategy.

It orchestrates a chain of underlying KV providers: trying the primary first, and then sequentially trying fallbacks in case of failure or cache miss (depending on the configured read strategy).

Attributes:

Name	Type	Description
primary	KVStoreProtocol	The main, preferred KV store provider.
fallbacks	List[KVStoreProtocol]	Ordered list of secondary providers.
write_strategy	WriteStrategyProtocol	Strategy defining how write operations are handled across providers.
read_strategy	ReadStrategyProtocol	Strategy defining when the fallback chain is triggered.
enable_self_healing	bool	If True, successful reads from a fallback provider will asynchronously write the value back to the failed/missing providers.

__init__(primary_provider, fallback_providers, read_strategy_type=FallbackReadStrategyType.ON_MISS_OR_ERROR, write_strategy_type=FallbackWriteStrategyType.WRITE_AROUND, enable_self_healing=True)

Initializes the FallbackKVProvider.

Parameters:

Name	Type	Description	Default
primary_provider	KVStoreProtocol	The primary KV store instance.	required
fallback_providers	List[KVStoreProtocol]	A list of secondary providers, ordered by preference.	required
read_strategy_type	FallbackReadStrategyType	Defines the trigger for the fallback read chain.	ON_MISS_OR_ERROR
write_strategy_type	FallbackWriteStrategyType	Defines how write operations propagate.	WRITE_AROUND
enable_self_healing	bool	Enables background healing of failed providers on successful reads.	True

Raises:

Type	Description
ValueError	If primary_provider is None.

clear() async

Clears all keys in the provider(s) based on the configured write strategy.

delete(key) async

Deletes a key based on the configured write strategy.

exists(key) async

Checks for the existence of a key by delegating the check/fallback logic to the configured read strategy.

Parameters:

Name	Type	Description	Default
key	str	The key to check.	required

Returns:

Name	Type	Description
bool	bool	True if the key exists in any available provider, False otherwise.

get(key) async

Retrieves a value by delegating the complex read/fallback logic to the configured read strategy.

Parameters:

Name	Type	Description	Default
key	str	The key to retrieve.	required

Returns:

Type	Description
Optional[Any]	Optional[Any]: The value from the first successful provider, or None.

get_final_client() async

Returns the raw client of the primary provider for direct access.

get_sync_client()

Returns the synchronous client of the primary provider for direct access.

heal(key, value, providers_to_heal) async

Asynchronously attempts to heal failed or missed providers by writing the successfully retrieved value back to them.

This method is non-blocking and is executed as a fire-and-forget background task.

Parameters:

Name	Type	Description	Default
key	str	The key that was successfully retrieved.	required
value	Any	The value associated with the key.	required
providers_to_heal	List[KVStoreProtocol]	The chain of providers (including the primary) that failed or missed the key.	required

is_available()

Returns True if AT LEAST ONE provider is connected and ready.

set(key, value, ttl=None, nx=False) async

Sets a key-value pair based on the configured write strategy.