Package `couchdb3`

CouchDB3

CouchDB3 is a wrapper around the CouchDB API. For more detailed information, please refer to the documentation.

Disclaimer

Big parts of the documentation (and thus docstrings) have been copied from CouchDB's API's great official documentation.

Requirements

Python version >= 3.11
CouchDB version 3.x

Installation

Installing via PyPi

pip install couchdb3

Installing via Github

python -m pip install git+https://github.com/n-Vlahovic/couchdb3.git

Installing from source

git clone https://github.com/n-Vlahovic/couchdb3
python -m pip install -e couchdb3

Import styles

All public classes are available flat from couchdb3 for backward compatibility. Explicit subpackage imports are also supported for clarity:

# Flat (backward-compatible)
from couchdb3 import Server, AsyncServer, Document

# Explicit sync subpackage
from couchdb3.sync import Server, Database, Partition

# Explicit async subpackage
from couchdb3.aio import AsyncServer, AsyncDatabase, AsyncPartition

# Shared types always from couchdb3 directly
from couchdb3 import Document, ViewResult, ViewRow, exceptions

Quickstart — Sync

Connecting to a database server

import couchdb3

client = couchdb3.Server(
    "http://user:password@127.0.0.1:5984"
)

# Checking if the server is up
print(client.up())
# True

User and password can also be passed as keyword parameters:

client = couchdb3.Server(
    "127.0.0.1:5984",  # Scheme omitted - will assume http protocol
    user="user",
    password="password"
)

Both approaches are equivalent. Clients can also be used as context managers:

with couchdb3.Server("http://user:password@127.0.0.1:5984") as client:
    # Do stuff
    ...

Getting or creating a database

dbname = "mydb"
db = client.get(dbname) if dbname in client else client.create(dbname)
print(db)
# Database: mydb

Creating a document

mydoc = {
    "_id": "mydoc-id",
    "name": "Hello",
    "type": "World"
}
print(db.save(mydoc))
# ('mydoc-id', True, '1-24fa3b3fd2691da9649dd6abe3cafc7e')

Note: Database.save requires the document to have an id (i.e. a key _id), Database.create does not.

Updating a document

To update an existing document, retrieving the revision is paramount. In the example below, dbdoc contains the key _rev and the builtin dict.update function is used to update the document before saving it.

mydoc = {
    "_id": "mydoc-id",
    "name": "Hello World",
    "type": "Hello World"
}
dbdoc = db.get(mydoc["_id"])
dbdoc.update(mydoc)
print(db.save(dbdoc))
# ('mydoc-id', True, '2-374aa8f0236b9120242ca64935e2e8f1')

Alternatively, one can use Database.rev to fetch the latest revision and overwrite the document:

mydoc = {
    "_id": "mydoc-id",
    "_rev": db.rev("mydoc-id"),
    "name": "Hello World",
    "type": "Hello World"
}
print(db.save(mydoc))
# ('mydoc-id', True, '3-d56b14b7ffb87960b51d03269990a30d')

Deleting a document

To delete a document, the docid and rev are needed:

docid = "mydoc-id"
print(db.delete(docid=docid, rev=db.rev(docid)))  # Fetch the revision on the go
# True

Fetching documents

# Fetch a single document (returns None if not found)
doc = db.get("mydoc-id")

# Subscript shorthand (raises KeyError if not found)
doc = db["mydoc-id"]

# Fetch all documents (metadata only)
result = db.all_docs()  # ViewResult

# Fetch all documents with full bodies
result = db.all_docs(include_docs=True)
for row in result.rows:
    print(row.id, row.doc)

# Fetch specific documents by ID
result = db.all_docs(keys=["id-1", "id-2"], include_docs=True)

# Batch fetch by ID
result = db.bulk_get(docs=[{"id": "id-1"}, {"id": "id-2"}])
for item in result:
    print(item["id"], item["docs"][0]["ok"])

Views

# 1. Create a design document with a map function
db.put_design("my-ddoc", views={
    "my-view": {
        "map": "function(doc) { if (doc.type === 'post') emit(doc._id, null); }"
    }
})

# 2. Query the view
result = db.view("my-ddoc", "my-view")                        # ViewResult
result = db.view("my-ddoc", "my-view", include_docs=True, limit=10)

# 3. Iterate results
for row in result.rows:
    print(row.id, row.key, row.value)

Mango queries

# 1. Create an index
db.save_index({"fields": ["type", "name"]}, ddoc="my-ddoc", name="type-name-idx")

# 2. Query with a selector
result = db.find({"type": {"$eq": "post"}}, fields=["_id", "name"], limit=10)
for doc in result["docs"]:
    print(doc)

# 3. Inspect the query plan
plan = db.explain({"type": {"$eq": "post"}}, limit=10)

Working with partitions

For a partitioned database, the Partition class offers a wrapper around partitions (acting similarly to collections in Mongo).

from couchdb3.sync import Server, Database, Partition

client: Server = Server(...)
db: Database = client["some-db"]
partition: Partition = db.get_partition("partition_id")

Partition instances prepend the partition's ID to document IDs (partition-id:doc-id) for simpler user interaction:

doc_id = "test-id"
print(doc_id in partition)  # no need to prepend the partition's ID
rev = partition.rev(doc_id)
partition.save({
    "_id": doc_id,  # no need to prepend the partition's ID
    "_rev": rev,
    ...
})

The partition ID will only be prepended if the document ID does not already start with it:

doc_id = "partition_id:test-id"
print(doc_id in partition)
rev = partition.rev(doc_id)
partition.save({
    "_id": doc_id,
    "_rev": rev,
    ...
})

Quickstart — Async

Connecting to a database server

import asyncio
from couchdb3.aio import AsyncServer

async def main():
    async with AsyncServer("http://user:password@127.0.0.1:5984") as client:
        print(await client.up())
        # True

asyncio.run(main())

Getting or creating a database

async with AsyncServer("http://user:password@127.0.0.1:5984") as client:
    dbname = "mydb"
    all_dbs = await client.all_dbs()
    db = await client.get(dbname) if dbname in all_dbs else await client.create(dbname)

Creating and fetching documents

async with AsyncServer("http://user:password@127.0.0.1:5984") as client:
    db = await client.get("mydb")

    # Save a document
    _id, ok, _rev = await db.save({"_id": "mydoc-id", "name": "Hello", "type": "World"})

    # Fetch a document
    doc = await db.get("mydoc-id")
    print(doc)

    # Delete a document
    await db.delete(docid="mydoc-id", rev=await db.rev("mydoc-id"))

Async context manager (manual lifecycle)

from couchdb3.aio import AsyncServer

client = AsyncServer("http://user:password@127.0.0.1:5984")
db = await client.get("mydb")
doc = await db.get("mydoc-id")
await client.aclose()  # must be called explicitly when not using async with

Sub-modules

couchdb3.aio: Asynchronous CouchDB client subpackage …
couchdb3.document
couchdb3.exceptions
couchdb3.sync: Synchronous CouchDB client subpackage …
couchdb3.utils
couchdb3.view