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

User Guide

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

• 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

Think of it as: “I am a lazy programmer and I just want set and get over Mongo.” or Mongo as a dict instead of a database.

Installation

pip install mongokv

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

mongodb://localhost:27017

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 that has a sensible str().
• Values must be BSON-serializable (anything PyMongo can store: dicts, lists, ints, strings, datetimes, etc.).
• 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.

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 asyncio.run(self._async_client.close()).
  • 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.

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.

Examples:

# Sync context
db = Mkv("mongodb://localhost:27017")
db.set("x", 1)               # OK
value = db.get("x")          # OK

# Async context
async def foo():
    db = Mkv("mongodb://localhost:27017")
    await db.set("x", 1)     # must await
    value = await db.get("x")

# Don’t do this in async code:
db.set("x", 1)   # returns a coroutine, but you never await it

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)