py-key-value

https://github.com/strawgate/py-key-value
Admin
A pluggable async Python library providing a unified interface for key-value storage across multiple...
Tokens:29,554
Snippets:334
Trust Score:9.7
License:Apache-2.0
Update:4 months ago
Show doc for...
Tokens:
Raw
### Development Store Example

Source: https://github.com/strawgate/py-key-value/blob/main/docs/stores.md

Use MemoryStore for fast iteration during development. No setup is required.

```python
# Development
from key_value.aio.stores.memory import MemoryStore
store = MemoryStore()
```

--------------------------------

### Async Key-Value Store Example

Source: https://github.com/strawgate/py-key-value/blob/main/README.md

Demonstrates basic put, get, and delete operations using an asynchronous key-value store. Requires an async context.

```python
import asyncio

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


async def example(key_value: AsyncKeyValue) -> None:
    await key_value.put(key="123", value={"name": "Alice"}, collection="users", ttl=3600)
    value = await key_value.get(key="123", collection="users")
    await key_value.delete(key="123", collection="users")


async def main():
    memory_store = MemoryStore()
    await example(key_value=memory_store)

asyncio.run(main())
```

--------------------------------

### Initialize RocksDBStore

Source: https://github.com/strawgate/py-key-value/blob/main/docs/stores.md

Example of initializing a RocksDBStore with a specified path.

```python
from key_value.aio.stores.rocksdb import RocksDBStore

store = RocksDBStore(path="./rocksdb")
```

--------------------------------

### Install Py-Key-Value with Backend Extras

Source: https://github.com/strawgate/py-key-value/blob/main/README.md

Install the async library with specific backend extras. Choose the backend that suits your needs.

```bash
# Async library
pip install py-key-value-aio

# With specific backend extras
pip install py-key-value-aio[memory]
pip install py-key-value-aio[disk]
pip install py-key-value-aio[dynamodb]
pip install py-key-value-aio[s3]
pip install py-key-value-aio[azure-tables]
pip install py-key-value-aio[elasticsearch]
pip install py-key-value-aio[firestore]
# or: aerospike, redis, mongodb, memcached, valkey, vault, registry, rocksdb,
# see below for all options
```

--------------------------------

### Install Valkey Store

Source: https://github.com/strawgate/py-key-value/blob/main/docs/stores.md

Install the necessary package for Valkey integration.

```bash
pip install py-key-value-aio[valkey]
```

--------------------------------

### Install py-key-value-aio with multiple backends

Source: https://github.com/strawgate/py-key-value/blob/main/docs/getting-started.md

Install the asynchronous library with multiple backend supports simultaneously.

```bash
# Install multiple backends at once
pip install py-key-value-aio[azure-tables,firestore,elasticsearch]
# Other available extras: duckdb, memcached, mongodb, opensearch,
# postgresql, rocksdb, s3, chdb, valkey, vault, keyring, aerospike
```

--------------------------------

### Install py-key-value-aio with Firestore support

Source: https://github.com/strawgate/py-key-value/blob/main/docs/getting-started.md

Install the asynchronous library with Firestore backend support.

```bash
# Firestore support
pip install py-key-value-aio[firestore]
```

--------------------------------

### Install py-key-value-aio with DynamoDB support

Source: https://github.com/strawgate/py-key-value/blob/main/docs/getting-started.md

Install the asynchronous library with DynamoDB backend support.

```bash
# DynamoDB support
pip install py-key-value-aio[dynamodb]
```

--------------------------------

### Install OpenSearch Package

Source: https://github.com/strawgate/py-key-value/blob/main/docs/stores.md

Install the py-key-value-aio package with OpenSearch support.

```bash
pip install py-key-value-aio[opensearch]
```

--------------------------------

### Install py-key-value-aio with Redis support

Source: https://github.com/strawgate/py-key-value/blob/main/docs/getting-started.md

Install the asynchronous library with Redis backend support.

```bash
# Redis support
pip install py-key-value-aio[redis]
```

--------------------------------

### Install py-key-value-aio with Elasticsearch support

Source: https://github.com/strawgate/py-key-value/blob/main/docs/getting-started.md

Install the asynchronous library with Elasticsearch backend support.

```bash
# Elasticsearch support
pip install py-key-value-aio[elasticsearch]
```

--------------------------------

### Install py-key-value-aio with MongoDB support

Source: https://github.com/strawgate/py-key-value/blob/main/docs/getting-started.md

Install the asynchronous library with MongoDB backend support.

```bash
# MongoDB support
pip install py-key-value-aio[mongodb]
```

--------------------------------

### Install PostgreSQL Support

Source: https://github.com/strawgate/py-key-value/blob/main/docs/stores.md

Install the necessary package for PostgreSQL integration. This command installs the core library with PostgreSQL extras.

```bash
pip install py-key-value-aio[postgresql]
```

--------------------------------

### Initialize KeyringStore

Source: https://github.com/strawgate/py-key-value/blob/main/docs/stores.md

Instantiate KeyringStore for OS-level secure storage. Requires the 'keyring' extra during installation.

```python
from key_value.aio.stores.keyring import KeyringStore

store = KeyringStore(service_name="py-key-value")
```

--------------------------------

### Initialize NullStore

Source: https://github.com/strawgate/py-key-value/blob/main/docs/stores.md

Example of initializing a NullStore, which discards all data.

```python
from key_value.aio.stores.null import NullStore

store = NullStore()
```

--------------------------------

### Install Memcached Support

Source: https://github.com/strawgate/py-key-value/blob/main/docs/stores.md

Install the necessary package for Memcached integration. This command installs the core library with Memcached extras.

```bash
pip install py-key-value-aio[memcached]
```

--------------------------------

### Install S3 Package

Source: https://github.com/strawgate/py-key-value/blob/main/docs/stores.md

Install the py-key-value-aio package with S3 support.

```bash
pip install py-key-value-aio[s3]
```

--------------------------------

### Install Windows Registry Store

Source: https://github.com/strawgate/py-key-value/blob/main/docs/stores.md

Command to install the Windows Registry store with its dependencies.

```bash
pip install py-key-value-aio[registry]
```

--------------------------------

### Install Firestore Store

Source: https://github.com/strawgate/py-key-value/blob/main/docs/stores.md

Install the necessary package for Firestore integration.

```bash
pip install py-key-value-aio[firestore]
```

--------------------------------

### Install KeyringStore dependency

Source: https://github.com/strawgate/py-key-value/blob/main/docs/stores.md

Install the necessary package for KeyringStore functionality.

```bash
pip install py-key-value-aio[keyring]
```

--------------------------------

### Install py-key-value with Redis support

Source: https://github.com/strawgate/py-key-value/blob/main/docs/index.md

Install the py-key-value async library with specific backend support for Redis.

```bash
pip install py-key-value-aio[redis]
```

--------------------------------

### Initialize SimpleStore

Source: https://github.com/strawgate/py-key-value/blob/main/docs/stores.md

Example of initializing a SimpleStore, an in-memory store for testing.

```python
from key_value.aio.stores.simple import SimpleStore

store = SimpleStore()
```

--------------------------------

### Create Custom Async Wrapper

Source: https://github.com/strawgate/py-key-value/blob/main/docs/wrappers.md

Example of creating a custom asynchronous wrapper by extending BaseWrapper and overriding the 'get' method.

```python
from key_value.aio.wrappers.base import BaseWrapper
from typing_extensions import override

class CustomWrapper(BaseWrapper):
    def __init__(self, key_value: AsyncKeyValue):
        self.key_value = key_value
        super().__init__()

    @override
    async def get(self, key: str, *, collection: str | None = None):
        # Add custom logic before
        print(f"Getting key: {key}")

        # Call wrapped store
        result = await self.key_value.get(key=key, collection=collection)

        # Add custom logic after
        print(f"Got result: {result}")

        return result
```

--------------------------------

### Quick Example: Storing and Retrieving Data with MemoryStore

Source: https://github.com/strawgate/py-key-value/blob/main/docs/index.md

Demonstrates basic usage of the MemoryStore for storing a dictionary value with a TTL and then retrieving it.

```python
from key_value.aio.stores.memory import MemoryStore

# Create a store
store = MemoryStore()

# Store a value with TTL
await store.put(
    key="user:123",
    value={"name": "Alice", "email": "alice@example.com"},
    collection="users",
    ttl=3600  # 1 hour
)

# Retrieve the value
user = await store.get(key="user:123", collection="users")
print(user)  # {"name": "Alice", "email": "alice@example.com"}
```

--------------------------------

### Sync dependencies

Source: https://github.com/strawgate/py-key-value/blob/main/DEVELOPING.md

Installs all project dependencies using the make utility.

```bash
make sync
```

--------------------------------

### Install Azure Tables Package

Source: https://github.com/strawgate/py-key-value/blob/main/docs/stores.md

Install the py-key-value-aio package with Azure Tables support.

```bash
pip install py-key-value-aio[azure-tables]
```

--------------------------------

### Store Compatibility Example

Source: https://github.com/strawgate/py-key-value/blob/main/docs/stores.md

Demonstrates switching between MemoryStore for development and RedisStore for production. All stores implement the same protocol, ensuring code compatibility.

```python
# Development
store = MemoryStore()

# Production
store = RedisStore(url="redis://localhost:6379/0")

# Your code works with both!
await store.put(key="user:123", value={"name": "Alice"}, collection="users")
```

--------------------------------

### Install py-key-value with multiple backend supports

Source: https://github.com/strawgate/py-key-value/blob/main/docs/index.md

Install the py-key-value async library with support for multiple backends simultaneously, such as Azure Tables, Firestore, and Elasticsearch.

```bash
pip install py-key-value-aio[azure-tables,firestore,elasticsearch]
```

--------------------------------

### Install py-key-value with DynamoDB support

Source: https://github.com/strawgate/py-key-value/blob/main/docs/index.md

Install the py-key-value async library with specific backend support for DynamoDB.

```bash
pip install py-key-value-aio[dynamodb]
```

--------------------------------

### Install ChDB Store

Source: https://github.com/strawgate/py-key-value/blob/main/docs/stores.md

Command to install the ChDB store with its dependencies.

```bash
pip install py-key-value-aio[chdb]
```

--------------------------------

### PydanticAdapter - Basic Example

Source: https://github.com/strawgate/py-key-value/blob/main/docs/adapters.md

Demonstrates how to initialize and use the PydanticAdapter to store and retrieve a Pydantic model.

```APIDOC
## PydanticAdapter - Basic Example

### Description
This example shows how to create a `PydanticAdapter` for a `User` model and perform basic `put` and `get` operations.

### Method
PUT, GET

### Endpoint
`/users` (implicitly via adapter)

### Parameters
#### Request Body (PUT)
- **key** (str) - Required - The key to store the value under.
- **value** (User) - Required - The Pydantic model instance to store.
- **collection** (str) - Optional - The collection to store the data in.

#### Query Parameters (GET)
- **key** (str) - Required - The key of the value to retrieve.
- **collection** (str) - Optional - The collection to retrieve the data from.

### Request Example (PUT)
```python
from pydantic import BaseModel
from key_value.aio.stores.memory import MemoryStore
from key_value.aio.adapters.pydantic import PydanticAdapter

class User(BaseModel):
    name: str
    email: str
    age: int

# Create adapter
adapter = PydanticAdapter(
    key_value=MemoryStore(),
    pydantic_model=User
)

# Store a user (type-safe)
user = User(name="Alice", email="alice@example.com", age=30)
await adapter.put(key="user:123", value=user, collection="users")
```

### Response (GET)
#### Success Response (200)
- **retrieved_user** (User) - The retrieved and validated Pydantic model instance.

#### Response Example (GET)
```python
# Retrieve and get a validated model
retrieved_user = await adapter.get(key="user:123", collection="users")
if retrieved_user:
    print(retrieved_user.name)  # Type-safe: "Alice"
    print(retrieved_user.email)  # Type-safe: "alice@example.com"
```
```

--------------------------------

### Install Aerospike Store

Source: https://github.com/strawgate/py-key-value/blob/main/docs/stores.md

Install the necessary package for Aerospike integration.

```bash
pip install py-key-value-aio[aerospike]
```

--------------------------------

### Install RocksDB Store

Source: https://github.com/strawgate/py-key-value/blob/main/docs/stores.md

Command to install the RocksDB store with its dependencies.

```bash
pip install py-key-value-aio[rocksdb]
```

--------------------------------

### Install py-key-value async library

Source: https://github.com/strawgate/py-key-value/blob/main/docs/index.md

Install the asynchronous version of the py-key-value library using pip.

```bash
pip install py-key-value-aio
```

--------------------------------

### Install DuckDB Store

Source: https://github.com/strawgate/py-key-value/blob/main/docs/stores.md

Command to install the DuckDB store with its dependencies.

```bash
pip install py-key-value-aio[duckdb]
```

--------------------------------

### Install optional backend dependencies

Source: https://github.com/strawgate/py-key-value/blob/main/DEVELOPING.md

Installs specific backend support packages via pip extras.

```bash
pip install py-key-value-aio[redis]      # Redis support
pip install py-key-value-aio[dynamodb]   # DynamoDB support
pip install py-key-value-aio[mongodb]    # MongoDB support
# etc. - see README.md for full list
```

--------------------------------

### Install VaultStore dependency

Source: https://github.com/strawgate/py-key-value/blob/main/docs/stores.md

Install the necessary package for VaultStore functionality.

```bash
pip install py-key-value-aio[vault]
```

--------------------------------

### CollectionRoutingWrapper Example

Source: https://github.com/strawgate/py-key-value/blob/main/docs/wrappers.md

Demonstrates how to route operations to different stores based on collection names. Requires importing necessary store and wrapper classes.

```python
from key_value.aio.stores.memory import MemoryStore
from key_value.aio.stores.redis import RedisStore
from key_value.aio.stores.disk import DiskStore
from key_value.aio.wrappers.routing import CollectionRoutingWrapper

store = CollectionRoutingWrapper(
    collection_map={
        "sessions": RedisStore(url="redis://localhost:6379/0"),
        "users": DiskStore(directory="./users"),
    },
    default_store=MemoryStore()
)

# "sessions" go to Redis, "users" go to Disk, everything else to memory
await store.put(key="s1", value={"user": "alice"}, collection="sessions")
await store.put(key="u1", value={"name": "Alice"}, collection="users")
await store.put(key="tmp", value={"data": "value"}, collection="cache")
```

--------------------------------

### Initialize FileTreeStore

Source: https://github.com/strawgate/py-key-value/blob/main/docs/stores.md

Directory-based storage with JSON files for visual inspection and debugging. Not recommended for production use due to performance and filesystem limitations. Install with `pip install py-key-value-aio[filetree]`.

```python
from key_value.aio.stores.filetree import FileTreeStore

store = FileTreeStore(directory="./debug-store")
```

--------------------------------

### Initialize VaultStore

Source: https://github.com/strawgate/py-key-value/blob/main/docs/stores.md

Instantiate VaultStore for integration with HashiCorp Vault. Requires the 'vault' extra during installation.

```python
from key_value.aio.stores.vault import VaultStore

store = VaultStore(
    url="http://localhost:8200",
    token="your-token"
)
```

--------------------------------

### Initialize DiskStore

Source: https://github.com/strawgate/py-key-value/blob/main/docs/stores.md

Persistent file-based storage using a single JSON file. Suitable for small to medium datasets. Install with `pip install py-key-value-aio[disk]`.

```python
from key_value.aio.stores.disk import DiskStore

store = DiskStore(directory="./cache")
```

--------------------------------

### Initialize MultiDiskStore

Source: https://github.com/strawgate/py-key-value/blob/main/docs/stores.md

Persistent storage with separate files per collection, organized by directory. Better performance with many collections. Install with `pip install py-key-value-aio[disk]`.

```python
from key_value.aio.stores.multi_disk import MultiDiskStore

store = MultiDiskStore(directory="./cache")
```

--------------------------------

### Production Caching Store Example

Source: https://github.com/strawgate/py-key-value/blob/main/docs/stores.md

Use RedisStore for high-performance, distributed production caching. Requires a Redis instance.

```python
# Production caching
from key_value.aio.stores.redis import RedisStore
store = RedisStore(url="redis://localhost:6379/0")
```

--------------------------------

### Serverless/Cloud Store Example

Source: https://github.com/strawgate/py-key-value/blob/main/docs/stores.md

Use DynamoDBStore for serverless environments like AWS Lambda. Requires table name and region.

```python
# AWS Lambda
from key_value.aio.stores.dynamodb import DynamoDBStore
store = DynamoDBStore(table_name="kv-store", region_name="us-east-1")
```

--------------------------------

### Combine Encryption, Compression, and PydanticAdapter

Source: https://github.com/strawgate/py-key-value/blob/main/docs/adapters.md

This example demonstrates layering FernetEncryptionWrapper and CompressionWrapper before initializing PydanticAdapter for type-safe, encrypted, and compressed storage.

```python
from pydantic import BaseModel
from key_value.aio.stores.memory import MemoryStore
from key_value.aio.wrappers.encryption.fernet import FernetEncryptionWrapper
from key_value.aio.wrappers.compression import CompressionWrapper
from key_value.aio.adapters.pydantic import PydanticAdapter
from cryptography.fernet import Fernet

class User(BaseModel):
    name: str
    email: str

# Create encrypted + compressed store
wrapped_store = CompressionWrapper(
    key_value=FernetEncryptionWrapper(
        key_value=MemoryStore(),
        fernet=Fernet(Fernet.generate_key())
    )
)

# Wrap with PydanticAdapter for type safety
adapter = PydanticAdapter(
    key_value=wrapped_store,
    pydantic_model=User
)

# Now you have type-safe, encrypted, and compressed storage!
await adapter.put(key="user:123", value=User(name="Alice", email="alice@example.com"))
```

--------------------------------

### Stacked Wrapper Example

Source: https://github.com/strawgate/py-key-value/blob/main/README.md

Illustrates chaining multiple wrappers for complex functionality, such as retry, timeout, compression, and fallback mechanisms.

```python
# Create a retriable redis store with timeout protection that is monitored,
# with compressed values, and a fallback to memory store! This probably isn't
# a good idea but you can do it!
store = 
LoggingWrapper(
    CompressionWrapper(
        FallbackWrapper(
            primary_key_value=RetryWrapper(
                TimeoutWrapper(
                    key_value=redis_store,
                )
            ),
            fallback_key_value=memory_store,
        )
    )
)
```

--------------------------------

### PydanticAdapter - Batch Operations

Source: https://github.com/strawgate/py-key-value/blob/main/docs/adapters.md

Demonstrates performing batch put and get operations with PydanticAdapter.

```APIDOC
## PydanticAdapter - Batch Operations

### Description
This example illustrates how to use `put_many` and `get_many` for efficient batch storage and retrieval of multiple Pydantic models.

### Method
PUT, GET

### Endpoint
`/users` (implicitly via adapter)

### Parameters
#### Request Body (PUT - `put_many`)
- **keys** (list[str]) - Required - A list of keys for the values to store.
- **values** (list[User]) - Required - A list of Pydantic model instances to store.
- **collection** (str) - Optional - The collection to store the data in.

#### Query Parameters (GET - `get_many`)
- **keys** (list[str]) - Required - A list of keys for the values to retrieve.
- **collection** (str) - Optional - The collection to retrieve the data from.

### Request Example (PUT - `put_many`)
```python
from pydantic import BaseModel
from key_value.aio.stores.memory import MemoryStore
from key_value.aio.adapters.pydantic import PydanticAdapter

class User(BaseModel):
    name: str
    email: str
    age: int

adapter = PydanticAdapter(
    key_value=MemoryStore(),
    pydantic_model=User,
    default_collection="users"
)

# Store multiple users
users = [
    User(name="Alice", email="alice@example.com", age=30),
    User(name="Bob", email="bob@example.com", age=25),
    User(name="Charlie", email="charlie@example.com", age=35),
]

await adapter.put_many(
    keys=["user:1", "user:2", "user:3"],
    values=users,
    collection="users"
)
```

### Response (GET - `get_many`)
#### Success Response (200)
- **retrieved** (list[User | None]) - A list containing the retrieved Pydantic model instances or `None` if not found.

#### Response Example (GET - `get_many`)
```python
# Retrieve multiple users
retrieved = await adapter.get_many(
    keys=["user:1", "user:2", "user:3"],
    collection="users"
)

for user in retrieved:
    if user:
        print(user.name)
```
```

--------------------------------

### PydanticAdapter Basic Example

Source: https://github.com/strawgate/py-key-value/blob/main/docs/adapters.md

Illustrates storing and retrieving a Pydantic model using PydanticAdapter. Ensures type-safe access to retrieved data.

```python
from pydantic import BaseModel
from key_value.aio.stores.memory import MemoryStore
from key_value.aio.adapters.pydantic import PydanticAdapter

class User(BaseModel):
    name: str
    email: str
    age: int

# Create adapter
adapter = PydanticAdapter(
    key_value=MemoryStore(),
    pydantic_model=User
)

# Store a user (type-safe)
user = User(name="Alice", email="alice@example.com", age=30)
await adapter.put(key="user:123", value=user, collection="users")

# Retrieve and get a validated model
retrieved_user = await adapter.get(key="user:123", collection="users")
if retrieved_user:
    print(retrieved_user.name)  # Type-safe: "Alice"
    print(retrieved_user.email)  # Type-safe: "alice@example.com"
```

--------------------------------

### PydanticAdapter Example

Source: https://github.com/strawgate/py-key-value/blob/main/README.md

Demonstrates using PydanticAdapter for type-safe storage of Pydantic models with a MemoryStore. Ensures data validation and serialization.

```python
import asyncio
from pydantic import BaseModel

from key_value.aio.adapters.pydantic import PydanticAdapter
from key_value.aio.stores.memory import MemoryStore

class User(BaseModel):
    name: str
    email: str

async def example():
    memory_store: MemoryStore = MemoryStore()

    user_adapter: PydanticAdapter[User] = PydanticAdapter(
        key_value=memory_store,
        pydantic_model=User,
        default_collection="users",
    )

    new_user: User = User(name="John Doe", email="john.doe@example.com")
    
    # Directly store the User model
    await user_adapter.put(
        key="john-doe",
        value=new_user,
    )

    # Retrieve the User model
    existing_user: User | None = await user_adapter.get(
        key="john-doe",
    )

asyncio.run(example())
```

--------------------------------

### Store Product instances with PydanticAdapter

Source: https://github.com/strawgate/py-key-value/blob/main/docs/adapters.md

Both BaseModelAdapter and PydanticAdapter can store BaseModel instances. This example shows storing a Product.

```python
# Both can store Product instances
product = Product(name="Widget", price=29.99)
await base_adapter.put(key="product:1", value=product)
await pydantic_adapter.put(key="product:1", value=product)
```

--------------------------------

### Chaining Wrappers and Adapters

Source: https://github.com/strawgate/py-key-value/blob/main/README.md

This example shows how to combine an Elasticsearch store with a single collection wrapper, a statistics wrapper, and a Pydantic adapter. Use this pattern when you need to layer multiple functionalities like data validation, metrics, and simplified collection management on top of a base store.

```python
import asyncio
from pydantic import BaseModel

from key_value.aio.adapters.pydantic import PydanticAdapter
from key_value.aio.wrappers.single_collection import SingleCollectionWrapper
from key_value.aio.wrappers.statistics import StatisticsWrapper
from key_value.aio.stores.elasticsearch import ElasticsearchStore


class User(BaseModel):
    name: str
    email: str

elasticsearch_store: ElasticsearchStore = ElasticsearchStore(
    url="https://localhost:9200", api_key="your-api-key", index="kv-store"
)

single_collection: SingleCollectionWrapper = SingleCollectionWrapper(
    key_value=elasticsearch_store, single_collection="users",
    default_collection="one-collection"
)


async def main(key_value: AsyncKeyValue):
    statistics_wrapper = StatisticsWrapper(key_value=key_value)
    users = PydanticAdapter(key_value=statistics_wrapper, pydantic_model=User)

    await users.put(
        key="u1", value=User(name="Jane", email="j@example.com"),
        collection="ignored"
    )
    user = await users.get(key="u1", collection="ignored")
    _ = statistics_wrapper.statistics  # access metrics


asyncio.run(main(key_value=single_collection))
```

--------------------------------

### Sensitive Data Store Example

Source: https://github.com/strawgate/py-key-value/blob/main/docs/stores.md

Use KeyringStore for handling sensitive data, leveraging OS-level encryption. Requires specifying a service name.

```python
# Sensitive data
from key_value.aio.stores.keyring import KeyringStore
store = KeyringStore(service_name="my-app")
```

--------------------------------

### Initialize ValkeyStore

Source: https://github.com/strawgate/py-key-value/blob/main/docs/stores.md

Instantiate ValkeyStore using host and port. Suitable for standalone Valkey instances.

```python
from key_value.aio.stores.valkey import ValkeyStore

store = ValkeyStore(host="localhost", port=6379)
```

--------------------------------

### Initialize DiskStore

Source: https://github.com/strawgate/py-key-value/blob/main/docs/getting-started.md

Set up the DiskStore by specifying the directory where data will be stored.

```python
from key_value.aio.stores.disk import DiskStore

store = DiskStore(directory="/path/to/storage")
```

--------------------------------

### DefaultValueWrapper Example

Source: https://github.com/strawgate/py-key-value/blob/main/docs/wrappers.md

Shows how to use DefaultValueWrapper to return a default value for missing keys, mimicking dict.get behavior. Includes setting a default TTL.

```python
from key_value.aio.stores.memory import MemoryStore
from key_value.aio.wrappers.default_value import DefaultValueWrapper

store = DefaultValueWrapper(
    key_value=MemoryStore(),
    default_value={"theme": "light", "language": "en"},
    default_ttl=3600
)

# Returns the default value instead of None for missing keys
prefs = await store.get(key="user:new", collection="preferences")
print(prefs)  # {"theme": "light", "language": "en"}
```

--------------------------------

### PydanticAdapter Batch Operations

Source: https://github.com/strawgate/py-key-value/blob/main/docs/adapters.md

Demonstrates using PydanticAdapter for batch put and get operations to improve performance when handling multiple items.

```python
# Store multiple users
users = [
    User(name="Alice", email="alice@example.com", age=30),
    User(name="Bob", email="bob@example.com", age=25),
    User(name="Charlie", email="charlie@example.com", age=35),
]

await adapter.put_many(
    keys=["user:1", "user:2", "user:3"],
    values=users,
    collection="users"
)

# Retrieve multiple users
retrieved = await adapter.get_many(
    keys=["user:1", "user:2", "user:3"],
    collection="users"
)

for user in retrieved:
    if user:
        print(user.name)
```

--------------------------------

### Initialize RaiseOnMissingAdapter

Source: https://github.com/strawgate/py-key-value/blob/main/docs/adapters.md

Use RaiseOnMissingAdapter to change 'get' operations to raise an error for missing keys instead of returning None. Requires importing MemoryStore, RaiseOnMissingAdapter, and MissingKeyError.

```python
from key_value.aio.stores.memory import MemoryStore
from key_value.aio.adapters.raise_on_missing import RaiseOnMissingAdapter
from key_value.aio.errors import MissingKeyError

adapter = RaiseOnMissingAdapter(
    key_value=MemoryStore()
)
```

--------------------------------

### ValkeyStore

Source: https://github.com/strawgate/py-key-value/blob/main/docs/api/stores.md

Valkey-backed key-value store (Redis-compatible). Valkey Cluster deployments can be configured by passing a GlideClusterClientConfiguration to ValkeyStore(config=...). The store creates the corresponding GlideClusterClient during async setup instead of requiring callers to create a connected client up front.

```APIDOC
## ValkeyStore

### Description
Valkey-backed key-value store (Redis-compatible).

Valkey Cluster deployments can be configured by passing a
`GlideClusterClientConfiguration` to `ValkeyStore(config=...)`. The store
creates the corresponding `GlideClusterClient` during async setup instead of
requiring callers to create a connected client up front.

### Class
key_value.aio.stores.valkey.ValkeyStore

### Methods
- __init__
```

--------------------------------

### Initialize DuckDBStore

Source: https://github.com/strawgate/py-key-value/blob/main/docs/stores.md

Demonstrates initializing DuckDBStore for in-memory, persistent, or with an existing connection.

```python
from key_value.aio.stores.duckdb import DuckDBStore

# In-memory database (default)
store = DuckDBStore()

# Persistent database
store = DuckDBStore(database_path="./my_store.duckdb")

# With existing connection
import duckdb
connection = duckdb.connect("./my_store.duckdb")
store = DuckDBStore(connection=connection)
```

--------------------------------

### Get existing and missing keys with RaiseOnMissingAdapter

Source: https://github.com/strawgate/py-key-value/blob/main/docs/adapters.md

Demonstrates retrieving an existing key and handling a missing key exception with RaiseOnMissingAdapter. Ensure KeyNotFoundError is imported.

```python
# Store a value
await adapter.put(key="user:123", value={"name": "Alice"}, collection="users")

# Get existing key - works normally
user = await adapter.get(key="user:123", collection="users")
print(user)  # {"name": "Alice"}

# Get missing key - raises error
try:
    user = await adapter.get(key="user:999", collection="users")
except KeyNotFoundError as e:
    print(f"Key not found: {e}")
```

--------------------------------

### Retrieve a value from MemoryStore

Source: https://github.com/strawgate/py-key-value/blob/main/docs/getting-started.md

Use the `get` method to retrieve a value by its key and collection from the MemoryStore. The retrieved value is printed to the console.

```python
user = await store.get(key="user:123", collection="users")
print(user)  # {"name": "Alice", "email": "alice@example.com"}
```

--------------------------------

### Initialize ChDBStore

Source: https://github.com/strawgate/py-key-value/blob/main/docs/stores.md

Illustrates initializing ChDBStore for in-memory, persistent storage, or with an existing chDB session.

```python
from key_value.aio.stores.chdb import ChDBStore

# In-memory database (default)
store = ChDBStore()

# Persistent database (chDB stores data in a directory)
store = ChDBStore(database_path="./my_store.chdb")

# With an existing chDB session
from chdb.session import Session
session = Session("./my_store.chdb")
store = ChDBStore(session=session)
```

--------------------------------

### StatisticsWrapper Example

Source: https://github.com/strawgate/py-key-value/blob/main/docs/wrappers.md

Utilize StatisticsWrapper to track operation statistics, including counts, hits, and misses for store operations. This wrapper is ideal for performance monitoring, cache hit rate analysis, and usage analytics. Statistics can be retrieved and reset.

```python
from key_value.aio.stores.memory import MemoryStore
from key_value.aio.wrappers.statistics import StatisticsWrapper

store = StatisticsWrapper(
    key_value=MemoryStore()
)

# Perform operations
await store.put(key="user:1", value={"name": "Alice"})
await store.get(key="user:1")  # Hit
await store.get(key="user:2")  # Miss

# Check statistics
stats = store.get_statistics()
print(f"Get operations: {stats.get.count}")
print(f"Get hits: {stats.get.hits}")
print(f"Get misses: {stats.get.misses}")
print(f"Hit rate: {stats.get.hit_rate()}")

# Reset statistics
store.reset_statistics()
```

--------------------------------

### RoutingWrapper Example

Source: https://github.com/strawgate/py-key-value/blob/main/docs/wrappers.md

Use RoutingWrapper to direct operations to different stores based on a provided routing function. The function determines the appropriate store for a given collection. This is ideal for multi-backend architectures and custom routing logic.

```python
from key_value.aio.stores.memory import MemoryStore
from key_value.aio.stores.redis import RedisStore
from key_value.aio.wrappers.routing import RoutingWrapper

redis_store = RedisStore(url="redis://localhost:6379/0")
memory_store = MemoryStore()

def route_by_collection(collection: str | None):
    if collection == "sessions":
        return redis_store
    return None

store = RoutingWrapper(
    routing_function=route_by_collection,
    default_store=memory_store
)

# Operations on "sessions" go to Redis, everything else to memory
await store.put(key="s1", value={"user": "alice"}, collection="sessions")
await store.put(key="tmp", value={"data": "value"}, collection="cache")
```

--------------------------------

### Initialize PostgreSQLStore with URL

Source: https://github.com/strawgate/py-key-value/blob/main/docs/stores.md

Initialize the PostgreSQLStore using a connection URL. Ensure PostgreSQL is running and accessible.

```python
from key_value.aio.stores.postgresql import PostgreSQLStore

# Using connection URL
store = PostgreSQLStore(url="postgresql://localhost:5432/mydb")
```

--------------------------------

### Accept AsyncKeyValue Protocol in Framework Initialization

Source: https://github.com/strawgate/py-key-value/blob/main/README.md

Demonstrates how to accept the AsyncKeyValue protocol in a framework's initialization. This allows the framework to be agnostic to the specific storage backend used.

```python
from key_value.aio.protocols.key_value import AsyncKeyValue

class YourFramework:
    def __init__(self, cache: AsyncKeyValue):
        self.cache = cache
```

--------------------------------

### Initialize PostgreSQLStore with Parameters

Source: https://github.com/strawgate/py-key-value/blob/main/docs/stores.md

Initialize the PostgreSQLStore using individual connection parameters. This is an alternative to using a URL.

```python
store = PostgreSQLStore(
    host="localhost",
    port=5432,
    database="mydb",
    user="myuser",
    password="mypass"
)
```

--------------------------------

### User Code: Choosing Storage Backend for Framework

Source: https://github.com/strawgate/py-key-value/blob/main/README.md

Illustrates how a user can instantiate a framework with different storage backends. This highlights the flexibility provided by the AsyncKeyValue protocol, allowing users to switch between memory, Redis, and other stores.

```python
# User's code - they control the storage backend
from your_framework import YourFramework
from key_value.aio.stores.redis import RedisStore
from key_value.aio.stores.memory import MemoryStore

# Development
framework = YourFramework(cache=MemoryStore())

# Production
framework = YourFramework(
    cache=RedisStore(url="redis://localhost:6379/0")
)

# Production with Redis TLS
framework = YourFramework(
    cache=RedisStore(
        url="rediss://redis.example.com:6380/0",
        ssl_ca_certs="/etc/ssl/certs/redis-ca.pem",
    )
)
```

--------------------------------

### Initialize ValkeyStore with Cluster Configuration

Source: https://github.com/strawgate/py-key-value/blob/main/docs/stores.md

Instantiate ValkeyStore for a Valkey Cluster deployment using GlideClusterClientConfiguration. The client is lazily created.

```python
from glide_shared.config import GlideClusterClientConfiguration, NodeAddress

from key_value.aio.stores.valkey import ValkeyStore

cluster_store = ValkeyStore(
    config=GlideClusterClientConfiguration(
        addresses=[NodeAddress("valkey-cluster.example.com", 6379)]
    )
)
```

--------------------------------

### Initialize WindowsRegistryStore

Source: https://github.com/strawgate/py-key-value/blob/main/docs/stores.md

Demonstrates initializing WindowsRegistryStore with hive and registry path.

```python
from key_value.aio.stores.registry import WindowsRegistryStore

store = WindowsRegistryStore(
    hive="HKEY_CURRENT_USER",
    registry_path="Software\py-key-value"
)
```

--------------------------------

### Framework Integration with Different Backends

Source: https://github.com/strawgate/py-key-value/blob/main/docs/getting-started.md

Instantiate YourFramework with different storage backends like MemoryStore or RedisStore. This demonstrates flexibility for development and production environments.

```python
from your_framework import YourFramework
from key_value.aio.stores.memory import MemoryStore
from key_value.aio.stores.redis import RedisStore

# Development
framework = YourFramework(cache=MemoryStore())
```

```python
# Production
framework = YourFramework(
    cache=RedisStore(url="redis://localhost:6379/0")
)
```

```python
# Production with Redis TLS
framework = YourFramework(
    cache=RedisStore(
        url="rediss://redis.example.com:6380/0",
        ssl_ca_certs="/etc/ssl/certs/redis-ca.pem",
    )
)
```

--------------------------------

### Initialize DynamoDBStore

Source: https://github.com/strawgate/py-key-value/blob/main/docs/getting-started.md

Configure the DynamoDBStore by providing the table name and AWS region.

```python
from key_value.aio.stores.dynamodb import DynamoDBStore

store = DynamoDBStore(
    table_name="my-kv-store",
    region_name="us-east-1"
)
```

--------------------------------

### Initialize MemcachedStore

Source: https://github.com/strawgate/py-key-value/blob/main/docs/stores.md

Initialize the MemcachedStore with the host and port of the Memcached server. Ensure Memcached is running.

```python
from key_value.aio.stores.memcached import MemcachedStore

store = MemcachedStore(host="127.0.0.1", port=11211)
```

--------------------------------

### Initialize RedisStore

Source: https://github.com/strawgate/py-key-value/blob/main/docs/getting-started.md

Connect to a Redis instance using a connection URL. For TLS, use 'rediss://' or specify SSL options.

```python
from key_value.aio.stores.redis import RedisStore

store = RedisStore(url="redis://localhost:6379/0")
```

```python
store = RedisStore(
    url="rediss://redis.example.com:6380/0",
    ssl_ca_certs="/etc/ssl/certs/redis-ca.pem",
)
```

--------------------------------

### Initialize AerospikeStore with Host List

Source: https://github.com/strawgate/py-key-value/blob/main/docs/stores.md

Instantiate AerospikeStore using a list of host tuples. Requires specifying namespace and set name.

```python
from key_value.aio.stores.aerospike import AerospikeStore

# Using host list
store = AerospikeStore(
    hosts=[("localhost", 3000)],
    namespace="test",
    set_name="kv-store"
)
```

--------------------------------

### Create a MemoryStore instance

Source: https://github.com/strawgate/py-key-value/blob/main/docs/getting-started.md

Instantiate a MemoryStore for basic usage. This store is in-memory and suitable for testing or simple applications.

```python
from key_value.aio.stores.memory import MemoryStore

store = MemoryStore()
```

--------------------------------

### RaiseOnMissingAdapter

Source: https://github.com/strawgate/py-key-value/blob/main/docs/adapters.md

Explains how RaiseOnMissingAdapter modifies the behavior of get operations to raise an error when a key is not found.

```APIDOC
## RaiseOnMissingAdapter

### Description
The `RaiseOnMissingAdapter` enhances the `get` and `get_many` operations by raising a `KeyNotFoundError` when a requested key does not exist in the store, promoting fail-fast behavior.

### Use Cases
- Enforcing required data
- Fail-fast behavior
- APIs where missing data is an error

### Method
`get`, `get_many`

### Endpoint
N/A (In-memory operation)

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
```python
from key_value.aio.stores.memory import MemoryStore
from key_value.aio.adapters.raise_on_missing import RaiseOnMissingAdapter
from key_value.aio.errors import KeyNotFoundError

adapter = RaiseOnMissingAdapter(key_value=MemoryStore())

# Store a value
await adapter.put(key="user:123", value={"name": "Alice"}, collection="users")

# Get existing key - works normally
user = await adapter.get(key="user:123", collection="users")
print(user)  # {"name": "Alice"}

# Get missing key - raises error
try:
    user = await adapter.get(key="user:999", collection="users")
except KeyNotFoundError as e:
    print(f"Key not found: {e}")

# Batch Operations
try:
    users = await adapter.get_many(keys=["user:1", "user:999", "user:3"], collection="users")
except KeyNotFoundError as e:
    print(f"One or more keys not found: {e}")
```

### Response
#### Success Response (200)
Returns the requested value(s) if found.

#### Error Response (404)
- **KeyNotFoundError**: Raised when a requested key is not found.
```

--------------------------------

### Perform batch operations

Source: https://github.com/strawgate/py-key-value/blob/main/docs/getting-started.md

Improve performance by using batch operations for putting, getting, and deleting multiple items at once.

```python
# Put multiple values at once
await store.put_many(
    keys=["user:1", "user:2", "user:3"],
    values=[
        {"name": "Alice"},
        {"name": "Bob"},
        {"name": "Charlie"}
    ],
    collection="users"
)

# Get multiple values at once
users = await store.get_many(
    keys=["user:1", "user:2", "user:3"],
    collection="users"
)

# Delete multiple values at once
count = await store.delete_many(
    keys=["user:1", "user:2", "user:3"],
    collection="users"
)
```

--------------------------------

### Initialize OpenSearchStore with Existing Client

Source: https://github.com/strawgate/py-key-value/blob/main/docs/stores.md

Initialize the OpenSearchStore with an existing OpenSearchPy client. This allows for more control over the client configuration.

```python
from opensearchpy import AsyncOpenSearch
from key_value.aio.stores.opensearch import OpenSearchStore

client = AsyncOpenSearch(hosts=["https://localhost:9200"])
store = OpenSearchStore(
    opensearch_client=client,
    index_prefix="kv_store"
)
```

--------------------------------

### Initialize AerospikeStore with Existing Client

Source: https://github.com/strawgate/py-key-value/blob/main/docs/stores.md

Instantiate AerospikeStore using an existing aerospike client object. Ensure the client is connected.

```python
import aerospike
client = aerospike.client({"hosts": [("localhost", 3000)]})
client.connect()
store = AerospikeStore(client=client, namespace="test")
```

--------------------------------

### Use PostgreSQLStore for Data Operations

Source: https://github.com/strawgate/py-key-value/blob/main/docs/stores.md

Perform put and get operations asynchronously with PostgreSQLStore. Requires an active store context.

```python
async with store:
    await store.put(key="user_1", value={"name": "Alice"}, collection="users")
    user = await store.get(key="user_1", collection="users")
```

--------------------------------

### Initialize S3Store

Source: https://github.com/strawgate/py-key-value/blob/main/docs/stores.md

Initialize the S3Store for durable, scalable key-value storage on AWS S3. Requires bucket name and region name. Supports large values and lifecycle policies.

```python
from key_value.aio.stores.s3 import S3Store

store = S3Store(
    bucket_name="my-kv-bucket",
    region_name="us-east-1"
)
```

--------------------------------

### Create and Use LoggingWrapper with MemoryStore in Python

Source: https://github.com/strawgate/py-key-value/blob/main/docs/api/index.md

Demonstrates how to instantiate a MemoryStore and wrap it with LoggingWrapper to add logging capabilities. This is useful for debugging or monitoring store operations.

```python
from key_value.aio.stores.memory import MemoryStore
from key_value.aio.wrappers.logging import LoggingWrapper

# Create a store with logging
store = LoggingWrapper(MemoryStore())

# Use the store
await store.put("key", "value")
result = await store.get("key")
```

--------------------------------

### Initialize FirestoreStore

Source: https://github.com/strawgate/py-key-value/blob/main/docs/stores.md

Instantiate FirestoreStore with Google credentials, database name, and optional sanitization strategies for keys and collections.

```python
from key_value.aio.stores.firestore import (
    FirestoreStore,
    FirestoreV1CollectionSanitizationStrategy,
    FirestoreV1KeySanitizationStrategy,
)

store = FirestoreStore(
    credentials=google_credentials,
    database="firestore-db",
    key_sanitization_strategy=FirestoreV1KeySanitizationStrategy(),
    collection_sanitization_strategy=FirestoreV1CollectionSanitizationStrategy(),
)
```

--------------------------------

### Initialize OpenSearchStore with URL and API Key

Source: https://github.com/strawgate/py-key-value/blob/main/docs/stores.md

Initialize the OpenSearchStore using URL and API key. Collections are stored in separate indices with values in flattened fields.

```python
from key_value.aio.stores.opensearch import OpenSearchStore

# Using URL and API key
store = OpenSearchStore(
    url="https://localhost:9200",
    api_key="your-api-key",
    index_prefix="kv_store"
)
```

--------------------------------

### Combining Adapters and Wrappers

Source: https://github.com/strawgate/py-key-value/blob/main/docs/adapters.md

Illustrates how to combine multiple wrappers (e.g., encryption, compression) with adapters like PydanticAdapter for a layered storage solution.

```APIDOC
## Combining Adapters and Wrappers

### Description
This example shows how to chain multiple wrappers (like encryption and compression) around a store before passing it to an adapter (like `PydanticAdapter`) to achieve a layered storage solution with enhanced security and efficiency.

### Method
`put`, `get` (and other methods inherited from underlying adapters/wrappers)

### Endpoint
N/A (In-memory operation)

### Parameters
N/A

### Request Example
```python
from pydantic import BaseModel
from key_value.aio.stores.memory import MemoryStore
from key_value.aio.wrappers.encryption.fernet import FernetEncryptionWrapper
from key_value.aio.wrappers.compression import CompressionWrapper
from key_value.aio.adapters.pydantic import PydanticAdapter
from cryptography.fernet import Fernet

class User(BaseModel):
    name: str
    email: str

# Create an encrypted and compressed store
wrapped_store = CompressionWrapper(
    key_value=FernetEncryptionWrapper(
        key_value=MemoryStore(),
        fernet=Fernet(Fernet.generate_key())
    )
)

# Wrap the layered store with PydanticAdapter for type safety
adapter = PydanticAdapter(
    key_value=wrapped_store,
    pydantic_model=User
)

# Store data - it will be Pydantic validated, encrypted, and compressed
await adapter.put(key="user:123", value=User(name="Alice", email="alice@example.com"))

# Retrieve data - it will be decrypted, decompressed, and Pydantic validated
retrieved_user = await adapter.get(key="user:123")
print(retrieved_user)
```

### Response
#### Success Response (200)
Returns the processed value (e.g., Pydantic model instance) after decryption and decompression.

#### Response Example
```json
{
  "name": "Alice",
  "email": "alice@example.com"
}
```
```

--------------------------------

### Async Key-Value Protocols

Source: https://github.com/strawgate/py-key-value/blob/main/README.md

Defines the asynchronous interface for interacting with key-value stores, including methods for getting, putting, deleting, and managing Time-To-Live (TTL) for data.

```python
get(key: str, collection: str | None = None) -> dict[str, Any] | None:
get_many(keys: list[str], collection: str | None = None) -> list[dict[str, Any] | None]:

put(key: str, value: dict[str, Any], collection: str | None = None, ttl: SupportsFloat | None = None) -> None:
put_many(keys: list[str], values: Sequence[dict[str, Any]], collection: str | None = None, ttl: SupportsFloat | None = None) -> None:

delete(key: str, collection: str | None = None) -> bool:
delete_many(keys: list[str], collection: str | None = None) -> int:

ttl(key: str, collection: str | None = None) -> tuple[dict[str, Any] | None, float | None]:
ttl_many(keys: list[str], collection: str | None = None) -> list[tuple[dict[str, Any] | None, float | None]]:
```

--------------------------------

### Create a Custom Adapter

Source: https://github.com/strawgate/py-key-value/blob/main/docs/adapters.md

Illustrates how to create a custom adapter by wrapping an AsyncKeyValue instance and defining custom methods. The custom_method retrieves data from a specific collection.

```python
from key_value.aio.protocols.key_value import AsyncKeyValue

class CustomAdapter:
    def __init__(self, key_value: AsyncKeyValue):
        self._key_value = key_value

    async def custom_method(self, key: str) -> dict:
        # Implement custom logic
        value = await self._key_value.get(key=key, collection="custom")
        if value is None:
            return {}
        return value
```

--------------------------------

### TimeoutWrapper Example

Source: https://github.com/strawgate/py-key-value/blob/main/docs/wrappers.md

The TimeoutWrapper enforces timeout constraints on all store operations. If an operation exceeds the specified timeout, an asyncio.TimeoutError is raised. This is useful for preventing operations from hanging indefinitely and enforcing SLA requirements.

```python
from key_value.aio.stores.redis import RedisStore
from key_value.aio.wrappers.timeout import TimeoutWrapper

store = TimeoutWrapper(
    key_value=RedisStore(url="redis://localhost:6379/0"),
    timeout=1.0  # 1 second timeout
)

# Raises asyncio.TimeoutError if operation takes > 1 second
user = await store.get(key="user:123", collection="users")
```

--------------------------------

### TTLClampWrapper Example

Source: https://github.com/strawgate/py-key-value/blob/main/docs/wrappers.md

The TTLClampWrapper ensures that Time-To-Live (TTL) values for operations are within a specified minimum and maximum range. This is beneficial for enforcing backend-specific TTL limitations or preventing excessively long or short TTLs.

```python
from key_value.aio.stores.memory import MemoryStore
from key_value.aio.wrappers.ttl_clamp import TTLClampWrapper

store = TTLClampWrapper(
    key_value=MemoryStore(),
    min_ttl=60,      # Minimum 1 minute
    max_ttl=86400    # Maximum 1 day
)

# TTL is clamped to range [60, 86400]
await store.put(
    key="test",
    value={"data": "value"},
    ttl=30  # Clamped to 60
)
```

--------------------------------

### SingleCollectionWrapper Example

Source: https://github.com/strawgate/py-key-value/blob/main/docs/wrappers.md

Use SingleCollectionWrapper to force all operations to use a specific collection, ignoring any collection parameter passed to the operation. This is useful for simplifying stores that do not require multiple collections or for migrating from non-collection-based stores.

```python
from key_value.aio.stores.memory import MemoryStore
from key_value.aio.wrappers.single_collection import SingleCollectionWrapper

store = SingleCollectionWrapper(
    key_value=MemoryStore(),
    collection="default"
)

# All operations use "default" collection regardless of parameter
await store.put(key="test", value={"data": "value"}, collection="ignored")
```

--------------------------------

### Query DuckDBStore Data with SQL

Source: https://github.com/strawgate/py-key-value/blob/main/docs/stores.md

Shows how to store data and then query it directly using DuckDB's SQL interface.

```python
# After storing some data
await store.put("user:1", {"name": "Alice", "age": 30}, collection="users")
await store.put("user:2", {"name": "Bob", "age": 25}, collection="users")

# Query directly via DuckDB
result = store._connection.execute("""
    SELECT key, value->>'name' as name
    FROM kv_entries
    WHERE collection = 'users'
""").fetchall()
```

--------------------------------

### Initialize MongoDBStore

Source: https://github.com/strawgate/py-key-value/blob/main/docs/stores.md

Initialize the MongoDBStore for document-oriented data. Requires a MongoDB connection URL. Suitable for applications already using MongoDB.

```python
from key_value.aio.stores.mongodb import MongoDBStore

store = MongoDBStore(url="mongodb://localhost:27017/test")
```
Tokens:
Raw