mongoKV /ˈmɑːŋɡoʊ kiː ˈvæljuː/

User Guide

mongoKV is a tiny key–value store wrapper around MongoDB using PyMongo.

• One collection per instance, one document per key
• Works synchronously and asynchronously with the same API
• Stores any BSON-serializable Python object as the value, keys are strings
• Automatically generates IDs when you don’t care about the key
• Fully tested. MongoDB handles locking, durability, and concurrency.
• This project is intentionally small, stable, and boring. Breaking changes are avoided
• mongoKV is not a cache or a Redis replacement — it’s a simple, durable KV API on top of MongoDB

Think of it as MongoDB as a dict instead of a database, for lazy programmers who just want to save some simple data.

Installation

pip install mongokv

You’ll also need a running MongoDB instance and a valid connection URI, for example:

mongodb://localhost:27017

Check out An Introduction to MongoDB Connection Strings for help on getting a MongoDB instance up and running. A super easy way to do this is to use Atlas, that way your instance is in the cloud and you won't need a Mongo server running on your local device.

Core Concepts

Each key is stored as a single MongoDB document:

{
  "_id": "<str(key)>",
  "value": <your_data>
}

• Keys are always stored as strings (str(key)), but you can pass any object with a sensible str().
• Values must be BSON-serializable (anything PyMongo can store: dicts, lists, ints, strings, datetimes, etc.).
• This is a normal MongoDB collection. You can always access the same data directly using PyMongo as your needs grow — no migrations required.
• Safe across threads, processes, and ASGI workers — MongoDB handles locking and atomicity.
• The library auto-detects whether it’s running inside an asyncio event loop:
  • Inside async code: methods return coroutines → you must await them.
  • Outside async code: methods run synchronously and return results directly.

Same methods, two modes. Start simple, grow into MongoDB when you need more.

Quickstart

Synchronous usage

from mongokv import Mkv

db = Mkv("mongodb://localhost:27017")

# Set a value with an explicit key
db.set("username", "harrison")

# Get it back
print(db.get("username"))  # -> "harrison"

# Set a value with an auto-generated key
note_id = db.set(None, {"title": "mongoKV", "tags": ["mongo", "kv"]})
print(note_id)  # -> e.g. "677eec65e92eeca23513fe99"

# List all keys
print(db.all())  # -> ["username", "677eec65e92eeca23513fe99", ...]

# Remove a key
db.remove("username")

# Clear everything
db.purge()

# Close connections when you’re done
db.close()

Asynchronous usage

import asyncio
from mongokv import Mkv

db = Mkv("mongodb://localhost:27017")

async def main():
    # Set a value with explicit key
    await db.set("counter", 1)

    # Increment and save again
    current = await db.get("counter", 0)
    await db.set("counter", current + 1)

    # Auto-generated key
    new_id = await db.set(None, {"hello": "world"})
    print("New id:", new_id)

    # List all keys
    keys = await db.all()
    print("Keys:", keys)

    # Remove key
    deleted = await db.remove("counter")
    print("Deleted:", deleted)

    # Purge all keys
    await db.purge()

    # Close connections
    await db.close()

asyncio.run(main())

class Mkv

Create a new mongoKV instance.

Parameters

• mongo_uri (str): MongoDB connection string.
• db_name (str, default "mkv"): Database name.
• collection_name (str, default "kv"): Collection name.

Internally it maintains:

• An async client & collection:
  • self._async_client = AsyncMongoClient(mongo_uri)
  • self.db, self.collection
• A sync client & collection:
  • self._sync_client = MongoClient(mongo_uri)
  • self._sync_db, self._sync_collection

These are used automatically based on sync/async context.

set

Set a value for a given key. If key is None, a new ObjectId-based key will be generated.

Behavior

• Sync: executes immediately, returns the string key.
• Async: returns a coroutine; awaiting it yields the string key.

Parameters

• key (str or None):
  • If not None: converted to str(key) and used as _id.
  • If None: a new ObjectId() is generated and used as _id.
• value (Any): Any BSON-serializable object to store.

Returns

str: The key used for this record.

Examples (sync)

db.set("user:1", {"name": "Alice"})
auto_id = db.set(None, {"name": "Bob"})

Examples (async)

key = await db.set("session:123", {"state": "active"})
new_id = await db.set(None, {"foo": "bar"})

get

Fetch the value for a given key.

Behavior

• Sync: executes immediately.
• Async: returns a coroutine; awaiting it yields the result.
• Missing keys:
  • If default is not provided, a missing key raises KeyError.
  • If default is provided (even None), that value is returned when missing.

Parameters

• key (str): Key to look up (converted to str(key) for _id).
• default (Any, optional): Fallback to return if the key does not exist. If omitted, missing keys raise KeyError.

Returns

The stored value. If the key is missing, returns default if provided, otherwise raises KeyError.

Examples (sync)

# strict (dict-like)
try:
    user = db.get("user:1")
except KeyError:
    user = None

# fallback behavior
user = db.get("user:1", default={})
missing = db.get("nope", default="N/A")   # -> "N/A"
none_ok = db.get("nope", default=None)    # -> None

Examples (async)

# strict (dict-like)
try:
    user = await db.get("user:1")
except KeyError:
    user = None

# fallback behavior
user = await db.get("user:1", default={})
none_ok = await db.get("nope", default=None)

remove

Delete a single document by key.

Behavior

• Sync: executes immediately, returns True if a document was deleted, False otherwise.
• Async: returns a coroutine; awaiting it yields True or False.

Parameters

• key (str): Key to remove (converted to str(key) for _id).

Returns

bool: True if a document was deleted, False otherwise.

Examples (sync)

deleted = db.remove("user:1")
if deleted:
    print("User removed")

Examples (async)

deleted = await db.remove("user:1")

all

List all keys (all _id values) in the collection.

Behavior

• Sync: executes immediately, returns a list[str].
• Async: returns a coroutine; awaiting it yields a list[str].

Returns

list[str]: All keys currently stored.

Examples (sync)

keys = db.all()
print(keys)  # -> ["user:1", "note:abc", ...]

Examples (async)

keys = await db.all()

purge

Delete all documents in the collection. Use with care.

Behavior

• Sync: executes immediately, returns True once the operation completes.
• Async: returns a coroutine; awaiting it yields True.

Returns

bool: Always True if no exception is raised.

Examples (sync)

db.purge()  # All keys gone

Examples (async)

await db.purge()

close

Close both the async and sync MongoDB clients.

Behavior

• Sync:
  • Calls self._async_client.close() using asyncio.run().
  • Calls self._sync_client.close().
  • Returns None.
• Async: returns a coroutine that:
  • await self._async_client.close()
  • self._sync_client.close()
  • Awaited result is None.

Returns

None

Examples (sync)

db.close()

Examples (async)

await db.close()

Closing is optional in short-lived scripts, but recommended in long-running apps and tests.

When is it sync vs async?

The library uses:

def in_async() -> bool:
    try:
        asyncio.get_running_loop()
        return True
    except RuntimeError:
        return False

• If there is a running event loop, methods like set, get, etc. return coroutines.
• If there is no running loop, they execute synchronously.
• If you do not await your method calls in an async context there will be an error because your a returning a non-awaited coroutine.

Error Handling

The current implementation doesn’t wrap MongoDB errors – any connection or operation issue will bubble up as a PyMongo/AsyncMongo exception. Typical things you might see:

• Connection failures
• Serialization errors for non-BSON-compatible values
• Authentication/authorization errors if your URI is locked down

You can catch them at the call site:

from pymongo.errors import PyMongoError

try:
    db.set("x", object())
except PyMongoError as e:
    print("Mongo blew up:", e)