`context.kv()` – EphemeralKV API Reference¶

EphemeralKV is a process‑local, transient key–value store for small JSON‑serializable values and short‑lived coordination. It is thread‑safe (RLock), supports TTLs, and provides a few list helpers. Do not use for large blobs or durable state.

The API below is consistent across KV backends. In context, this KV is ephemeral; other implementations may be pluggable later.

Concepts & Defaults¶

Scope: In‑process only; cleared on restart. Not replicated.
Thread‑safety: Internal reads/writes guarded by RLock.
TTL: Expiry is checked lazily on access and via purge_expired().
Prefixing: Instances may prepend a prefix to all keys for namespacing.

1. Core Methods¶

set(key, value, *, ttl_s)

Store a value in the ephemeral key-value store with optional expiration.

This method inserts or updates the value for a given key, optionally setting a time-to-live (TTL) in seconds. If a TTL is provided, the entry will expire and be automatically removed after the specified duration.

Examples:

Basic usage to store a value:

await context.kv().set("session_token", "abc123")

Storing a value with a 10-minute expiration:

await context.kv().set("user_id", 42, ttl_s=600)

Parameters:

Name	Type	Description	Default
`key`	`str`	The string key under which to store the value.	required
`value`	`Any`	The value to store (any serializable object).	required
`ttl_s`	`int \| None`	Optional expiration time in seconds. If None, the value does not expire.	`None`

Returns:

Type	Description
`None`	None

get(key, default)

Retrieve the value associated with a key from the ephemeral key-value store.

This method checks for key existence and expiration, automatically removing expired entries. If the key does not exist or is expired, the provided default value is returned.

Examples:

Basic usage to fetch a value:

value = await context.kv().get("session_token")

Providing a default if the key is missing or expired:

user_id = await context.kv().get("user_id", default=None)

Parameters:

Name	Type	Description	Default
`key`	`str`	The string key to look up.	required
`default`	`Any`	The value to return if the key is not found or has expired.	`None`

Returns:

Type	Description
`Any`	The stored value if present and not expired; otherwise, the `default`.

delete(key)

Remove a key and its associated value from the ephemeral key-value store.

This method deletes the specified key from the store if it exists. If the key does not exist, the operation is a no-op and does not raise an error.

Examples:

Basic usage to delete a key:

await context.kv().delete("session_token")

Deleting a user-specific cache entry:

await context.kv().delete(f"user_cache:{user_id}")

Parameters:

Name	Type	Description	Default
`key`	`str`	The string key to remove from the store.	required

Returns:

Type	Description
`None`	None

list_append_unique(key, items, *, id_key, ttl_s)

Append unique dictionary items to a list stored in the ephemeral key-value store.

This method ensures that only items with unique id_key values are added to the list associated with the given key. If the key does not exist, a new list is created. Optionally, a time-to-live (TTL) can be set for the entry.

Examples:

Basic usage to append unique items:

await context.kv().list_append_unique("recent_users", [{"id": 1, "name": "Alice"}])

Appending multiple items with a custom ID key and expiration:

await context.kv().list_append_unique(
    "tasks",
    [{"task_id": 42, "desc": "Review PR"}],
    id_key="task_id",
    ttl_s=3600
)

Parameters:

Name	Type	Description	Default
`key`	`str`	The string key under which the list is stored.	required
`items`	`list[dict]`	A list of dictionaries to append. Only items with unique `id_key` values (not already present in the list) will be added.	required
`id_key`	`str`	The dictionary key used to determine uniqueness (default: `"id"`).	`'id'`
`ttl_s`	`int \| None`	Optional expiration time in seconds for the updated list. If None, the list does not expire.	`None`

Returns:

Type	Description
`list[dict]`	list[dict]: The updated list of dictionaries after appending unique items.

Notes

This method is used for lists of dictionaries where each dictionary has a unique identifier. For example, it can be used to maintain a list of recent user actions, ensuring no duplicates based on user ID.

Example of the stored list structure:

[
    {"id": 1, "name": "Alice"},
    {"id": 2, "name": "Bob"},
    ...
]

list_pop_all(key)

Atomically remove and return all items from a list stored in the ephemeral key-value store.

This method retrieves the entire list associated with the given key and removes the key from the store. If the key does not exist or does not contain a list, an empty list is returned. This operation is atomic and ensures no items are left behind after the call.

Examples:

Basic usage to pop all items from a list:

items = await context.kv().list_pop_all("recent_events")

Handling the case where the key may not exist:

logs = await context.kv().list_pop_all("logs")  # returns [] if "logs" is missing

Parameters:

Name	Type	Description	Default
`key`	`str`	The string key under which the list is stored.	required

Returns:

Name	Type	Description
`list`	`list`	The list of items that were stored under the key, or an empty list if the key was not found.

2. Optional Helpers¶

mset(kv, *, ttl_s)

Set multiple key-value pairs in the ephemeral key-value store.

This asynchronous method iterates over the provided dictionary and sets each key-value pair in the store, optionally applying a time-to-live (TTL) to each entry. If a TTL is specified, each key will expire after the given number of seconds.

Examples:

Basic usage to set multiple values:

await context.kv().mset({"foo": 1, "bar": "baz"})

Setting multiple values with a TTL of 60 seconds:

await context.kv().mset({"session": "abc", "count": 42}, ttl_s=60)

Parameters:

Name	Type	Description	Default
`kv`	`dict[str, Any]`	A dictionary mapping string keys to values to be stored.	required
`ttl_s`	`int \| None`	Optional; the time-to-live for each key in seconds. If None, keys do not expire.	`None`

Returns:

Type	Description
`None`	None

mget(keys)

Retrieve multiple values from the ephemeral key-value store in a single call.

This method fetches the values associated with each key in the provided list, preserving the order of the input. If a key does not exist or is expired, None is returned in its place.

Examples:

Basic usage to fetch several values:

values = await context.kv().mget(["user_id", "session_token", "profile"])

Handling missing or expired keys:

results = await context.kv().mget(["foo", "bar"])
# results might be [None, "bar_value"] if "foo" is missing or expired

Parameters:

Name	Type	Description	Default
`keys`	`list[str]`	A list of string keys to retrieve from the store.	required

Returns:

Type	Description
`list[Any]`	list[Any]: A list of values corresponding to the input keys. If a key is not found or has expired, its position in the list will be `None`.

expire(key, ttl_s)

Set or update the expiration time (TTL) for a key in the ephemeral key-value store.

This method updates the expiration timestamp for an existing key, causing it to expire and be automatically removed after the specified number of seconds. If the key does not exist, this operation is a no-op.

Examples:

Basic usage to set a 5-minute expiration:

await context.kv().expire("session_token", ttl_s=300)

Updating the TTL for a cached user profile:

await context.kv().expire("user_profile:42", ttl_s=60)

Parameters:

Name	Type	Description	Default
`key`	`str`	The string key whose expiration time should be set or updated.	required
`ttl_s`	`int`	The time-to-live in seconds from now. After this duration, the key will expire.	required

Returns:

Type	Description
`None`	None

purge_expired(limit)

Remove expired key-value entries from the ephemeral store.

This method scans the internal data store for entries whose expiration timestamp has passed and removes them, up to the specified limit. It is intended to be called periodically to keep the store clean and efficient.

Examples:

Purge up to 1000 expired entries:

removed = await context.kv().purge_expired()

Purge a custom number of expired entries:

removed = await context.kv().purge_expired(limit=500)

Parameters:

Name	Type	Description	Default
`limit`	`int`	The maximum number of expired entries to remove in a single call.	`1000`

Returns:

Name	Type	Description
`int`	`int`	The number of expired entries that were removed from the store.

context.kv() – EphemeralKV API Reference¶

Concepts & Defaults¶

1. Core Methods¶

2. Optional Helpers¶

`context.kv()` – EphemeralKV API Reference¶