CouchDB3

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

Contents

• Disclaimer
• Requirements
• Installation
• Import styles
• Quickstart
  • Connecting to a database server
  • Getting or creating a database
  • Creating a document
  • Updating a document
  • Deleting a document
  • Fetching documents
  • Views
  • Mango queries
  • Changes feed
  • Working with partitions
• Async client
  • Connecting to a database server
  • Getting or creating a database
  • Creating a document
  • Updating a document
  • Deleting a document
  • Fetching documents
  • Views
  • Mango queries
  • Changes feed
  • Controlling concurrency
  • Working with async partitions

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

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 into the Server constructor as keyword parameters, e.g.

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

Both approaches are equivalent, i.e. in both cases the instance's scheme,host,port,user,password will be identical.

Further, clients can be used with 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)

Changes feed

Database.changes() wraps GET /{db}/_changes (and POST /{db}/_changes when filtering by document IDs or a Mango selector).

# All recent changes (normal feed — returns immediately)
result = db.changes()
for row in result["results"]:
    print(row["id"], row["changes"])

# Only changes since a known sequence
result = db.changes(since="now")  # wait marker — use last_seq to poll later
last_seq = result["last_seq"]
result = db.changes(since=last_seq)  # only changes after that point

# Filter to specific document IDs (POST /_changes?filter=_doc_ids)
result = db.changes(doc_ids=["doc-1", "doc-2"])

# Filter with a Mango selector (POST /_changes?filter=_selector)
result = db.changes(selector={"type": {"$eq": "post"}})

# Include the full document body in each result
result = db.changes(include_docs=True)
for row in result["results"]:
    print(row["id"], row.get("doc"))

# Long-poll: server holds the connection open until at least one change arrives
result = db.changes(feed="longpoll", since="now", timeout=30_000)

Note — continuous and eventsource feeds

feed="continuous" and feed="eventsource" are not supported by changes() and will raise ValueError. These modes stream newline-delimited JSON over a persistent HTTP connection, which requires response-body streaming rather than a single buffered read. Support for streaming feeds will be added in a future changes_stream() method. For now, a polling loop over feed="longpoll" with since=last_seq is a practical alternative for most real-time use cases.

Working with partitions

For a partitioned database, the couchdb3.sync.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 append the partition's ID to the document IDs (partition-id:doc-id) for a simpler user interaction, e.g.

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

The partition ID will only be appended provided document IDs do not start with partition-id, e.g. the following will work and be equivalent to the previous example

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

Async client

couchdb3 ships an async client built on httpx.AsyncClient. It mirrors the sync API exactly, with async def methods and async with context manager support.

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())

user and password can also be passed as keyword parameters, and the manual lifecycle is also supported via await client.aclose():

client = AsyncServer("127.0.0.1:5984", user="user", password="password")
# ... do stuff ...
await client.aclose()

Note: AsyncServer does not implement __getitem__ — Python does not allow __getitem__ to be a coroutine. Use await client.get(name) instead of client[name]. Similarly, for the async equivalent of the sync name in server check, use await client.has_db(name).

Getting or creating a database

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

Creating a document

async with AsyncServer("http://user:password@127.0.0.1:5984") as client:
    db = await client.get("mydb")
    mydoc = {"_id": "mydoc-id", "name": "Hello", "type": "World"}
    print(await db.save(mydoc))
    # ('mydoc-id', True, '1-24fa3b3fd2691da9649dd6abe3cafc7e')

Updating a document

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

Deleting a document

docid = "mydoc-id"
print(await db.delete(docid=docid, rev=await db.rev(docid)))
# True

Fetching documents

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

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

# Batch fetch by ID
result = await 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
await db.put_design(
    "my-ddoc",
    views={"my-view": {"map": "function(doc) { if (doc.type === 'post') emit(doc._id, null); }"}},
)

# 2. Query the view
result = await db.view("my-ddoc", "my-view")
result = await 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
await db.save_index({"fields": ["type", "name"]}, ddoc="my-ddoc", name="type-name-idx")

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

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

Changes feed

AsyncDatabase.changes() mirrors the sync API exactly.

# All recent changes
result = await db.changes()
for row in result["results"]:
    print(row["id"], row["changes"])

# Only changes since a known sequence
result = await db.changes(since="now")
last_seq = result["last_seq"]
result = await db.changes(since=last_seq)

# Filter to specific document IDs (POST /_changes?filter=_doc_ids)
result = await db.changes(doc_ids=["doc-1", "doc-2"])

# Filter with a Mango selector (POST /_changes?filter=_selector)
result = await db.changes(selector={"type": {"$eq": "post"}})

# Include the full document body in each result
result = await db.changes(include_docs=True)
for row in result["results"]:
    print(row["id"], row.get("doc"))

# Long-poll: server holds the connection open until at least one change arrives
result = await db.changes(feed="longpoll", since="now", timeout=30_000)

Note — continuous and eventsource feeds

feed="continuous" and feed="eventsource" are not supported by changes() and will raise ValueError. These modes stream newline-delimited JSON over a persistent HTTP connection, which requires response-body streaming rather than a single buffered read. Support for streaming feeds will be added in a future changes_stream() method. For now, a polling loop over feed="longpoll" with since=last_seq is a practical alternative for most real-time use cases.

Controlling concurrency

httpx.AsyncClient pools connections internally. For application-level concurrency control use asyncio.Semaphore, and optionally tune the connection pool via httpx.Limits:

import asyncio
import httpx
from couchdb3.aio import AsyncServer

# Limit to 10 concurrent CouchDB operations
sem = asyncio.Semaphore(10)


async def fetch(db, docid):
    async with sem:
        return await db.get(docid)


# Optionally tune the underlying connection pool
client = AsyncServer(
    "http://user:password@127.0.0.1:5984",
    session=httpx.AsyncClient(
        limits=httpx.Limits(max_connections=20, max_keepalive_connections=10)
    ),
)

Working with async partitions

from couchdb3.aio import AsyncServer, AsyncDatabase, AsyncPartition

async with AsyncServer("http://user:password@127.0.0.1:5984") as client:
    db: AsyncDatabase = await client.get("some-db")
    partition: AsyncPartition = await db.get_partition("partition_id")

    doc_id = "test-id"
    await partition.save(
        {
            "_id": doc_id,  # no need to append the partition's ID
            "type": "example",
        }
    )
    doc = await partition.get(doc_id)