Async Key-Value Store - Pluggable Interface

Warnings

• breaking This library is strictly async-only. A synchronous wrapper is not planned, so ensure your application uses `async/await` patterns consistently when integrating `py-key-value-aio`.
  Fix: Rewrite blocking code paths to use `async/await` and integrate with an `asyncio` event loop. Do not attempt to use this library in synchronous contexts directly.
• gotcha Values retrieved from the store are 'managed entries' (often JSON-serialized) and are copies, not 'live' objects. Modifying the retrieved dictionary or Pydantic model will not affect the stored value until you explicitly `put` it back. The library ensures type safety but does not return the same in-memory instance.
  Fix: Always assume retrieved values are copies. If modifications need to be persisted, explicitly call `await key_value_store.put(key, updated_value, ...)` after modifying the local copy.
• gotcha The library utilizes `beartype` for runtime type checking on core protocol methods. Violations will raise `TypeError`. This can be unexpected if not accounted for during development or if data types are inconsistent.
  Fix: Ensure that values passed to `put` and retrieved data conform to the expected types, especially when using Pydantic models. For debugging or specific scenarios, runtime type checks can be disabled by setting the environment variable `PY_KEY_VALUE_DISABLE_BEARTYPE=true`.
• gotcha While `py-key-value-aio` provides many backend integrations, their maturity and specific limitations may vary. Relying on certain backends in production without thorough review of their specific documentation and caveats within the `py-key-value` monorepo can lead to unexpected behavior.
  Fix: Always consult the official `py-key-value` GitHub monorepo documentation for details on individual backend implementations (e.g., DynamoDB, S3, Redis) before deploying them in production environments.

Install

• pip install py-key-value-aio Core library
• pip install py-key-value-aio[memory] With in-memory backend
• pip install py-key-value-aio[redis] With Redis backend (example, many others available)

Imports

• AsyncKeyValue

  from key_value.aio.protocols.key_value import AsyncKeyValue

• MemoryStore

  from key_value.aio.stores.memory import MemoryStore

  Ensure you import from the 'aio' submodule for the async version of the library.

Quickstart

This quickstart demonstrates the core `put`, `get`, and `delete` operations using the `AsyncKeyValue` protocol and an in-memory store. It highlights the asynchronous nature of the library and the common pattern of accepting the `AsyncKeyValue` protocol in your application logic, allowing the underlying store implementation to be swapped easily.

import asyncio
from key_value.aio.protocols.key_value import AsyncKeyValue
from key_value.aio.stores.memory import MemoryStore

async def example_usage(key_value_store: AsyncKeyValue):
    print(f"Putting key 'user:123' with value {{'name': 'Alice'}}")
    await key_value_store.put(key="user:123", value={"name": "Alice"}, collection="users", ttl=3600)
    
    print("Retrieving 'user:123'...")
    user_data = await key_value_store.get(key="user:123", collection="users")
    print(f"Retrieved: {user_data}")

    print("Deleting 'user:123'...")
    await key_value_store.delete(key="user:123", collection="users")

    print("Attempting to retrieve 'user:123' again...")
    deleted_user_data = await key_value_store.get(key="user:123", collection="users")
    print(f"Retrieved after deletion: {deleted_user_data}") # Should be None

async def main():
    # Example with MemoryStore
    memory_store = MemoryStore()
    await example_usage(memory_store)

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