Module couchdb3.aio

Asynchronous CouchDB client subpackage.

from couchdb3.aio import AsyncServer, AsyncDatabase, AsyncPartition

Shared types (Document, ViewResult, ViewRow, exceptions, utils) are not re-exported here — import them from couchdb3 directly.

Sub-modules

couchdb3.aio.async_base

couchdb3.aio.async_database

couchdb3.aio.async_server

Classes

class AsyncDatabase (name: str,
*,
url: str | None = None,
port: int | None = None,
user: str | None = None,
password: str | None = None,
disable_ssl_verification: bool = False,
auth_method: str | None = None,
timeout: int | None = 300,
session: httpx.AsyncClient | None = None)

Expand source code

class AsyncDatabase(AsyncBase):
    """
    Async CouchDB database client. Mirrors `Database` with `async def` methods throughout.

    Note: `__getitem__` is not supported on async classes — Python does not allow
    `__getitem__` to be a coroutine. Use `await db.get(docid)` instead.
    """

    def __init__(
        self,
        name: str,
        *,
        url: str | None = None,
        port: int | None = None,
        user: str | None = None,
        password: str | None = None,
        disable_ssl_verification: bool = False,
        auth_method: str | None = None,
        timeout: int | None = DEFAULT_TIMEOUT,
        session: httpx.AsyncClient | None = None,
        _server: Any = None,
    ) -> None:
        """

        Parameters
        ----------
        name : str
            The name of the database.
        url : str
            The url of the CouchDB server formatted as `scheme://user:password@host:port`. For example:

                "http://user:password@127.0.0.1:5984"
                "https://couchdb.example.com"
        port : int
            The port of the CouchDB server. Can also be supplied via the url.
        user : str
            The CouchDB admin username. Can also be supplied via the url.
        password : str
            The CouchDB admin password. Can also be supplied via the url.
        disable_ssl_verification : bool
            Controls whether to verify the server's TLS certificate. Set to `True` when connecting
            to a server with self-signed TLS certificates. Default `False`.
        auth_method : str
            Authentication method. Choices are `cookie` or `basic`.
            Default is `couchdb3.utils.DEFAULT_AUTH_METHOD`.
        timeout : int
            The default timeout for requests. Default c.f. `couchdb3.utils.DEFAULT_TIMEOUT`.
        session : httpx.AsyncClient
            A specific async client to use. Optional — if not provided, a new client will be
            initialized.
        _server : AsyncServer
            The owning `AsyncServer` instance. Set internally by `AsyncServer.get()` to keep
            the server alive for the lifetime of this database object. Not part of the public
            constructor API — pass `None` (default) when constructing an `AsyncDatabase`
            directly.
        """
        super().__init__(
            url=url,
            session=session,
            port=port,
            user=user,
            password=password,
            disable_ssl_verification=disable_ssl_verification,
            auth_method=auth_method,
            timeout=timeout,
        )
        if not validate_db_name(name=name):
            raise NameComplianceError(
                "Database name does not comply with the CouchDB requirements. "
                "See https://docs.couchdb.org/en/latest/api/database/common.html#put--db."
            )
        self.name = name
        self.root = name
        self._server = _server

    @property
    def server(self):
        """
        The `AsyncServer` instance this database was obtained from, or `None` if the database
        was constructed directly (i.e. not via `AsyncServer.get()`).

        Read-only. Setting this attribute raises `AttributeError`.

        Returns
        -------
        AsyncServer | None
        """
        return self._server

    def __repr__(self) -> str:
        """
        Basic repr.

        Returns
        -------
        str : The instance's representation.
        """
        return f"{super().__repr__()}: {self.name}"

    async def all_docs(
        self,
        partition: str | None = None,
        keys: Iterable[str] | None = None,
        **kwargs,
    ) -> ViewResult:
        """
        Executes the built-in _all_docs view, returning all documents in the database
        (or partition).

        Parameters
        ----------
        partition : str
            Filter using the partition's name (only valid for partitioned databases).
            Default is `None`.
        keys : Iterable[str]
            Return only documents where the key matches one of the keys specified.
            Default is `None`.
        kwargs
            Further `AsyncDatabase.view` parameters.

        Returns
        -------
        ViewResult
        """
        return await self.view(
            f"_partition/{partition}/_all_docs" if partition else "_all_docs",
            keys=keys,
            **kwargs,
        )

    async def design_docs(
        self,
        *,
        conflicts: bool | None = None,
        descending: bool | None = None,
        endkey: str | None = None,
        include_docs: bool | None = None,
        keys: Iterable[str] | None = None,
        limit: int | None = None,
        skip: int | None = None,
        startkey: str | None = None,
        update_seq: bool | None = None,
    ) -> ViewResult:
        """
        Executes the built-in `_design_docs` view, returning all the design documents in the database.

        This is a shorthand for `_all_docs` filtered to the `_design/` key range.

        Parameters
        ----------
        conflicts : bool
            Include conflicts information. Ignored if `include_docs` isn't `True`. Default is `None`.
        descending : bool
            Return the documents in descending order by key. Default is `None`.
        endkey : str
            Stop returning records when the specified key is reached. Default is `None`.
        include_docs : bool
            Include the associated document with each row. Default is `None`.
        keys : Iterable[str]
            Return only documents where the key matches one of the keys specified in the argument.
            Default is `None`.
        limit : int
            Limit the number of the returned documents. Default is `None`.
        skip : int
            Skip this number of records before starting to return the results. Default is `None`.
        startkey : str
            Return records starting with the specified key. Default is `None`.
        update_seq : bool
            Whether to include an `update_seq` value indicating the sequence id of the database.
            Default is `None`.

        Returns
        -------
        ViewResult
        """
        return ViewResult(
            **(
                await self._get(
                    resource="_design_docs",
                    query_kwargs=rm_nones_from_dict(
                        {
                            "conflicts": conflicts,
                            "descending": descending,
                            "endkey": endkey,
                            "include_docs": include_docs,
                            "keys": keys,
                            "limit": limit,
                            "skip": skip,
                            "startkey": startkey,
                            "update_seq": update_seq,
                        }
                    ),
                )
            ).json()
        )

    async def bulk_docs(
        self,
        docs: list[dict | Document],
        new_edits: bool = True,
    ) -> list[dict]:
        """
        Create or update multiple documents in a single request.

        Parameters
        ----------
        docs : list[dict | Document]
            List of document objects.
        new_edits : bool
            If `False`, prevents the database from assigning new revision IDs. Default `True`.

        Returns
        -------
        list[dict] : Each item contains `id`, `ok`, and `rev`.
        """
        return (
            await self._post(resource="_bulk_docs", body={"docs": docs, "new_edits": new_edits})
        ).json()

    async def bulk_get(
        self,
        docs: list[dict | Document],
        revs: bool = False,
    ) -> list[dict]:
        """
        Query several documents in bulk.

        Parameters
        ----------
        docs : list[dict | Document]
            List of document objects, with `id`, and optionally `rev` and `atts_since`.
        revs : bool
            Give the revisions history. Default `False`.

        Returns
        -------
        list[dict]
        """
        return (
            (
                await self._post(
                    resource="_bulk_get",
                    body={"docs": [extract_document_id_and_rev(_) for _ in docs]},
                    query_kwargs={"revs": revs},
                )
            )
            .json()
            .get("results", [])
        )

    async def compact(self, ddoc: str | None = None) -> bool:
        """
        Request compaction of the database.

        Parameters
        ----------
        ddoc : str
            A design document name. If provided, compacts the view indexes for that ddoc.

        Returns
        -------
        bool: `True` upon compaction request successfully sent.
        """
        resource = "_compact"
        if ddoc:
            resource += f"/{ddoc}"
        return (await self._post(resource=resource)).json().get("ok")

    async def copy(
        self,
        docid: str,
        destid: str,
        rev: str | None = None,
        destrev: str | None = None,
    ) -> tuple[str, bool, str]:
        """
        Copy an existing document to a new or existing document.

        Parameters
        ----------
        docid : str
            The ID of the document to copy.
        destid : str
            The target document's ID.
        rev : str
            A specific revision of the document to copy.
        destrev : str
            If the target document already exists, its current revision.

        Returns
        -------
        tuple[str, bool, str] : (id, ok, rev)
        """
        destination = destid
        if destrev:
            destination += f"?rev={destrev}"
        data = (
            await self._request(
                method="COPY",
                resource=docid,
                headers={"Destination": destination},
                query_kwargs={"rev": rev},
            )
        ).json()
        return data["id"], data["ok"], data["rev"]

    async def create(
        self,
        doc: dict | Document,
        *,
        batch: bool | None = None,
    ) -> tuple[str, bool, str]:
        """
        Create a new document without specifying an ID.

        Parameters
        ----------
        doc : dict | Document
            A dictionary or Document instance.
        batch : bool
            Stores document in batch mode. Default `None`.

        Returns
        -------
        tuple[str, bool, str] : (id, ok, rev)
        """
        data = (
            await self._post(body=doc, query_kwargs={"batch": "ok" if batch is True else None})
        ).json()
        return data["id"], data["ok"], data["rev"]

    async def delete(self, docid: str, rev: str, *, batch: bool | None = None) -> bool:
        """
        Delete a document.

        Parameters
        ----------
        docid : str
            The document's id.
        rev : str
            The document's current revision.
        batch : bool
            Stores document in batch mode. Default `None`.

        Returns
        -------
        bool : `True` upon successful deletion.
        """
        await self._delete(
            resource=docid,
            query_kwargs={"rev": rev, "batch": "ok" if batch is True else None},
        )
        return True

    async def delete_attachment(
        self, docid: str, attname: str, rev: str, *, batch: bool = False
    ) -> bool:
        """
        Delete an attachment.

        Parameters
        ----------
        docid : str
            The document's id.
        attname : str
            The attachment's name.
        rev : str
            The document's current revision.
        batch : bool
            Stores in batch mode. Default `False`.

        Returns
        -------
        bool : `True` upon successful deletion.
        """
        await self._delete(
            resource=f"{docid}/{attname}",
            query_kwargs={"rev": rev, "batch": "ok" if batch is True else None},
        )
        return True

    async def explain(
        self,
        selector: dict,
        limit: int = 25,
        skip: int = 0,
        sort: list[dict] | None = None,
        fields: list[str] | None = None,
        use_index: str | list[str] | None = None,
        conflicts: bool = False,
        r: int = 1,
        bookmark: str | None = None,
        update: bool = True,
        stable: bool | None = None,
        execution_stats: bool = False,
    ) -> dict:
        """
        Shows which index is being used by the query. Parameters are the same as
        `AsyncDatabase.find`.

        Parameters
        ----------
        selector : dict
            JSON object describing criteria used to select documents.
        limit : int
            Maximum number of results returned. Default is `25`.
        skip : int
            Skip the first `n` results. Default is `0`.
        sort : list[dict]
            JSON array following CouchDB's sort syntax. Default is `None`.
        fields : list[str]
            Fields to return. Default is `None` (entire object).
        use_index : str | list[str]
            Instruct a query to use a specific index. Default is `None`.
        conflicts : bool
            Include conflicted documents. Default is `False`.
        r : int
            Read quorum. Default is `1`.
        bookmark : str
            Paging bookmark. Default is `None`.
        update : bool
            Whether to update the index prior to returning the result. Default is `True`.
        stable : bool
            Whether to use a stable set of shards. Default is `None`.
        execution_stats : bool
            Include execution statistics. Default is `False`.

        Returns
        -------
        dict
        """
        return (
            await self._post(
                resource="_explain",
                body=rm_nones_from_dict(
                    {
                        "selector": selector,
                        "limit": limit,
                        "skip": skip,
                        "sort": sort,
                        "fields": fields,
                        "use_index": use_index,
                        "conflicts": conflicts,
                        "r": r,
                        "bookmark": bookmark,
                        "update": update,
                        "stable": stable,
                        "execution_stats": execution_stats,
                    }
                ),
            )
        ).json()

    async def find(
        self,
        selector: dict,
        limit: int = 25,
        skip: int = 0,
        sort: list[dict] | None = None,
        fields: list[str] | None = None,
        use_index: str | list[str] | None = None,
        conflicts: bool = False,
        r: int = 1,
        bookmark: str | None = None,
        update: bool = True,
        stable: bool | None = None,
        execution_stats: bool = False,
        partition: str | None = None,
    ) -> dict:
        """
        Find documents using a declarative JSON querying syntax.

        Parameters
        ----------
        selector : dict
            JSON object describing criteria used to select documents.
        limit : int
            Maximum number of results returned. Default is `25`.
        skip : int
            Skip the first `n` results. Default is `0`.
        sort : list[dict]
            JSON array following CouchDB's sort syntax. Default is `None`.
        fields : list[str]
            Fields to return. Default is `None` (entire object).
        use_index : str | list[str]
            Instruct a query to use a specific index. Default is `None`.
        conflicts : bool
            Include conflicted documents. Default is `False`.
        r : int
            Read quorum. Default is `1`.
        bookmark : str
            Paging bookmark. Default is `None`.
        update : bool
            Whether to update the index prior to returning the result. Default is `True`.
        stable : bool
            Whether to use a stable set of shards. Default is `None`.
        execution_stats : bool
            Include execution statistics. Default is `False`.
        partition : str
            An optional partition ID. Only valid for partitioned databases. Default `None`.

        Returns
        -------
        dict : Keys are `bookmark`, `docs`, `warning`.
        """
        return (
            await self._post(
                resource=partitioned_db_resource_parser(
                    resource="_find",
                    partition=partition,
                ),
                body=rm_nones_from_dict(
                    {
                        "selector": selector,
                        "limit": limit,
                        "skip": skip,
                        "sort": sort,
                        "fields": fields,
                        "use_index": use_index,
                        "conflicts": conflicts,
                        "r": r,
                        "bookmark": bookmark,
                        "update": update,
                        "stable": stable,
                        "execution_stats": execution_stats,
                    }
                ),
            )
        ).json()

    async def indexes(self) -> dict:
        """
        Get a list of all indexes in the database.

        Returns
        -------
        dict : Keys are `total_rows` and `indexes`.
        """
        return (await self._get(resource="_index")).json()

    async def get(
        self,
        docid: str,
        *,
        attachments: bool | None = None,
        att_encoding_info: bool | None = None,
        atts_since: Iterable[str] | None = None,
        conflicts: bool | None = None,
        deleted_conflicts: bool | None = None,
        latest: bool | None = None,
        local_seq: bool | None = None,
        meta: bool | None = None,
        open_revs: Iterable[str] | None = None,
        rev: str | None = None,
        revs: bool | None = None,
        revs_info: bool | None = None,
        check: bool | None = None,
        default_value: Any | None = None,
    ) -> Document | Any:
        """
        Get a document by id.

        Parameters
        ----------
        docid : str
            The document's id.
        attachments : bool
            Includes attachment bodies in response. Default `None`.
        att_encoding_info : bool
            Includes encoding information in attachment stubs. Default `None`.
        atts_since : Iterable[str]
            Includes attachments only since specified revisions. Default `None`.
        conflicts : bool
            Includes conflict information. Default `None`.
        deleted_conflicts : bool
            Includes deleted conflicted revisions. Default `None`.
        latest : bool
            Forces retrieving the latest leaf revision. Default `None`.
        local_seq : bool
            Includes last update sequence for the document. Default `None`.
        meta : bool
            Equivalent to specifying all conflicts, deleted_conflicts and revs_info.
            Default `None`.
        open_revs : Iterable[str]
            Retrieves documents of specified leaf revisions. Default `None`.
        rev : str
            Retrieves document of specified revision. Default `None`.
        revs : bool
            Includes list of all known document revisions. Default `None`.
        revs_info : bool
            Includes detailed information for all known document revisions. Default `None`.
        check : bool
            If `True`, raise an exception if the document is not found. Default `None`.
        default_value : Any
            Value to return if `check=False` and document is not found. Default `None`.

        Returns
        -------
        Document | Any
        """
        try:
            return Document(
                **(
                    await self._get(
                        resource=docid,
                        query_kwargs={
                            "attachments": attachments,
                            "att_encoding_info": att_encoding_info,
                            "atts_since": atts_since,
                            "conflicts": conflicts,
                            "deleted_conflicts": deleted_conflicts,
                            "latest": latest,
                            "local_seq": local_seq,
                            "meta": meta,
                            "open_revs": open_revs,
                            "rev": rev,
                            "revs": revs,
                            "revs_info": revs_info,
                        },
                    )
                ).json()
            )
        except (CouchDBError, httpx.RequestError):
            if check:
                raise
            return default_value

    async def get_attachment(
        self,
        docid: str,
        attname: str,
        rev: str | None = None,
    ) -> AttachmentDocument:
        """
        Get a document's attachment.

        Parameters
        ----------
        docid : str
            The document's id.
        attname : str
            The attachment's name.
        rev : str
            A specific revision. Default `None`.

        Returns
        -------
        AttachmentDocument
        """
        response = await self._get(f"{docid}/{attname}", query_kwargs={"rev": rev})
        content_md5 = response.headers.get("content-md5")
        digest_value = f"md5-{content_md5}" if content_md5 else None
        return AttachmentDocument(
            content=response.content,
            content_encoding=response.headers.get("content-encoding"),
            content_length=response.headers.get("content-length"),
            content_type=response.headers.get("content-type"),
            digest=digest_value,
        )

    async def get_design(self, ddoc: str, **kwargs) -> Document:
        """
        Get a design document.

        Parameters
        ----------
        ddoc : str
            The design document's name.
        kwargs
            Further `AsyncDatabase.get` parameters.

        Returns
        -------
        Document
        """
        return await self.get(docid=f"_design/{ddoc}", **kwargs)

    async def purge(self, data: dict) -> dict:
        """
        Permanently purge the given `(id, rev)` pairs.

        Parameters
        ----------
        data : dict
            A dictionary with document IDs as keys and list of revisions as values.

        Returns
        -------
        dict
        """
        return (await self._post(resource="_purge", body=data)).json()

    async def put_attachment(
        self,
        docid: str,
        attname: str,
        path: str | None = None,
        *,
        content: bytes | None = None,
        content_type: str | None = None,
        rev: str | None = None,
    ) -> tuple[str, bool, str]:
        """
        Upload content as an attachment to the specified document.

        Parameters
        ----------
        docid : str
            The document's id.
        attname : str
            The attachment's name.
        path : str
            Path to a local file to upload. Mutually exclusive with `content`.
            The file is streamed to CouchDB in chunks, so its full contents are never
            loaded into memory.
        content : bytes
            Raw bytes to upload. Mutually exclusive with `path`.
        content_type : str
            The attachment's MIME type. Required when `content` is provided.
        rev : str
            The document's current revision. Required for existing documents.

        Returns
        -------
        tuple[str, bool, str] : (id, ok, rev)
        """
        if (not content and not path) or (content and path):
            raise ValueError(
                'Precisely one of the arguments "content" and "path" must be provided.'
            )
        if content and not content_type:
            raise ValueError('Argument "content_type" cannot be empty when "content" is provided.')
        resource = f"{docid}/{attname}"
        query_kwargs = {"rev": rev}
        content_type = content_type if content_type else mimetypes.guess_type(path)[0]
        if path:
            with open(path, "rb") as file:
                content_length = os.fstat(file.fileno()).st_size
                response = await self._put(
                    resource=resource,
                    query_kwargs=query_kwargs,
                    content=_stream_file_content(file),
                    headers={
                        "content-type": content_type,
                        "content-length": str(content_length),
                    },
                )
        else:
            response = await self._put(
                resource=resource,
                query_kwargs=query_kwargs,
                content=content,
                headers={"content-type": content_type},
            )
        data = response.json()
        return data["id"], data["ok"], data["rev"]

    async def put_design(
        self,
        ddoc: str,
        *,
        rev: str | None = None,
        language: str | None = None,
        options: dict | None = None,
        filters: dict | None = None,
        updates: dict | None = None,
        validate_doc_update: str | None = None,
        views: dict | None = None,
        autoupdate: bool | None = None,
        partitioned: bool | None = None,
        **kwargs,
    ) -> tuple[str, bool, str]:
        """
        Create or update a named design document.

        Parameters
        ----------
        ddoc : str
            The design document's name.
        rev : str
            The design document's revision in case of an update.
        language : str
            Query Server to process design document functions.
        options : dict
            View's default options.
        filters : dict
            Filter functions definition.
        updates : dict
            Update functions definition.
        validate_doc_update : str
            Validate document update function source.
        views : dict
            View functions definition.
        autoupdate : bool
            Indicates whether to automatically build indexes.
        partitioned : bool
            Set to `True` for a partitioned design.
        kwargs
            Further `AsyncDatabase.save` parameters.

        Returns
        -------
        tuple[str, bool, str] : (id, ok, rev)
        """
        if partitioned:
            options = {**(options or {}), "partitioned": partitioned}
        return await self.save(
            doc=rm_nones_from_dict(
                {
                    "_id": f"_design/{ddoc}",
                    "_rev": rev,
                    "language": language,
                    "options": options,
                    "filters": filters,
                    "updates": updates,
                    "validate_doc_update": validate_doc_update,
                    "views": views,
                    "autoupdate": autoupdate,
                }
            ),
            **kwargs,
        )

    async def save(
        self,
        doc: dict | Document,
        batch: bool | None = None,
        new_edits: bool | None = None,
        path: str | None = None,
    ) -> tuple[str, bool, str]:
        """
        Create a new named document, or a new revision of an existing document.

        Parameters
        ----------
        doc : dict | Document
            A dictionary or Document instance with a valid `_id`, and `_rev` if updating.
        batch : bool
            Store document in batch mode. Default `None`.
        new_edits : bool
            Prevents insertion of a conflicting document. Default `None`.
        path : str
            Database path, e.g. `_design`. Default `None`.

        Returns
        -------
        tuple[str, bool, str] : (id, ok, rev)
        """
        batch = "ok" if batch else None
        data = (
            await self._put(
                resource=f"{path}/{doc.get('_id')}" if path else doc.get("_id"),
                body=doc,
                query_kwargs={
                    "batch": "ok" if batch else None,
                    "new_edits": new_edits,
                    "rev": doc.get("_rev"),
                },
            )
        ).json()
        return data["id"], data["ok"], data["rev"]

    async def save_index(
        self,
        index: dict,
        ddoc: str | None = None,
        name: str | None = None,
        index_type: str | None = "json",
        partitioned: bool | None = None,
    ) -> tuple[str, str, str]:
        """
        Create a new index on a database.

        Parameters
        ----------
        index : dict
            Dictionary describing the index to create.
        ddoc : str
            Name of the design document. Default `None` (auto-generated).
        name : str
            Name of the index. Default `None` (auto-generated).
        index_type : str
            `json` or `text`. Default `json`.
        partitioned : bool
            Whether the index is partitioned or global. Default `None`.

        Returns
        -------
        tuple[str, str, str] : (result, id, name)
        """
        data = (
            await self._post(
                resource="_index",
                body=rm_nones_from_dict(
                    {
                        "index": index,
                        "ddoc": ddoc,
                        "name": name,
                        "type": index_type,
                        "partitioned": partitioned,
                    }
                ),
            )
        ).json()
        return data["result"], data["id"], data["name"]

    async def delete_index(self, ddoc: str, name: str, index_type: str = "json") -> bool:
        """
        Delete an index from a database. For more info, please refer to
        [the official documentation](https://docs.couchdb.org/en/main/api/database/find.html#db-index).

        Parameters
        ----------
        ddoc : str
            Name of the design document the index belongs to. A `_design/` prefix is stripped
            automatically.
        name : str
            Name of the index.
        index_type : str
            Can be `json` or `text`. Defaults to `json`.

        Returns
        -------
        bool : `True` upon successful deletion.
        """
        ddoc = ddoc.removeprefix("_design/")
        return (await self._delete(resource=f"_index/{ddoc}/{index_type}/{name}")).json().get("ok")

    async def security(self) -> SecurityDocument:
        """
        Returns the current security object from the specified database.

        Returns
        -------
        SecurityDocument
        """
        data = (await self._get(resource="_security")).json()
        return SecurityDocument(**data)

    async def update_security(
        self,
        admins: dict | SecurityDocumentElement | None = None,
        members: dict | SecurityDocumentElement | None = None,
    ) -> bool:
        """
        Update database security.

        Parameters
        ----------
        admins : dict | SecurityDocumentElement
            Object with `names` and `roles` fields.
        members : dict | SecurityDocumentElement
            Object with `names` and `roles` fields.

        Returns
        -------
        bool : Operation status.
        """
        return (
            await self._put(resource="_security", body={"admins": admins, "members": members})
        ).json()["ok"]

    async def view(
        self,
        ddoc: str,
        view: str | None = None,
        *,
        partition: str | None = None,
        conflicts: bool | None = None,
        descending: bool | None = None,
        endkey: Any | None = None,
        endkey_docid: str | None = None,
        group: bool | None = None,
        group_level: int | None = None,
        include_docs: bool | None = None,
        attachments: bool | None = None,
        att_encoding_info: bool | None = None,
        inclusive_end: bool | None = None,
        key: str | None = None,
        keys: Iterable[str] | None = None,
        limit: int | None = None,
        reduce: bool | None = None,
        skip: int | None = None,
        sort: bool | None = None,
        stable: bool | None = None,
        startkey: Any | None = None,
        startkey_docid: str | None = None,
        update: str | None = None,
        update_seq: bool | None = None,
    ) -> ViewResult:
        """
        Executes the specified view function from the specified design document.

        Parameters
        ----------
        ddoc : str
            The corresponding design document's id.
        view : str
            The view's id.
        partition : str
            An optional partition ID. Only valid for partitioned databases. Default `None`.
        conflicts : bool
            Include conflicts information in response. Default `None`.
        descending : bool
            Return documents in descending order. Default `None`.
        endkey : Any
            Stop returning records at this key. Default `None`.
        endkey_docid : str
            Stop returning records at this document ID. Default `None`.
        group : bool
            Group results using the reduce function. Default `None`.
        group_level : int
            Specify the group level. Default `None`.
        include_docs : bool
            Include the associated document with each row. Default `None`.
        attachments : bool
            Include Base64-encoded attachment content. Default `None`.
        att_encoding_info : bool
            Include encoding information in attachment stubs. Default `None`.
        inclusive_end : bool
            Whether the end key should be included in the result. Default `None`.
        key : str
            Return only documents matching this key. Default `None`.
        keys : Iterable[str]
            Return only documents matching these keys. Default `None`.
        limit : int
            Maximum number of documents to return. Default `None`.
        reduce : bool
            Use the reduction function. Default `None`.
        skip : int
            Skip this number of records. Default `None`.
        sort : bool
            Sort returned rows. Default `None`.
        stable : bool
            Use a stable set of shards. Default `None`.
        startkey : Any
            Return records starting with this key. Default `None`.
        startkey_docid : str
            Return records starting with this document ID. Default `None`.
        update : str
            Whether to update the view prior to responding (`true`, `false`, `lazy`).
            Default `None`.
        update_seq : bool
            Include the `update_seq` value in the response. Default `None`.

        Returns
        -------
        ViewResult
        """
        path = partitioned_db_resource_parser(
            resource="_design",
            partition=partition,
        )
        return ViewResult(
            **(
                await self._get(
                    resource=f"{path}/{ddoc}/_view/{view}" if (ddoc and view) else ddoc,
                    query_kwargs={
                        "conflicts": conflicts,
                        "descending": descending,
                        "endkey": endkey,
                        "endkey_docid": endkey_docid,
                        "group": group,
                        "group_level": group_level,
                        "include_docs": include_docs,
                        "attachments": attachments,
                        "att_encoding_info": att_encoding_info,
                        "inclusive_end": inclusive_end,
                        "key": key,
                        "keys": keys,
                        "limit": limit,
                        "reduce": reduce,
                        "skip": skip,
                        "sorted": sort,
                        "stable": stable,
                        "startkey": startkey,
                        "startkey_docid": startkey_docid,
                        "update": update,
                        "update_seq": update_seq,
                    },
                )
            ).json()
        )

    async def changes(
        self,
        *,
        doc_ids: list[str] | None = None,
        conflicts: bool | None = None,
        descending: bool | None = None,
        feed: str | None = None,
        filter: str | None = None,
        heartbeat: int | None = None,
        include_docs: bool | None = None,
        attachments: bool | None = None,
        att_encoding_info: bool | None = None,
        limit: int | None = None,
        since: str | None = None,
        style: str | None = None,
        timeout: int | None = None,
        view: str | None = None,
        seq_interval: int | None = None,
        selector: dict | None = None,
    ) -> dict:
        """
        Returns a sorted list of changes made to documents in the database. Only the most
        recent change for a given document is included.

        When `doc_ids` is provided the request is sent as ``POST /{db}/_changes`` with
        ``filter=_doc_ids``. When `selector` is provided it is sent as
        ``POST /{db}/_changes`` with ``filter=_selector``. All other cases use
        ``GET /{db}/_changes``.

        .. note::
            ``feed='continuous'`` and ``feed='eventsource'`` are **not** supported by this
            method. Passing either value raises :class:`ValueError`. Streaming feeds will be
            addressed in a future ``changes_stream()`` method.

        Parameters
        ----------
        doc_ids : list[str]
            List of document IDs to filter the changes feed. Triggers a POST request with
            ``filter=_doc_ids``. Mutually exclusive with `selector`.
        conflicts : bool
            Include conflicts information. Only effective when `include_docs` is `True`.
        descending : bool
            Return changes in descending sequence order. Default `False`.
        feed : str
            Feed type. Supported values: ``'normal'`` (default), ``'longpoll'``.
            ``'continuous'`` and ``'eventsource'`` are not supported.
        filter : str
            Name of a filter function (``'design_doc/filter_name'``, ``'_design'``, or
            ``'_view'``). Do not pass ``'_doc_ids'`` or ``'_selector'`` manually — use the
            `doc_ids` / `selector` parameters instead.
        heartbeat : int
            Milliseconds between heartbeat newlines for ``longpoll`` feed.
        include_docs : bool
            Include the associated document with each result. Default `False`.
        attachments : bool
            Include Base64-encoded attachment content when `include_docs` is `True`.
        att_encoding_info : bool
            Include encoding info in attachment stubs when `include_docs` is `True`.
        limit : int
            Maximum number of rows to return.
        since : str
            Return only changes after the given update sequence. Use ``'now'`` to get only
            future changes.
        style : str
            Revision style. ``'main_only'`` (default) or ``'all_docs'``.
        timeout : int
            Maximum milliseconds to wait for a change (``longpoll`` only).
        view : str
            View function to use as a filter (requires ``filter='_view'``).
        seq_interval : int
            Calculate update sequence every N results (reduces server load on large
            sharded databases).
        selector : dict
            Mango selector to filter documents. Triggers a POST request with
            ``filter=_selector``. Mutually exclusive with `doc_ids`.

        Returns
        -------
        dict : A dictionary with the following keys.

          - ``last_seq`` (`str`) — last change update sequence
          - ``pending`` (`int`) — count of remaining items in the feed
          - ``results`` (`list`) — list of change objects, each with ``id``, ``seq``,
            ``changes``, and optionally ``deleted`` / ``doc``

        Raises
        ------
        ValueError
            If ``feed`` is ``'continuous'`` or ``'eventsource'``.
        CouchDBError
            If both `doc_ids` and `selector` are provided.
        """
        if feed in ("continuous", "eventsource"):
            raise ValueError(
                f"feed={feed!r} is not supported by changes(). Use feed='normal' or "
                "'longpoll'. Streaming feeds will be available via changes_stream() "
                "in a future release."
            )
        if doc_ids is not None and selector is not None:
            raise CouchDBError("Arguments 'doc_ids' and 'selector' are mutually exclusive.")
        query_kwargs = {
            "conflicts": conflicts,
            "descending": descending,
            "feed": feed,
            "filter": filter,
            "heartbeat": heartbeat,
            "include_docs": include_docs,
            "attachments": attachments,
            "att_encoding_info": att_encoding_info,
            "limit": limit,
            "since": since,
            "style": style,
            "timeout": timeout,
            "view": view,
            "seq_interval": seq_interval,
        }
        if doc_ids is not None:
            query_kwargs["filter"] = "_doc_ids"
            return (
                await self._post(
                    resource="_changes",
                    body={"doc_ids": doc_ids},
                    query_kwargs=query_kwargs,
                )
            ).json()
        if selector is not None:
            query_kwargs["filter"] = "_selector"
            return (
                await self._post(
                    resource="_changes",
                    body={"selector": selector},
                    query_kwargs=query_kwargs,
                )
            ).json()
        return (await self._get(resource="_changes", query_kwargs=query_kwargs)).json()

    @asynccontextmanager
    async def changes_stream(
        self,
        *,
        doc_ids: list[str] | None = None,
        conflicts: bool | None = None,
        descending: bool | None = None,
        filter: str | None = None,
        heartbeat: int | None = None,
        include_docs: bool | None = None,
        attachments: bool | None = None,
        att_encoding_info: bool | None = None,
        limit: int | None = None,
        since: str | None = None,
        style: str | None = None,
        timeout: int | None = None,
        view: str | None = None,
        seq_interval: int | None = None,
        selector: dict | None = None,
    ) -> AsyncIterator[dict]:
        """
        Stream changes made to documents in the database using the ``continuous`` changes
        feed. Unlike `AsyncDatabase.changes`, this does not buffer the whole response — it
        yields each change object as it arrives over a persistent HTTP connection.

        This method must be used as an async context manager. The underlying connection is
        kept open for the duration of the ``async with`` block and closed automatically on
        exit.

        When `doc_ids` is provided the request is sent as ``POST /{db}/_changes`` with
        ``filter=_doc_ids``. When `selector` is provided it is sent as
        ``POST /{db}/_changes`` with ``filter=_selector``. All other cases use
        ``GET /{db}/_changes``.

        Parameters
        ----------
        doc_ids : list[str]
            List of document IDs to filter the changes feed. Triggers a POST request with
            ``filter=_doc_ids``. Mutually exclusive with `selector`.
        conflicts : bool
            Include conflicts information. Only effective when `include_docs` is `True`.
        descending : bool
            Return changes in descending sequence order. Default `False`.
        filter : str
            Name of a filter function (``'design_doc/filter_name'``, ``'_design'``, or
            ``'_view'``). Do not pass ``'_doc_ids'`` or ``'_selector'`` manually — use the
            `doc_ids` / `selector` parameters instead.
        heartbeat : int
            Milliseconds between heartbeat newlines for ``continuous`` feed.
        include_docs : bool
            Include the associated document with each result. Default `False`.
        attachments : bool
            Include Base64-encoded attachment content when `include_docs` is `True`.
        att_encoding_info : bool
            Include encoding info in attachment stubs when `include_docs` is `True`.
        limit : int
            Maximum number of rows to return. On the ``continuous`` feed this caps the
            number of emitted rows but does not close the connection on CouchDB 3.3.x —
            the stream keeps blocking until the caller breaks out of the loop or the
            server times out.
        since : str
            Return only changes after the given update sequence. Use ``'now'`` to get only
            future changes.
        style : str
            Revision style. ``'main_only'`` (default) or ``'all_docs'``.
        timeout : int
            Maximum milliseconds to wait for a change, sent to CouchDB as the ``timeout``
            query parameter (not the HTTP connection timeout). Default `None` (no timeout).
        view : str
            View function to use as a filter (requires ``filter='_view'``).
        seq_interval : int
            Calculate update sequence every N results (reduces server load on large
            sharded databases).
        selector : dict
            Mango selector to filter documents. Triggers a POST request with
            ``filter=_selector``. Mutually exclusive with `doc_ids`.

        Yields
        ------
        AsyncIterator[dict] : An async iterator yielding change objects, each with ``id``,
            ``seq``, ``changes``, and optionally ``deleted`` / ``doc``.

        Raises
        ------
        CouchDBError
            If both `doc_ids` and `selector` are provided.

        Notes
        -----
        - The stream stays open indefinitely. Break out of the loop (or exit the
          ``async with`` block) to close the connection.
        - When ``limit`` is set, CouchDB's documentation mentions that the feed may end
          with a terminal ``{"last_seq": "..."}`` object that has no ``id`` key. This
          object is not emitted by CouchDB 3.3.x and its presence may vary by server
          version, so callers indexing ``row["id"]`` should guard against rows without an
          ``id`` key.

        Examples
        --------
        >>> async with db.changes_stream(since="now") as stream:
        ...     async for row in stream:
        ...         print(row["id"], row["seq"])
        """
        if doc_ids is not None and selector is not None:
            raise CouchDBError("Arguments 'doc_ids' and 'selector' are mutually exclusive.")
        query_kwargs = rm_nones_from_dict(
            {
                "feed": "continuous",
                "conflicts": conflicts,
                "descending": descending,
                "filter": filter,
                "heartbeat": heartbeat,
                "include_docs": include_docs,
                "attachments": attachments,
                "att_encoding_info": att_encoding_info,
                "limit": limit,
                "since": since,
                "style": style,
                "timeout": timeout,
                "view": view,
                "seq_interval": seq_interval,
            }
        )
        path = f"{self.root}/_changes"
        url = build_url(
            scheme=self.scheme,
            host=self.host,
            path=path,
            port=self.port,
            **query_kwargs,
        )
        req_kwargs = {}
        if self.auth_method == "basic":
            req_kwargs["auth"] = self._auth
        elif self.auth_method == "cookie":
            if self._is_auth_token_expired():
                await self._renew_auth_token()
        method = "GET"
        body = None
        if doc_ids is not None:
            method, body = "POST", {"doc_ids": doc_ids}
        elif selector is not None:
            method, body = "POST", {"selector": selector}

        async with self.session.stream(
            method, url, json=body, timeout=self.timeout, **req_kwargs
        ) as response:
            check_response(response=response)

            async def _iter() -> AsyncIterator[dict]:
                async for line in response.aiter_lines():
                    if line := line.strip():
                        yield json.loads(line)

            yield _iter()

    async def get_partition(self, partition_id: str) -> AsyncPartition:
        """
        Get a given partition.

        Parameters
        ----------
        partition_id : str
            The partition's ID.

        Returns
        -------
        AsyncPartition
        """
        return AsyncPartition(
            partition_id=partition_id,
            name=self.name,
            url=self.url,
            port=self.port,
            user=self._user,
            password=self._password,
            disable_ssl_verification=self.disable_ssl_verification,
            auth_method=self.auth_method,
            session=self.session,  # shared — child sets _owns_session=False
            _database=self,
        )

Async CouchDB database client. Mirrors Database with async def methods throughout.

Note: __getitem__ is not supported on async classes — Python does not allow __getitem__ to be a coroutine. Use await db.get(docid) instead.

Parameters

name : str

The name of the database.

url : str

The url of the CouchDB server formatted as scheme://user:password@host:port. For example:

"http://user:password@127.0.0.1:5984"
"https://couchdb.example.com"

port : int

The port of the CouchDB server. Can also be supplied via the url.

user : str

The CouchDB admin username. Can also be supplied via the url.

password : str

The CouchDB admin password. Can also be supplied via the url.

disable_ssl_verification : bool

Controls whether to verify the server's TLS certificate. Set to True when connecting to a server with self-signed TLS certificates. Default False.

auth_method : str

Authentication method. Choices are cookie or basic. Default is DEFAULT_AUTH_METHOD.

timeout : int

The default timeout for requests. Default c.f. DEFAULT_TIMEOUT.

session : httpx.AsyncClient

A specific async client to use. Optional — if not provided, a new client will be initialized.

_server : AsyncServer

The owning AsyncServer instance. Set internally by AsyncServer.get() to keep the server alive for the lifetime of this database object. Not part of the public constructor API — pass None (default) when constructing an AsyncDatabase directly.

Ancestors

• AsyncBase

Subclasses

• AsyncPartition

Instance variables

prop server

Expand source code

@property
def server(self):
    """
    The `AsyncServer` instance this database was obtained from, or `None` if the database
    was constructed directly (i.e. not via `AsyncServer.get()`).

    Read-only. Setting this attribute raises `AttributeError`.

    Returns
    -------
    AsyncServer | None
    """
    return self._server

The AsyncServer instance this database was obtained from, or None if the database was constructed directly (i.e. not via AsyncServer.get()).

Read-only. Setting this attribute raises AttributeError.

Returns

AsyncServer | None

 

Methods

async def all_docs(self, partition: str | None = None, keys: Iterable[str] | None = None, **kwargs) ‑> ViewResult

Expand source code

async def all_docs(
    self,
    partition: str | None = None,
    keys: Iterable[str] | None = None,
    **kwargs,
) -> ViewResult:
    """
    Executes the built-in _all_docs view, returning all documents in the database
    (or partition).

    Parameters
    ----------
    partition : str
        Filter using the partition's name (only valid for partitioned databases).
        Default is `None`.
    keys : Iterable[str]
        Return only documents where the key matches one of the keys specified.
        Default is `None`.
    kwargs
        Further `AsyncDatabase.view` parameters.

    Returns
    -------
    ViewResult
    """
    return await self.view(
        f"_partition/{partition}/_all_docs" if partition else "_all_docs",
        keys=keys,
        **kwargs,
    )

Executes the built-in _all_docs view, returning all documents in the database (or partition).

Parameters

partition : str

Filter using the partition's name (only valid for partitioned databases). Default is None.

keys : Iterable[str]

Return only documents where the key matches one of the keys specified. Default is None.

kwargs

Further AsyncDatabase.view() parameters.

Returns

ViewResult

 

async def bulk_docs(self, docs: list[dict | Document], new_edits: bool = True) ‑> list[dict]

Expand source code

async def bulk_docs(
    self,
    docs: list[dict | Document],
    new_edits: bool = True,
) -> list[dict]:
    """
    Create or update multiple documents in a single request.

    Parameters
    ----------
    docs : list[dict | Document]
        List of document objects.
    new_edits : bool
        If `False`, prevents the database from assigning new revision IDs. Default `True`.

    Returns
    -------
    list[dict] : Each item contains `id`, `ok`, and `rev`.
    """
    return (
        await self._post(resource="_bulk_docs", body={"docs": docs, "new_edits": new_edits})
    ).json()

Create or update multiple documents in a single request.

Parameters

docs : list[dict | Document]

List of document objects.

new_edits : bool

If False, prevents the database from assigning new revision IDs. Default True.

Returns

list[dict] : Each item contains id, ok, and rev.

async def bulk_get(self, docs: list[dict | Document], revs: bool = False) ‑> list[dict]

Expand source code

async def bulk_get(
    self,
    docs: list[dict | Document],
    revs: bool = False,
) -> list[dict]:
    """
    Query several documents in bulk.

    Parameters
    ----------
    docs : list[dict | Document]
        List of document objects, with `id`, and optionally `rev` and `atts_since`.
    revs : bool
        Give the revisions history. Default `False`.

    Returns
    -------
    list[dict]
    """
    return (
        (
            await self._post(
                resource="_bulk_get",
                body={"docs": [extract_document_id_and_rev(_) for _ in docs]},
                query_kwargs={"revs": revs},
            )
        )
        .json()
        .get("results", [])
    )

Query several documents in bulk.

Parameters

docs : list[dict | Document]

List of document objects, with id, and optionally rev and atts_since.

revs : bool

Give the revisions history. Default False.

Returns

list[dict]

 

async def changes(self,
*,
doc_ids: list[str] | None = None,
conflicts: bool | None = None,
descending: bool | None = None,
feed: str | None = None,
filter: str | None = None,
heartbeat: int | None = None,
include_docs: bool | None = None,
attachments: bool | None = None,
att_encoding_info: bool | None = None,
limit: int | None = None,
since: str | None = None,
style: str | None = None,
timeout: int | None = None,
view: str | None = None,
seq_interval: int | None = None,
selector: dict | None = None) ‑> dict

Expand source code

async def changes(
    self,
    *,
    doc_ids: list[str] | None = None,
    conflicts: bool | None = None,
    descending: bool | None = None,
    feed: str | None = None,
    filter: str | None = None,
    heartbeat: int | None = None,
    include_docs: bool | None = None,
    attachments: bool | None = None,
    att_encoding_info: bool | None = None,
    limit: int | None = None,
    since: str | None = None,
    style: str | None = None,
    timeout: int | None = None,
    view: str | None = None,
    seq_interval: int | None = None,
    selector: dict | None = None,
) -> dict:
    """
    Returns a sorted list of changes made to documents in the database. Only the most
    recent change for a given document is included.

    When `doc_ids` is provided the request is sent as ``POST /{db}/_changes`` with
    ``filter=_doc_ids``. When `selector` is provided it is sent as
    ``POST /{db}/_changes`` with ``filter=_selector``. All other cases use
    ``GET /{db}/_changes``.

    .. note::
        ``feed='continuous'`` and ``feed='eventsource'`` are **not** supported by this
        method. Passing either value raises :class:`ValueError`. Streaming feeds will be
        addressed in a future ``changes_stream()`` method.

    Parameters
    ----------
    doc_ids : list[str]
        List of document IDs to filter the changes feed. Triggers a POST request with
        ``filter=_doc_ids``. Mutually exclusive with `selector`.
    conflicts : bool
        Include conflicts information. Only effective when `include_docs` is `True`.
    descending : bool
        Return changes in descending sequence order. Default `False`.
    feed : str
        Feed type. Supported values: ``'normal'`` (default), ``'longpoll'``.
        ``'continuous'`` and ``'eventsource'`` are not supported.
    filter : str
        Name of a filter function (``'design_doc/filter_name'``, ``'_design'``, or
        ``'_view'``). Do not pass ``'_doc_ids'`` or ``'_selector'`` manually — use the
        `doc_ids` / `selector` parameters instead.
    heartbeat : int
        Milliseconds between heartbeat newlines for ``longpoll`` feed.
    include_docs : bool
        Include the associated document with each result. Default `False`.
    attachments : bool
        Include Base64-encoded attachment content when `include_docs` is `True`.
    att_encoding_info : bool
        Include encoding info in attachment stubs when `include_docs` is `True`.
    limit : int
        Maximum number of rows to return.
    since : str
        Return only changes after the given update sequence. Use ``'now'`` to get only
        future changes.
    style : str
        Revision style. ``'main_only'`` (default) or ``'all_docs'``.
    timeout : int
        Maximum milliseconds to wait for a change (``longpoll`` only).
    view : str
        View function to use as a filter (requires ``filter='_view'``).
    seq_interval : int
        Calculate update sequence every N results (reduces server load on large
        sharded databases).
    selector : dict
        Mango selector to filter documents. Triggers a POST request with
        ``filter=_selector``. Mutually exclusive with `doc_ids`.

    Returns
    -------
    dict : A dictionary with the following keys.

      - ``last_seq`` (`str`) — last change update sequence
      - ``pending`` (`int`) — count of remaining items in the feed
      - ``results`` (`list`) — list of change objects, each with ``id``, ``seq``,
        ``changes``, and optionally ``deleted`` / ``doc``

    Raises
    ------
    ValueError
        If ``feed`` is ``'continuous'`` or ``'eventsource'``.
    CouchDBError
        If both `doc_ids` and `selector` are provided.
    """
    if feed in ("continuous", "eventsource"):
        raise ValueError(
            f"feed={feed!r} is not supported by changes(). Use feed='normal' or "
            "'longpoll'. Streaming feeds will be available via changes_stream() "
            "in a future release."
        )
    if doc_ids is not None and selector is not None:
        raise CouchDBError("Arguments 'doc_ids' and 'selector' are mutually exclusive.")
    query_kwargs = {
        "conflicts": conflicts,
        "descending": descending,
        "feed": feed,
        "filter": filter,
        "heartbeat": heartbeat,
        "include_docs": include_docs,
        "attachments": attachments,
        "att_encoding_info": att_encoding_info,
        "limit": limit,
        "since": since,
        "style": style,
        "timeout": timeout,
        "view": view,
        "seq_interval": seq_interval,
    }
    if doc_ids is not None:
        query_kwargs["filter"] = "_doc_ids"
        return (
            await self._post(
                resource="_changes",
                body={"doc_ids": doc_ids},
                query_kwargs=query_kwargs,
            )
        ).json()
    if selector is not None:
        query_kwargs["filter"] = "_selector"
        return (
            await self._post(
                resource="_changes",
                body={"selector": selector},
                query_kwargs=query_kwargs,
            )
        ).json()
    return (await self._get(resource="_changes", query_kwargs=query_kwargs)).json()

Returns a sorted list of changes made to documents in the database. Only the most recent change for a given document is included.

When doc_ids is provided the request is sent as POST /{db}/_changes with filter=_doc_ids. When selector is provided it is sent as POST /{db}/_changes with filter=_selector. All other cases use GET /{db}/_changes.

Note

feed='continuous' and feed='eventsource' are not supported by this method. Passing either value raises :class:ValueError. Streaming feeds will be addressed in a future changes_stream() method.

Parameters

doc_ids : list[str]

List of document IDs to filter the changes feed. Triggers a POST request with filter=_doc_ids. Mutually exclusive with selector.

conflicts : bool

Include conflicts information. Only effective when include_docs is True.

descending : bool

Return changes in descending sequence order. Default False.

feed : str

Feed type. Supported values: 'normal' (default), 'longpoll'. 'continuous' and 'eventsource' are not supported.

filter : str

Name of a filter function ('design_doc/filter_name', '_design', or '_view'). Do not pass '_doc_ids' or '_selector' manually — use the doc_ids / selector parameters instead.

heartbeat : int

Milliseconds between heartbeat newlines for longpoll feed.

include_docs : bool

Include the associated document with each result. Default False.

attachments : bool

Include Base64-encoded attachment content when include_docs is True.

att_encoding_info : bool

Include encoding info in attachment stubs when include_docs is True.

limit : int

Maximum number of rows to return.

since : str

Return only changes after the given update sequence. Use 'now' to get only future changes.

style : str

Revision style. 'main_only' (default) or 'all_docs'.

timeout : int

Maximum milliseconds to wait for a change (longpoll only).

view : str

View function to use as a filter (requires filter='_view').

seq_interval : int

Calculate update sequence every N results (reduces server load on large sharded databases).

selector : dict

Mango selector to filter documents. Triggers a POST request with filter=_selector. Mutually exclusive with doc_ids.

Returns

dict : A dictionary with the following keys.

• last_seq (str) — last change update sequence
• pending (int) — count of remaining items in the feed
• results (list) — list of change objects, each with id, seq, changes, and optionally deleted / doc

Raises

ValueError

If feed is 'continuous' or 'eventsource'.

CouchDBError

If both doc_ids and selector are provided.

async def changes_stream(self,
*,
doc_ids: list[str] | None = None,
conflicts: bool | None = None,
descending: bool | None = None,
filter: str | None = None,
heartbeat: int | None = None,
include_docs: bool | None = None,
attachments: bool | None = None,
att_encoding_info: bool | None = None,
limit: int | None = None,
since: str | None = None,
style: str | None = None,
timeout: int | None = None,
view: str | None = None,
seq_interval: int | None = None,
selector: dict | None = None) ‑> AsyncIterator[dict]

Expand source code

@asynccontextmanager
async def changes_stream(
    self,
    *,
    doc_ids: list[str] | None = None,
    conflicts: bool | None = None,
    descending: bool | None = None,
    filter: str | None = None,
    heartbeat: int | None = None,
    include_docs: bool | None = None,
    attachments: bool | None = None,
    att_encoding_info: bool | None = None,
    limit: int | None = None,
    since: str | None = None,
    style: str | None = None,
    timeout: int | None = None,
    view: str | None = None,
    seq_interval: int | None = None,
    selector: dict | None = None,
) -> AsyncIterator[dict]:
    """
    Stream changes made to documents in the database using the ``continuous`` changes
    feed. Unlike `AsyncDatabase.changes`, this does not buffer the whole response — it
    yields each change object as it arrives over a persistent HTTP connection.

    This method must be used as an async context manager. The underlying connection is
    kept open for the duration of the ``async with`` block and closed automatically on
    exit.

    When `doc_ids` is provided the request is sent as ``POST /{db}/_changes`` with
    ``filter=_doc_ids``. When `selector` is provided it is sent as
    ``POST /{db}/_changes`` with ``filter=_selector``. All other cases use
    ``GET /{db}/_changes``.

    Parameters
    ----------
    doc_ids : list[str]
        List of document IDs to filter the changes feed. Triggers a POST request with
        ``filter=_doc_ids``. Mutually exclusive with `selector`.
    conflicts : bool
        Include conflicts information. Only effective when `include_docs` is `True`.
    descending : bool
        Return changes in descending sequence order. Default `False`.
    filter : str
        Name of a filter function (``'design_doc/filter_name'``, ``'_design'``, or
        ``'_view'``). Do not pass ``'_doc_ids'`` or ``'_selector'`` manually — use the
        `doc_ids` / `selector` parameters instead.
    heartbeat : int
        Milliseconds between heartbeat newlines for ``continuous`` feed.
    include_docs : bool
        Include the associated document with each result. Default `False`.
    attachments : bool
        Include Base64-encoded attachment content when `include_docs` is `True`.
    att_encoding_info : bool
        Include encoding info in attachment stubs when `include_docs` is `True`.
    limit : int
        Maximum number of rows to return. On the ``continuous`` feed this caps the
        number of emitted rows but does not close the connection on CouchDB 3.3.x —
        the stream keeps blocking until the caller breaks out of the loop or the
        server times out.
    since : str
        Return only changes after the given update sequence. Use ``'now'`` to get only
        future changes.
    style : str
        Revision style. ``'main_only'`` (default) or ``'all_docs'``.
    timeout : int
        Maximum milliseconds to wait for a change, sent to CouchDB as the ``timeout``
        query parameter (not the HTTP connection timeout). Default `None` (no timeout).
    view : str
        View function to use as a filter (requires ``filter='_view'``).
    seq_interval : int
        Calculate update sequence every N results (reduces server load on large
        sharded databases).
    selector : dict
        Mango selector to filter documents. Triggers a POST request with
        ``filter=_selector``. Mutually exclusive with `doc_ids`.

    Yields
    ------
    AsyncIterator[dict] : An async iterator yielding change objects, each with ``id``,
        ``seq``, ``changes``, and optionally ``deleted`` / ``doc``.

    Raises
    ------
    CouchDBError
        If both `doc_ids` and `selector` are provided.

    Notes
    -----
    - The stream stays open indefinitely. Break out of the loop (or exit the
      ``async with`` block) to close the connection.
    - When ``limit`` is set, CouchDB's documentation mentions that the feed may end
      with a terminal ``{"last_seq": "..."}`` object that has no ``id`` key. This
      object is not emitted by CouchDB 3.3.x and its presence may vary by server
      version, so callers indexing ``row["id"]`` should guard against rows without an
      ``id`` key.

    Examples
    --------
    >>> async with db.changes_stream(since="now") as stream:
    ...     async for row in stream:
    ...         print(row["id"], row["seq"])
    """
    if doc_ids is not None and selector is not None:
        raise CouchDBError("Arguments 'doc_ids' and 'selector' are mutually exclusive.")
    query_kwargs = rm_nones_from_dict(
        {
            "feed": "continuous",
            "conflicts": conflicts,
            "descending": descending,
            "filter": filter,
            "heartbeat": heartbeat,
            "include_docs": include_docs,
            "attachments": attachments,
            "att_encoding_info": att_encoding_info,
            "limit": limit,
            "since": since,
            "style": style,
            "timeout": timeout,
            "view": view,
            "seq_interval": seq_interval,
        }
    )
    path = f"{self.root}/_changes"
    url = build_url(
        scheme=self.scheme,
        host=self.host,
        path=path,
        port=self.port,
        **query_kwargs,
    )
    req_kwargs = {}
    if self.auth_method == "basic":
        req_kwargs["auth"] = self._auth
    elif self.auth_method == "cookie":
        if self._is_auth_token_expired():
            await self._renew_auth_token()
    method = "GET"
    body = None
    if doc_ids is not None:
        method, body = "POST", {"doc_ids": doc_ids}
    elif selector is not None:
        method, body = "POST", {"selector": selector}

    async with self.session.stream(
        method, url, json=body, timeout=self.timeout, **req_kwargs
    ) as response:
        check_response(response=response)

        async def _iter() -> AsyncIterator[dict]:
            async for line in response.aiter_lines():
                if line := line.strip():
                    yield json.loads(line)

        yield _iter()

Stream changes made to documents in the database using the continuous changes feed. Unlike AsyncDatabase.changes(), this does not buffer the whole response — it yields each change object as it arrives over a persistent HTTP connection.

This method must be used as an async context manager. The underlying connection is kept open for the duration of the async with block and closed automatically on exit.

When doc_ids is provided the request is sent as POST /{db}/_changes with filter=_doc_ids. When selector is provided it is sent as POST /{db}/_changes with filter=_selector. All other cases use GET /{db}/_changes.

Parameters

doc_ids : list[str]

List of document IDs to filter the changes feed. Triggers a POST request with filter=_doc_ids. Mutually exclusive with selector.

conflicts : bool

Include conflicts information. Only effective when include_docs is True.

descending : bool

Return changes in descending sequence order. Default False.

filter : str

Name of a filter function ('design_doc/filter_name', '_design', or '_view'). Do not pass '_doc_ids' or '_selector' manually — use the doc_ids / selector parameters instead.

heartbeat : int

Milliseconds between heartbeat newlines for continuous feed.

include_docs : bool

Include the associated document with each result. Default False.

attachments : bool

Include Base64-encoded attachment content when include_docs is True.

att_encoding_info : bool

Include encoding info in attachment stubs when include_docs is True.

limit : int

Maximum number of rows to return. On the continuous feed this caps the number of emitted rows but does not close the connection on CouchDB 3.3.x — the stream keeps blocking until the caller breaks out of the loop or the server times out.

since : str

Return only changes after the given update sequence. Use 'now' to get only future changes.

style : str

Revision style. 'main_only' (default) or 'all_docs'.

timeout : int

Maximum milliseconds to wait for a change, sent to CouchDB as the timeout query parameter (not the HTTP connection timeout). Default None (no timeout).

view : str

View function to use as a filter (requires filter='_view').

seq_interval : int

Calculate update sequence every N results (reduces server load on large sharded databases).

selector : dict

Mango selector to filter documents. Triggers a POST request with filter=_selector. Mutually exclusive with doc_ids.

Yields

AsyncIterator[dict] : An async iterator yielding change objects, each with id, seq, changes, and optionally deleted / doc.

Raises

CouchDBError

If both doc_ids and selector are provided.

Notes

• The stream stays open indefinitely. Break out of the loop (or exit the async with block) to close the connection.
• When limit is set, CouchDB's documentation mentions that the feed may end with a terminal {"last_seq": "..."} object that has no id key. This object is not emitted by CouchDB 3.3.x and its presence may vary by server version, so callers indexing row["id"] should guard against rows without an id key.

Examples

>>> async with db.changes_stream(since="now") as stream:
...     async for row in stream:
...         print(row["id"], row["seq"])

async def compact(self, ddoc: str | None = None) ‑> bool

Expand source code

async def compact(self, ddoc: str | None = None) -> bool:
    """
    Request compaction of the database.

    Parameters
    ----------
    ddoc : str
        A design document name. If provided, compacts the view indexes for that ddoc.

    Returns
    -------
    bool: `True` upon compaction request successfully sent.
    """
    resource = "_compact"
    if ddoc:
        resource += f"/{ddoc}"
    return (await self._post(resource=resource)).json().get("ok")

Request compaction of the database.

Parameters

ddoc : str

A design document name. If provided, compacts the view indexes for that ddoc.

Returns

bool: True upon compaction request successfully sent.

async def copy(self, docid: str, destid: str, rev: str | None = None, destrev: str | None = None) ‑> tuple[str, bool, str]

Expand source code

async def copy(
    self,
    docid: str,
    destid: str,
    rev: str | None = None,
    destrev: str | None = None,
) -> tuple[str, bool, str]:
    """
    Copy an existing document to a new or existing document.

    Parameters
    ----------
    docid : str
        The ID of the document to copy.
    destid : str
        The target document's ID.
    rev : str
        A specific revision of the document to copy.
    destrev : str
        If the target document already exists, its current revision.

    Returns
    -------
    tuple[str, bool, str] : (id, ok, rev)
    """
    destination = destid
    if destrev:
        destination += f"?rev={destrev}"
    data = (
        await self._request(
            method="COPY",
            resource=docid,
            headers={"Destination": destination},
            query_kwargs={"rev": rev},
        )
    ).json()
    return data["id"], data["ok"], data["rev"]

Copy an existing document to a new or existing document.

Parameters

docid : str

The ID of the document to copy.

destid : str

The target document's ID.

rev : str

A specific revision of the document to copy.

destrev : str

If the target document already exists, its current revision.

Returns

tuple[str, bool, str] : (id, ok, rev)

 

async def create(self, doc: dict | Document, *, batch: bool | None = None) ‑> tuple[str, bool, str]

Expand source code

async def create(
    self,
    doc: dict | Document,
    *,
    batch: bool | None = None,
) -> tuple[str, bool, str]:
    """
    Create a new document without specifying an ID.

    Parameters
    ----------
    doc : dict | Document
        A dictionary or Document instance.
    batch : bool
        Stores document in batch mode. Default `None`.

    Returns
    -------
    tuple[str, bool, str] : (id, ok, rev)
    """
    data = (
        await self._post(body=doc, query_kwargs={"batch": "ok" if batch is True else None})
    ).json()
    return data["id"], data["ok"], data["rev"]

Create a new document without specifying an ID.

Parameters

doc : dict | Document

A dictionary or Document instance.

batch : bool

Stores document in batch mode. Default None.

Returns

tuple[str, bool, str] : (id, ok, rev)

 

async def delete(self, docid: str, rev: str, *, batch: bool | None = None) ‑> bool

Expand source code

async def delete(self, docid: str, rev: str, *, batch: bool | None = None) -> bool:
    """
    Delete a document.

    Parameters
    ----------
    docid : str
        The document's id.
    rev : str
        The document's current revision.
    batch : bool
        Stores document in batch mode. Default `None`.

    Returns
    -------
    bool : `True` upon successful deletion.
    """
    await self._delete(
        resource=docid,
        query_kwargs={"rev": rev, "batch": "ok" if batch is True else None},
    )
    return True

Delete a document.

Parameters

docid : str

The document's id.

rev : str

The document's current revision.

batch : bool

Stores document in batch mode. Default None.

Returns

bool : True upon successful deletion.

async def delete_attachment(self, docid: str, attname: str, rev: str, *, batch: bool = False) ‑> bool

Expand source code

async def delete_attachment(
    self, docid: str, attname: str, rev: str, *, batch: bool = False
) -> bool:
    """
    Delete an attachment.

    Parameters
    ----------
    docid : str
        The document's id.
    attname : str
        The attachment's name.
    rev : str
        The document's current revision.
    batch : bool
        Stores in batch mode. Default `False`.

    Returns
    -------
    bool : `True` upon successful deletion.
    """
    await self._delete(
        resource=f"{docid}/{attname}",
        query_kwargs={"rev": rev, "batch": "ok" if batch is True else None},
    )
    return True

Delete an attachment.

Parameters

docid : str

The document's id.

attname : str

The attachment's name.

rev : str

The document's current revision.

batch : bool

Stores in batch mode. Default False.

Returns

bool : True upon successful deletion.

async def delete_index(self, ddoc: str, name: str, index_type: str = 'json') ‑> bool

Expand source code

async def delete_index(self, ddoc: str, name: str, index_type: str = "json") -> bool:
    """
    Delete an index from a database. For more info, please refer to
    [the official documentation](https://docs.couchdb.org/en/main/api/database/find.html#db-index).

    Parameters
    ----------
    ddoc : str
        Name of the design document the index belongs to. A `_design/` prefix is stripped
        automatically.
    name : str
        Name of the index.
    index_type : str
        Can be `json` or `text`. Defaults to `json`.

    Returns
    -------
    bool : `True` upon successful deletion.
    """
    ddoc = ddoc.removeprefix("_design/")
    return (await self._delete(resource=f"_index/{ddoc}/{index_type}/{name}")).json().get("ok")

Delete an index from a database. For more info, please refer to the official documentation.

Parameters

ddoc : str

Name of the design document the index belongs to. A _design/ prefix is stripped automatically.

name : str

Name of the index.

index_type : str

Can be json or text. Defaults to json.

Returns

bool : True upon successful deletion.

async def design_docs(self,
*,
conflicts: bool | None = None,
descending: bool | None = None,
endkey: str | None = None,
include_docs: bool | None = None,
keys: Iterable[str] | None = None,
limit: int | None = None,
skip: int | None = None,
startkey: str | None = None,
update_seq: bool | None = None) ‑> ViewResult

Expand source code

async def design_docs(
    self,
    *,
    conflicts: bool | None = None,
    descending: bool | None = None,
    endkey: str | None = None,
    include_docs: bool | None = None,
    keys: Iterable[str] | None = None,
    limit: int | None = None,
    skip: int | None = None,
    startkey: str | None = None,
    update_seq: bool | None = None,
) -> ViewResult:
    """
    Executes the built-in `_design_docs` view, returning all the design documents in the database.

    This is a shorthand for `_all_docs` filtered to the `_design/` key range.

    Parameters
    ----------
    conflicts : bool
        Include conflicts information. Ignored if `include_docs` isn't `True`. Default is `None`.
    descending : bool
        Return the documents in descending order by key. Default is `None`.
    endkey : str
        Stop returning records when the specified key is reached. Default is `None`.
    include_docs : bool
        Include the associated document with each row. Default is `None`.
    keys : Iterable[str]
        Return only documents where the key matches one of the keys specified in the argument.
        Default is `None`.
    limit : int
        Limit the number of the returned documents. Default is `None`.
    skip : int
        Skip this number of records before starting to return the results. Default is `None`.
    startkey : str
        Return records starting with the specified key. Default is `None`.
    update_seq : bool
        Whether to include an `update_seq` value indicating the sequence id of the database.
        Default is `None`.

    Returns
    -------
    ViewResult
    """
    return ViewResult(
        **(
            await self._get(
                resource="_design_docs",
                query_kwargs=rm_nones_from_dict(
                    {
                        "conflicts": conflicts,
                        "descending": descending,
                        "endkey": endkey,
                        "include_docs": include_docs,
                        "keys": keys,
                        "limit": limit,
                        "skip": skip,
                        "startkey": startkey,
                        "update_seq": update_seq,
                    }
                ),
            )
        ).json()
    )

Executes the built-in _design_docs view, returning all the design documents in the database.

This is a shorthand for _all_docs filtered to the _design/ key range.

Parameters

conflicts : bool

Include conflicts information. Ignored if include_docs isn't True. Default is None.

descending : bool

Return the documents in descending order by key. Default is None.

endkey : str

Stop returning records when the specified key is reached. Default is None.

include_docs : bool

Include the associated document with each row. Default is None.

keys : Iterable[str]

Return only documents where the key matches one of the keys specified in the argument. Default is None.

limit : int

Limit the number of the returned documents. Default is None.

skip : int

Skip this number of records before starting to return the results. Default is None.

startkey : str

Return records starting with the specified key. Default is None.

update_seq : bool

Whether to include an update_seq value indicating the sequence id of the database. Default is None.

Returns

ViewResult

 

async def explain(self,
selector: dict,
limit: int = 25,
skip: int = 0,
sort: list[dict] | None = None,
fields: list[str] | None = None,
use_index: str | list[str] | None = None,
conflicts: bool = False,
r: int = 1,
bookmark: str | None = None,
update: bool = True,
stable: bool | None = None,
execution_stats: bool = False) ‑> dict

Expand source code

async def explain(
    self,
    selector: dict,
    limit: int = 25,
    skip: int = 0,
    sort: list[dict] | None = None,
    fields: list[str] | None = None,
    use_index: str | list[str] | None = None,
    conflicts: bool = False,
    r: int = 1,
    bookmark: str | None = None,
    update: bool = True,
    stable: bool | None = None,
    execution_stats: bool = False,
) -> dict:
    """
    Shows which index is being used by the query. Parameters are the same as
    `AsyncDatabase.find`.

    Parameters
    ----------
    selector : dict
        JSON object describing criteria used to select documents.
    limit : int
        Maximum number of results returned. Default is `25`.
    skip : int
        Skip the first `n` results. Default is `0`.
    sort : list[dict]
        JSON array following CouchDB's sort syntax. Default is `None`.
    fields : list[str]
        Fields to return. Default is `None` (entire object).
    use_index : str | list[str]
        Instruct a query to use a specific index. Default is `None`.
    conflicts : bool
        Include conflicted documents. Default is `False`.
    r : int
        Read quorum. Default is `1`.
    bookmark : str
        Paging bookmark. Default is `None`.
    update : bool
        Whether to update the index prior to returning the result. Default is `True`.
    stable : bool
        Whether to use a stable set of shards. Default is `None`.
    execution_stats : bool
        Include execution statistics. Default is `False`.

    Returns
    -------
    dict
    """
    return (
        await self._post(
            resource="_explain",
            body=rm_nones_from_dict(
                {
                    "selector": selector,
                    "limit": limit,
                    "skip": skip,
                    "sort": sort,
                    "fields": fields,
                    "use_index": use_index,
                    "conflicts": conflicts,
                    "r": r,
                    "bookmark": bookmark,
                    "update": update,
                    "stable": stable,
                    "execution_stats": execution_stats,
                }
            ),
        )
    ).json()

Shows which index is being used by the query. Parameters are the same as AsyncDatabase.find().

Parameters

selector : dict

JSON object describing criteria used to select documents.

limit : int

Maximum number of results returned. Default is 25.

skip : int

Skip the first n results. Default is 0.

sort : list[dict]

JSON array following CouchDB's sort syntax. Default is None.

fields : list[str]

Fields to return. Default is None (entire object).

use_index : str | list[str]

Instruct a query to use a specific index. Default is None.

conflicts : bool

Include conflicted documents. Default is False.

r : int

Read quorum. Default is 1.

bookmark : str

Paging bookmark. Default is None.

update : bool

Whether to update the index prior to returning the result. Default is True.

stable : bool

Whether to use a stable set of shards. Default is None.

execution_stats : bool

Include execution statistics. Default is False.

Returns

dict

 

async def find(self,
selector: dict,
limit: int = 25,
skip: int = 0,
sort: list[dict] | None = None,
fields: list[str] | None = None,
use_index: str | list[str] | None = None,
conflicts: bool = False,
r: int = 1,
bookmark: str | None = None,
update: bool = True,
stable: bool | None = None,
execution_stats: bool = False,
partition: str | None = None) ‑> dict

Expand source code

async def find(
    self,
    selector: dict,
    limit: int = 25,
    skip: int = 0,
    sort: list[dict] | None = None,
    fields: list[str] | None = None,
    use_index: str | list[str] | None = None,
    conflicts: bool = False,
    r: int = 1,
    bookmark: str | None = None,
    update: bool = True,
    stable: bool | None = None,
    execution_stats: bool = False,
    partition: str | None = None,
) -> dict:
    """
    Find documents using a declarative JSON querying syntax.

    Parameters
    ----------
    selector : dict
        JSON object describing criteria used to select documents.
    limit : int
        Maximum number of results returned. Default is `25`.
    skip : int
        Skip the first `n` results. Default is `0`.
    sort : list[dict]
        JSON array following CouchDB's sort syntax. Default is `None`.
    fields : list[str]
        Fields to return. Default is `None` (entire object).
    use_index : str | list[str]
        Instruct a query to use a specific index. Default is `None`.
    conflicts : bool
        Include conflicted documents. Default is `False`.
    r : int
        Read quorum. Default is `1`.
    bookmark : str
        Paging bookmark. Default is `None`.
    update : bool
        Whether to update the index prior to returning the result. Default is `True`.
    stable : bool
        Whether to use a stable set of shards. Default is `None`.
    execution_stats : bool
        Include execution statistics. Default is `False`.
    partition : str
        An optional partition ID. Only valid for partitioned databases. Default `None`.

    Returns
    -------
    dict : Keys are `bookmark`, `docs`, `warning`.
    """
    return (
        await self._post(
            resource=partitioned_db_resource_parser(
                resource="_find",
                partition=partition,
            ),
            body=rm_nones_from_dict(
                {
                    "selector": selector,
                    "limit": limit,
                    "skip": skip,
                    "sort": sort,
                    "fields": fields,
                    "use_index": use_index,
                    "conflicts": conflicts,
                    "r": r,
                    "bookmark": bookmark,
                    "update": update,
                    "stable": stable,
                    "execution_stats": execution_stats,
                }
            ),
        )
    ).json()

Find documents using a declarative JSON querying syntax.

Parameters

selector : dict

JSON object describing criteria used to select documents.

limit : int

Maximum number of results returned. Default is 25.

skip : int

Skip the first n results. Default is 0.

sort : list[dict]

JSON array following CouchDB's sort syntax. Default is None.

fields : list[str]

Fields to return. Default is None (entire object).

use_index : str | list[str]

Instruct a query to use a specific index. Default is None.

conflicts : bool

Include conflicted documents. Default is False.

r : int

Read quorum. Default is 1.

bookmark : str

Paging bookmark. Default is None.

update : bool

Whether to update the index prior to returning the result. Default is True.

stable : bool

Whether to use a stable set of shards. Default is None.

execution_stats : bool

Include execution statistics. Default is False.

partition : str

An optional partition ID. Only valid for partitioned databases. Default None.

Returns

dict : Keys are bookmark, docs, warning.

async def get(self,
docid: str,
*,
attachments: bool | None = None,
att_encoding_info: bool | None = None,
atts_since: Iterable[str] | None = None,
conflicts: bool | None = None,
deleted_conflicts: bool | None = None,
latest: bool | None = None,
local_seq: bool | None = None,
meta: bool | None = None,
open_revs: Iterable[str] | None = None,
rev: str | None = None,
revs: bool | None = None,
revs_info: bool | None = None,
check: bool | None = None,
default_value: Any | None = None) ‑> Document | Any

Expand source code

async def get(
    self,
    docid: str,
    *,
    attachments: bool | None = None,
    att_encoding_info: bool | None = None,
    atts_since: Iterable[str] | None = None,
    conflicts: bool | None = None,
    deleted_conflicts: bool | None = None,
    latest: bool | None = None,
    local_seq: bool | None = None,
    meta: bool | None = None,
    open_revs: Iterable[str] | None = None,
    rev: str | None = None,
    revs: bool | None = None,
    revs_info: bool | None = None,
    check: bool | None = None,
    default_value: Any | None = None,
) -> Document | Any:
    """
    Get a document by id.

    Parameters
    ----------
    docid : str
        The document's id.
    attachments : bool
        Includes attachment bodies in response. Default `None`.
    att_encoding_info : bool
        Includes encoding information in attachment stubs. Default `None`.
    atts_since : Iterable[str]
        Includes attachments only since specified revisions. Default `None`.
    conflicts : bool
        Includes conflict information. Default `None`.
    deleted_conflicts : bool
        Includes deleted conflicted revisions. Default `None`.
    latest : bool
        Forces retrieving the latest leaf revision. Default `None`.
    local_seq : bool
        Includes last update sequence for the document. Default `None`.
    meta : bool
        Equivalent to specifying all conflicts, deleted_conflicts and revs_info.
        Default `None`.
    open_revs : Iterable[str]
        Retrieves documents of specified leaf revisions. Default `None`.
    rev : str
        Retrieves document of specified revision. Default `None`.
    revs : bool
        Includes list of all known document revisions. Default `None`.
    revs_info : bool
        Includes detailed information for all known document revisions. Default `None`.
    check : bool
        If `True`, raise an exception if the document is not found. Default `None`.
    default_value : Any
        Value to return if `check=False` and document is not found. Default `None`.

    Returns
    -------
    Document | Any
    """
    try:
        return Document(
            **(
                await self._get(
                    resource=docid,
                    query_kwargs={
                        "attachments": attachments,
                        "att_encoding_info": att_encoding_info,
                        "atts_since": atts_since,
                        "conflicts": conflicts,
                        "deleted_conflicts": deleted_conflicts,
                        "latest": latest,
                        "local_seq": local_seq,
                        "meta": meta,
                        "open_revs": open_revs,
                        "rev": rev,
                        "revs": revs,
                        "revs_info": revs_info,
                    },
                )
            ).json()
        )
    except (CouchDBError, httpx.RequestError):
        if check:
            raise
        return default_value

Get a document by id.

Parameters

docid : str

The document's id.

attachments : bool

Includes attachment bodies in response. Default None.

att_encoding_info : bool

Includes encoding information in attachment stubs. Default None.

atts_since : Iterable[str]

Includes attachments only since specified revisions. Default None.

conflicts : bool

Includes conflict information. Default None.

deleted_conflicts : bool

Includes deleted conflicted revisions. Default None.

latest : bool

Forces retrieving the latest leaf revision. Default None.

local_seq : bool

Includes last update sequence for the document. Default None.

meta : bool

Equivalent to specifying all conflicts, deleted_conflicts and revs_info. Default None.

open_revs : Iterable[str]

Retrieves documents of specified leaf revisions. Default None.

rev : str

Retrieves document of specified revision. Default None.

revs : bool

Includes list of all known document revisions. Default None.

revs_info : bool

Includes detailed information for all known document revisions. Default None.

check : bool

If True, raise an exception if the document is not found. Default None.

default_value : Any

Value to return if check=False and document is not found. Default None.

Returns

Document | Any

 

async def get_attachment(self, docid: str, attname: str, rev: str | None = None) ‑> AttachmentDocument

Expand source code

async def get_attachment(
    self,
    docid: str,
    attname: str,
    rev: str | None = None,
) -> AttachmentDocument:
    """
    Get a document's attachment.

    Parameters
    ----------
    docid : str
        The document's id.
    attname : str
        The attachment's name.
    rev : str
        A specific revision. Default `None`.

    Returns
    -------
    AttachmentDocument
    """
    response = await self._get(f"{docid}/{attname}", query_kwargs={"rev": rev})
    content_md5 = response.headers.get("content-md5")
    digest_value = f"md5-{content_md5}" if content_md5 else None
    return AttachmentDocument(
        content=response.content,
        content_encoding=response.headers.get("content-encoding"),
        content_length=response.headers.get("content-length"),
        content_type=response.headers.get("content-type"),
        digest=digest_value,
    )

Get a document's attachment.

Parameters

docid : str

The document's id.

attname : str

The attachment's name.

rev : str

A specific revision. Default None.

Returns

AttachmentDocument

 

async def get_design(self, ddoc: str, **kwargs) ‑> Document

Expand source code

async def get_design(self, ddoc: str, **kwargs) -> Document:
    """
    Get a design document.

    Parameters
    ----------
    ddoc : str
        The design document's name.
    kwargs
        Further `AsyncDatabase.get` parameters.

    Returns
    -------
    Document
    """
    return await self.get(docid=f"_design/{ddoc}", **kwargs)

Get a design document.

Parameters

ddoc : str

The design document's name.

kwargs

Further AsyncDatabase.get() parameters.

Returns

Document

 

async def get_partition(self, partition_id: str) ‑> AsyncPartition

Expand source code

async def get_partition(self, partition_id: str) -> AsyncPartition:
    """
    Get a given partition.

    Parameters
    ----------
    partition_id : str
        The partition's ID.

    Returns
    -------
    AsyncPartition
    """
    return AsyncPartition(
        partition_id=partition_id,
        name=self.name,
        url=self.url,
        port=self.port,
        user=self._user,
        password=self._password,
        disable_ssl_verification=self.disable_ssl_verification,
        auth_method=self.auth_method,
        session=self.session,  # shared — child sets _owns_session=False
        _database=self,
    )

Get a given partition.

Parameters

partition_id : str

The partition's ID.

Returns

AsyncPartition

 

async def indexes(self) ‑> dict

Expand source code

async def indexes(self) -> dict:
    """
    Get a list of all indexes in the database.

    Returns
    -------
    dict : Keys are `total_rows` and `indexes`.
    """
    return (await self._get(resource="_index")).json()

Get a list of all indexes in the database.

Returns

dict : Keys are total_rows and indexes.

async def purge(self, data: dict) ‑> dict

Expand source code

async def purge(self, data: dict) -> dict:
    """
    Permanently purge the given `(id, rev)` pairs.

    Parameters
    ----------
    data : dict
        A dictionary with document IDs as keys and list of revisions as values.

    Returns
    -------
    dict
    """
    return (await self._post(resource="_purge", body=data)).json()

Permanently purge the given (id, rev) pairs.

Parameters

data : dict

A dictionary with document IDs as keys and list of revisions as values.

Returns

dict

 

async def put_attachment(self,
docid: str,
attname: str,
path: str | None = None,
*,
content: bytes | None = None,
content_type: str | None = None,
rev: str | None = None) ‑> tuple[str, bool, str]

Expand source code

async def put_attachment(
    self,
    docid: str,
    attname: str,
    path: str | None = None,
    *,
    content: bytes | None = None,
    content_type: str | None = None,
    rev: str | None = None,
) -> tuple[str, bool, str]:
    """
    Upload content as an attachment to the specified document.

    Parameters
    ----------
    docid : str
        The document's id.
    attname : str
        The attachment's name.
    path : str
        Path to a local file to upload. Mutually exclusive with `content`.
        The file is streamed to CouchDB in chunks, so its full contents are never
        loaded into memory.
    content : bytes
        Raw bytes to upload. Mutually exclusive with `path`.
    content_type : str
        The attachment's MIME type. Required when `content` is provided.
    rev : str
        The document's current revision. Required for existing documents.

    Returns
    -------
    tuple[str, bool, str] : (id, ok, rev)
    """
    if (not content and not path) or (content and path):
        raise ValueError(
            'Precisely one of the arguments "content" and "path" must be provided.'
        )
    if content and not content_type:
        raise ValueError('Argument "content_type" cannot be empty when "content" is provided.')
    resource = f"{docid}/{attname}"
    query_kwargs = {"rev": rev}
    content_type = content_type if content_type else mimetypes.guess_type(path)[0]
    if path:
        with open(path, "rb") as file:
            content_length = os.fstat(file.fileno()).st_size
            response = await self._put(
                resource=resource,
                query_kwargs=query_kwargs,
                content=_stream_file_content(file),
                headers={
                    "content-type": content_type,
                    "content-length": str(content_length),
                },
            )
    else:
        response = await self._put(
            resource=resource,
            query_kwargs=query_kwargs,
            content=content,
            headers={"content-type": content_type},
        )
    data = response.json()
    return data["id"], data["ok"], data["rev"]

Upload content as an attachment to the specified document.

Parameters

docid : str

The document's id.

attname : str

The attachment's name.

path : str

Path to a local file to upload. Mutually exclusive with content. The file is streamed to CouchDB in chunks, so its full contents are never loaded into memory.

content : bytes

Raw bytes to upload. Mutually exclusive with path.

content_type : str

The attachment's MIME type. Required when content is provided.

rev : str

The document's current revision. Required for existing documents.

Returns

tuple[str, bool, str] : (id, ok, rev)

 

async def put_design(self,
ddoc: str,
*,
rev: str | None = None,
language: str | None = None,
options: dict | None = None,
filters: dict | None = None,
updates: dict | None = None,
validate_doc_update: str | None = None,
views: dict | None = None,
autoupdate: bool | None = None,
partitioned: bool | None = None,
**kwargs) ‑> tuple[str, bool, str]

Expand source code

async def put_design(
    self,
    ddoc: str,
    *,
    rev: str | None = None,
    language: str | None = None,
    options: dict | None = None,
    filters: dict | None = None,
    updates: dict | None = None,
    validate_doc_update: str | None = None,
    views: dict | None = None,
    autoupdate: bool | None = None,
    partitioned: bool | None = None,
    **kwargs,
) -> tuple[str, bool, str]:
    """
    Create or update a named design document.

    Parameters
    ----------
    ddoc : str
        The design document's name.
    rev : str
        The design document's revision in case of an update.
    language : str
        Query Server to process design document functions.
    options : dict
        View's default options.
    filters : dict
        Filter functions definition.
    updates : dict
        Update functions definition.
    validate_doc_update : str
        Validate document update function source.
    views : dict
        View functions definition.
    autoupdate : bool
        Indicates whether to automatically build indexes.
    partitioned : bool
        Set to `True` for a partitioned design.
    kwargs
        Further `AsyncDatabase.save` parameters.

    Returns
    -------
    tuple[str, bool, str] : (id, ok, rev)
    """
    if partitioned:
        options = {**(options or {}), "partitioned": partitioned}
    return await self.save(
        doc=rm_nones_from_dict(
            {
                "_id": f"_design/{ddoc}",
                "_rev": rev,
                "language": language,
                "options": options,
                "filters": filters,
                "updates": updates,
                "validate_doc_update": validate_doc_update,
                "views": views,
                "autoupdate": autoupdate,
            }
        ),
        **kwargs,
    )

Create or update a named design document.

Parameters

ddoc : str

The design document's name.

rev : str

The design document's revision in case of an update.

language : str

Query Server to process design document functions.

options : dict

View's default options.

filters : dict

Filter functions definition.

updates : dict

Update functions definition.

validate_doc_update : str

Validate document update function source.

views : dict

View functions definition.

autoupdate : bool

Indicates whether to automatically build indexes.

partitioned : bool

Set to True for a partitioned design.

kwargs

Further AsyncDatabase.save() parameters.

Returns

tuple[str, bool, str] : (id, ok, rev)

 

async def save(self,
doc: dict | Document,
batch: bool | None = None,
new_edits: bool | None = None,
path: str | None = None) ‑> tuple[str, bool, str]

Expand source code

async def save(
    self,
    doc: dict | Document,
    batch: bool | None = None,
    new_edits: bool | None = None,
    path: str | None = None,
) -> tuple[str, bool, str]:
    """
    Create a new named document, or a new revision of an existing document.

    Parameters
    ----------
    doc : dict | Document
        A dictionary or Document instance with a valid `_id`, and `_rev` if updating.
    batch : bool
        Store document in batch mode. Default `None`.
    new_edits : bool
        Prevents insertion of a conflicting document. Default `None`.
    path : str
        Database path, e.g. `_design`. Default `None`.

    Returns
    -------
    tuple[str, bool, str] : (id, ok, rev)
    """
    batch = "ok" if batch else None
    data = (
        await self._put(
            resource=f"{path}/{doc.get('_id')}" if path else doc.get("_id"),
            body=doc,
            query_kwargs={
                "batch": "ok" if batch else None,
                "new_edits": new_edits,
                "rev": doc.get("_rev"),
            },
        )
    ).json()
    return data["id"], data["ok"], data["rev"]

Create a new named document, or a new revision of an existing document.

Parameters

doc : dict | Document

A dictionary or Document instance with a valid _id, and _rev if updating.

batch : bool

Store document in batch mode. Default None.

new_edits : bool

Prevents insertion of a conflicting document. Default None.

path : str

Database path, e.g. _design. Default None.

Returns

tuple[str, bool, str] : (id, ok, rev)

 

async def save_index(self,
index: dict,
ddoc: str | None = None,
name: str | None = None,
index_type: str | None = 'json',
partitioned: bool | None = None) ‑> tuple[str, str, str]

Expand source code

async def save_index(
    self,
    index: dict,
    ddoc: str | None = None,
    name: str | None = None,
    index_type: str | None = "json",
    partitioned: bool | None = None,
) -> tuple[str, str, str]:
    """
    Create a new index on a database.

    Parameters
    ----------
    index : dict
        Dictionary describing the index to create.
    ddoc : str
        Name of the design document. Default `None` (auto-generated).
    name : str
        Name of the index. Default `None` (auto-generated).
    index_type : str
        `json` or `text`. Default `json`.
    partitioned : bool
        Whether the index is partitioned or global. Default `None`.

    Returns
    -------
    tuple[str, str, str] : (result, id, name)
    """
    data = (
        await self._post(
            resource="_index",
            body=rm_nones_from_dict(
                {
                    "index": index,
                    "ddoc": ddoc,
                    "name": name,
                    "type": index_type,
                    "partitioned": partitioned,
                }
            ),
        )
    ).json()
    return data["result"], data["id"], data["name"]

Create a new index on a database.

Parameters

index : dict

Dictionary describing the index to create.

ddoc : str

Name of the design document. Default None (auto-generated).

name : str

Name of the index. Default None (auto-generated).

index_type : str

json or text. Default json.

partitioned : bool

Whether the index is partitioned or global. Default None.

Returns

tuple[str, str, str] : (result, id, name)

 

async def security(self) ‑> SecurityDocument

Expand source code

async def security(self) -> SecurityDocument:
    """
    Returns the current security object from the specified database.

    Returns
    -------
    SecurityDocument
    """
    data = (await self._get(resource="_security")).json()
    return SecurityDocument(**data)

Returns the current security object from the specified database.

Returns

SecurityDocument

 

async def update_security(self,
admins: dict | SecurityDocumentElement | None = None,
members: dict | SecurityDocumentElement | None = None) ‑> bool

Expand source code

async def update_security(
    self,
    admins: dict | SecurityDocumentElement | None = None,
    members: dict | SecurityDocumentElement | None = None,
) -> bool:
    """
    Update database security.

    Parameters
    ----------
    admins : dict | SecurityDocumentElement
        Object with `names` and `roles` fields.
    members : dict | SecurityDocumentElement
        Object with `names` and `roles` fields.

    Returns
    -------
    bool : Operation status.
    """
    return (
        await self._put(resource="_security", body={"admins": admins, "members": members})
    ).json()["ok"]

Update database security.

Parameters

admins : dict | SecurityDocumentElement

Object with names and roles fields.

members : dict | SecurityDocumentElement

Object with names and roles fields.

Returns

bool : Operation status.

async def view(self,
ddoc: str,
view: str | None = None,
*,
partition: str | None = None,
conflicts: bool | None = None,
descending: bool | None = None,
endkey: Any | None = None,
endkey_docid: str | None = None,
group: bool | None = None,
group_level: int | None = None,
include_docs: bool | None = None,
attachments: bool | None = None,
att_encoding_info: bool | None = None,
inclusive_end: bool | None = None,
key: str | None = None,
keys: Iterable[str] | None = None,
limit: int | None = None,
reduce: bool | None = None,
skip: int | None = None,
sort: bool | None = None,
stable: bool | None = None,
startkey: Any | None = None,
startkey_docid: str | None = None,
update: str | None = None,
update_seq: bool | None = None) ‑> ViewResult

Expand source code

async def view(
    self,
    ddoc: str,
    view: str | None = None,
    *,
    partition: str | None = None,
    conflicts: bool | None = None,
    descending: bool | None = None,
    endkey: Any | None = None,
    endkey_docid: str | None = None,
    group: bool | None = None,
    group_level: int | None = None,
    include_docs: bool | None = None,
    attachments: bool | None = None,
    att_encoding_info: bool | None = None,
    inclusive_end: bool | None = None,
    key: str | None = None,
    keys: Iterable[str] | None = None,
    limit: int | None = None,
    reduce: bool | None = None,
    skip: int | None = None,
    sort: bool | None = None,
    stable: bool | None = None,
    startkey: Any | None = None,
    startkey_docid: str | None = None,
    update: str | None = None,
    update_seq: bool | None = None,
) -> ViewResult:
    """
    Executes the specified view function from the specified design document.

    Parameters
    ----------
    ddoc : str
        The corresponding design document's id.
    view : str
        The view's id.
    partition : str
        An optional partition ID. Only valid for partitioned databases. Default `None`.
    conflicts : bool
        Include conflicts information in response. Default `None`.
    descending : bool
        Return documents in descending order. Default `None`.
    endkey : Any
        Stop returning records at this key. Default `None`.
    endkey_docid : str
        Stop returning records at this document ID. Default `None`.
    group : bool
        Group results using the reduce function. Default `None`.
    group_level : int
        Specify the group level. Default `None`.
    include_docs : bool
        Include the associated document with each row. Default `None`.
    attachments : bool
        Include Base64-encoded attachment content. Default `None`.
    att_encoding_info : bool
        Include encoding information in attachment stubs. Default `None`.
    inclusive_end : bool
        Whether the end key should be included in the result. Default `None`.
    key : str
        Return only documents matching this key. Default `None`.
    keys : Iterable[str]
        Return only documents matching these keys. Default `None`.
    limit : int
        Maximum number of documents to return. Default `None`.
    reduce : bool
        Use the reduction function. Default `None`.
    skip : int
        Skip this number of records. Default `None`.
    sort : bool
        Sort returned rows. Default `None`.
    stable : bool
        Use a stable set of shards. Default `None`.
    startkey : Any
        Return records starting with this key. Default `None`.
    startkey_docid : str
        Return records starting with this document ID. Default `None`.
    update : str
        Whether to update the view prior to responding (`true`, `false`, `lazy`).
        Default `None`.
    update_seq : bool
        Include the `update_seq` value in the response. Default `None`.

    Returns
    -------
    ViewResult
    """
    path = partitioned_db_resource_parser(
        resource="_design",
        partition=partition,
    )
    return ViewResult(
        **(
            await self._get(
                resource=f"{path}/{ddoc}/_view/{view}" if (ddoc and view) else ddoc,
                query_kwargs={
                    "conflicts": conflicts,
                    "descending": descending,
                    "endkey": endkey,
                    "endkey_docid": endkey_docid,
                    "group": group,
                    "group_level": group_level,
                    "include_docs": include_docs,
                    "attachments": attachments,
                    "att_encoding_info": att_encoding_info,
                    "inclusive_end": inclusive_end,
                    "key": key,
                    "keys": keys,
                    "limit": limit,
                    "reduce": reduce,
                    "skip": skip,
                    "sorted": sort,
                    "stable": stable,
                    "startkey": startkey,
                    "startkey_docid": startkey_docid,
                    "update": update,
                    "update_seq": update_seq,
                },
            )
        ).json()
    )

Executes the specified view function from the specified design document.

Parameters

ddoc : str

The corresponding design document's id.

view : str

The view's id.

partition : str

An optional partition ID. Only valid for partitioned databases. Default None.

conflicts : bool

Include conflicts information in response. Default None.

descending : bool

Return documents in descending order. Default None.

endkey : Any

Stop returning records at this key. Default None.

endkey_docid : str

Stop returning records at this document ID. Default None.

group : bool

Group results using the reduce function. Default None.

group_level : int

Specify the group level. Default None.

include_docs : bool

Include the associated document with each row. Default None.

attachments : bool

Include Base64-encoded attachment content. Default None.

att_encoding_info : bool

Include encoding information in attachment stubs. Default None.

inclusive_end : bool

Whether the end key should be included in the result. Default None.

key : str

Return only documents matching this key. Default None.

keys : Iterable[str]

Return only documents matching these keys. Default None.

limit : int

Maximum number of documents to return. Default None.

reduce : bool

Use the reduction function. Default None.

skip : int

Skip this number of records. Default None.

sort : bool

Sort returned rows. Default None.

stable : bool

Use a stable set of shards. Default None.

startkey : Any

Return records starting with this key. Default None.

startkey_docid : str

Return records starting with this document ID. Default None.

update : str

Whether to update the view prior to responding (true, false, lazy). Default None.

update_seq : bool

Include the update_seq value in the response. Default None.

Returns

ViewResult

 

Inherited members

• AsyncBase:
  • aclose
  • check
  • info
  • rev
  • url

class AsyncPartition (partition_id: str,
name: str,
*,
url: str | None = None,
port: int | None = None,
user: str | None = None,
password: str | None = None,
disable_ssl_verification: bool = False,
auth_method: str | None = None,
timeout: int | None = None,
session: httpx.AsyncClient | None = None)

Expand source code

class AsyncPartition(AsyncDatabase):
    """
    Async CouchDB partition client. Mirrors `Partition` with `async def` methods throughout.
    """

    def __init__(
        self,
        partition_id: str,
        name: str,
        *,
        url: str | None = None,
        port: int | None = None,
        user: str | None = None,
        password: str | None = None,
        disable_ssl_verification: bool = False,
        auth_method: str | None = None,
        timeout: int | None = None,
        session: httpx.AsyncClient | None = None,
        _database: Any = None,
    ) -> None:
        """

        Parameters
        ----------
        partition_id : str
            The partition's ID.
        name : str
            The name of the database.
        url : str
            The url of the CouchDB server.
        port : int
            The port of the CouchDB server.
        user : str
            The CouchDB admin username.
        password : str
            The CouchDB admin password.
        disable_ssl_verification : bool
            Controls whether to verify the server's TLS certificate. Default `False`.
        auth_method : str
            Authentication method. Default is `couchdb3.utils.DEFAULT_AUTH_METHOD`.
        timeout : int
            The default timeout for requests. Default `None`.
        session : httpx.AsyncClient
            A specific async client to use. Optional.
        _database : AsyncDatabase
            The owning `AsyncDatabase` instance. Set internally by
            `AsyncDatabase.get_partition()` to keep the database alive for the lifetime of
            this partition object. Not part of the public constructor API — pass `None`
            (default) when constructing an `AsyncPartition` directly.
        """
        super().__init__(
            name=name,
            url=url,
            session=session,
            port=port,
            user=user,
            password=password,
            disable_ssl_verification=disable_ssl_verification,
            auth_method=auth_method,
            timeout=timeout,
        )
        self.partition_id = partition_id
        self._database = _database

    @property
    def database(self):
        """
        The `AsyncDatabase` instance this partition was obtained from, or `None` if the
        partition was constructed directly (i.e. not via `AsyncDatabase.get_partition()`).

        Read-only. Setting this attribute raises `AttributeError`.

        Returns
        -------
        AsyncDatabase | None
        """
        return self._database

    def __repr__(self) -> str:
        return f"{super().__repr__()}/{self.partition_id}"

    async def all_docs(self, keys: Iterable[str] | None = None, **kwargs) -> ViewResult:
        """
        Executes the built-in _all_docs view, returning all documents in the partition.

        Parameters
        ----------
        keys : Iterable[str]
            Return only documents matching these keys. Default `None`.
        kwargs
            Further `AsyncDatabase.view` parameters.

        Returns
        -------
        ViewResult
        """
        return await super().all_docs(partition=self.partition_id, keys=keys, **kwargs)

    async def info(self) -> dict:
        """
        Return the partition's info.

        Returns
        -------
        dict
        """
        return await super().info(partition=self.partition_id)

    async def find(
        self,
        selector: dict,
        limit: int = 25,
        skip: int = 0,
        sort: list[dict] | None = None,
        fields: list[str] | None = None,
        use_index: str | list[str] | None = None,
        conflicts: bool = False,
        r: int = 1,
        bookmark: str | None = None,
        update: bool = True,
        stable: bool | None = None,
        execution_stats: bool = False,
    ) -> dict:
        """
        See `AsyncDatabase.find`.
        """
        return await super().find(
            selector=selector,
            limit=limit,
            skip=skip,
            sort=sort,
            fields=fields,
            use_index=use_index,
            conflicts=conflicts,
            r=r,
            bookmark=bookmark,
            update=update,
            stable=stable,
            execution_stats=execution_stats,
            partition=self.partition_id,
        )

    async def view(
        self,
        ddoc: str,
        view: str | None = None,
        *,
        conflicts: bool | None = None,
        descending: bool | None = None,
        endkey: Any | None = None,
        endkey_docid: str | None = None,
        group: bool | None = None,
        group_level: int | None = None,
        include_docs: bool | None = None,
        attachments: bool | None = None,
        att_encoding_info: bool | None = None,
        inclusive_end: bool | None = None,
        key: str | None = None,
        keys: Iterable[str] | None = None,
        limit: int | None = None,
        reduce: bool | None = None,
        skip: int | None = None,
        sort: bool | None = None,
        stable: bool | None = None,
        startkey: Any | None = None,
        startkey_docid: str | None = None,
        update: str | None = None,
        update_seq: bool | None = None,
    ) -> ViewResult:
        """
        See `AsyncDatabase.view`.
        """
        return await super().view(
            ddoc=ddoc,
            view=view,
            partition=self.partition_id,
            conflicts=conflicts,
            descending=descending,
            endkey=endkey,
            endkey_docid=endkey_docid,
            group=group,
            group_level=group_level,
            include_docs=include_docs,
            attachments=attachments,
            att_encoding_info=att_encoding_info,
            inclusive_end=inclusive_end,
            key=key,
            keys=keys,
            limit=limit,
            reduce=reduce,
            skip=skip,
            sort=sort,
            stable=stable,
            startkey=startkey,
            startkey_docid=startkey_docid,
            update=update,
            update_seq=update_seq,
        )

    async def bulk_docs(self, docs: list[dict | Document], new_edits: bool = True) -> list[dict]:
        """See `AsyncDatabase.bulk_docs`. Prepends partition ID to document IDs."""
        return await super().bulk_docs(
            docs=[self.add_partition_to_doc(doc) for doc in docs],
            new_edits=new_edits,
        )

    async def bulk_get(self, docs: list[dict | Document], revs: bool = False) -> list[dict]:
        """See `AsyncDatabase.bulk_get`. Prepends partition ID to document IDs."""
        return await super().bulk_get(
            docs=[self.add_partition_to_bulk_get_doc(doc) for doc in docs],
            revs=revs,
        )

    async def copy(
        self,
        docid: str,
        destid: str,
        rev: str | None = None,
        destrev: str | None = None,
    ) -> tuple[str, bool, str]:
        """See `AsyncDatabase.copy`. Prepends partition ID to document IDs."""
        return await super().copy(
            docid=self.add_partition_to_str(docid),
            destid=self.add_partition_to_str(destid),
            rev=rev,
            destrev=destrev,
        )

    async def create(
        self, doc: dict | Document, *, batch: bool | None = None
    ) -> tuple[str, bool, str]:
        """See `AsyncDatabase.create`. Prepends partition ID to document ID."""
        return await super().create(
            doc=self.add_partition_to_doc(doc),
            batch=batch,
        )

    async def delete(self, docid: str, rev: str, *, batch: bool | None = None) -> bool:
        """See `AsyncDatabase.delete`. Prepends partition ID to document ID."""
        return await super().delete(
            docid=self.add_partition_to_str(docid),
            rev=rev,
            batch=batch,
        )

    async def delete_attachment(
        self, docid: str, attname: str, rev: str, *, batch: bool = False
    ) -> bool:
        """See `AsyncDatabase.delete_attachment`. Prepends partition ID to document ID."""
        return await super().delete_attachment(
            docid=self.add_partition_to_str(docid),
            attname=attname,
            rev=rev,
            batch=batch,
        )

    async def get(
        self,
        docid: str,
        *,
        attachments: bool | None = None,
        att_encoding_info: bool | None = None,
        atts_since: Iterable[str] | None = None,
        conflicts: bool | None = None,
        deleted_conflicts: bool | None = None,
        latest: bool | None = None,
        local_seq: bool | None = None,
        meta: bool | None = None,
        open_revs: Iterable[str] | None = None,
        rev: str | None = None,
        revs: bool | None = None,
        revs_info: bool | None = None,
        check: bool | None = False,
        default_value: Any | None = None,
    ) -> Document | Any:
        """See `AsyncDatabase.get`. Prepends partition ID to document ID."""
        return await super().get(
            docid=self.add_partition_to_str(docid),
            attachments=attachments,
            att_encoding_info=att_encoding_info,
            atts_since=atts_since,
            conflicts=conflicts,
            deleted_conflicts=deleted_conflicts,
            latest=latest,
            local_seq=local_seq,
            meta=meta,
            open_revs=open_revs,
            rev=rev,
            revs=revs,
            revs_info=revs_info,
            check=check,
            default_value=default_value,
        )

    async def get_attachment(
        self, docid: str, attname: str, rev: str | None = None
    ) -> AttachmentDocument:
        """See `AsyncDatabase.get_attachment`. Prepends partition ID to document ID."""
        return await super().get_attachment(
            docid=self.add_partition_to_str(docid),
            attname=attname,
            rev=rev,
        )

    async def put_attachment(
        self,
        docid: str,
        attname: str,
        path: str | None = None,
        *,
        content: bytes | None = None,
        content_type: str | None = None,
        rev: str | None = None,
    ) -> tuple[str, bool, str]:
        """See `AsyncDatabase.put_attachment`. Prepends partition ID to document ID."""
        return await super().put_attachment(
            docid=self.add_partition_to_str(docid),
            attname=attname,
            content_type=content_type,
            path=path,
            content=content,
            rev=rev,
        )

    async def rev(self, resource: str) -> str | None:
        """See `AsyncDatabase.rev`. Prepends partition ID to the resource."""
        return await super().rev(self.add_partition_to_str(resource))

    async def save(
        self,
        doc: dict | Document,
        batch: bool | None = None,
        new_edits: bool | None = None,
        path: str | None = None,
    ) -> tuple[str, bool, str]:
        """See `AsyncDatabase.save`. Prepends partition ID to document ID."""
        return await super().save(
            doc=self.add_partition_to_doc(doc),
            batch=batch,
            new_edits=new_edits,
            path=path,
        )

    def add_partition_to_str(self, string: str) -> str:
        """Append the instance's partition ID to a string if not already present."""
        if string.startswith(f"{self.partition_id}:"):
            return string
        return f"{self.partition_id}:{string}"

    def add_partition_to_doc(self, doc: Document | dict) -> Document | dict:
        """Append the instance's partition ID to the document's `_id`."""
        docid = doc.get("_id")
        if docid is None:
            return doc
        doc["_id"] = self.add_partition_to_str(docid)
        return doc

    def add_partition_to_bulk_get_doc(self, doc: Document | dict) -> Document | dict:
        """Append the instance's partition ID to a `bulk_get` document's `id` (or `_id`)."""
        key = "_id" if "_id" in doc else "id" if "id" in doc else None
        if key is None:
            return doc
        doc[key] = self.add_partition_to_str(doc[key])
        return doc

Async CouchDB partition client. Mirrors Partition with async def methods throughout.

Parameters

partition_id : str

The partition's ID.

name : str

The name of the database.

url : str

The url of the CouchDB server.

port : int

The port of the CouchDB server.

user : str

The CouchDB admin username.

password : str

The CouchDB admin password.

disable_ssl_verification : bool

Controls whether to verify the server's TLS certificate. Default False.

auth_method : str

Authentication method. Default is DEFAULT_AUTH_METHOD.

timeout : int

The default timeout for requests. Default None.

session : httpx.AsyncClient

A specific async client to use. Optional.

_database : AsyncDatabase

The owning AsyncDatabase instance. Set internally by AsyncDatabase.get_partition() to keep the database alive for the lifetime of this partition object. Not part of the public constructor API — pass None (default) when constructing an AsyncPartition directly.

Ancestors

• AsyncDatabase
• AsyncBase

Instance variables

prop database

Expand source code

@property
def database(self):
    """
    The `AsyncDatabase` instance this partition was obtained from, or `None` if the
    partition was constructed directly (i.e. not via `AsyncDatabase.get_partition()`).

    Read-only. Setting this attribute raises `AttributeError`.

    Returns
    -------
    AsyncDatabase | None
    """
    return self._database

The AsyncDatabase instance this partition was obtained from, or None if the partition was constructed directly (i.e. not via AsyncDatabase.get_partition()).

Read-only. Setting this attribute raises AttributeError.

Returns

AsyncDatabase | None

 

Methods

def add_partition_to_bulk_get_doc(self, doc: Document | dict) ‑> Document | dict

Expand source code

def add_partition_to_bulk_get_doc(self, doc: Document | dict) -> Document | dict:
    """Append the instance's partition ID to a `bulk_get` document's `id` (or `_id`)."""
    key = "_id" if "_id" in doc else "id" if "id" in doc else None
    if key is None:
        return doc
    doc[key] = self.add_partition_to_str(doc[key])
    return doc

Append the instance's partition ID to a bulk_get document's id (or _id).

def add_partition_to_doc(self, doc: Document | dict) ‑> Document | dict

Expand source code

def add_partition_to_doc(self, doc: Document | dict) -> Document | dict:
    """Append the instance's partition ID to the document's `_id`."""
    docid = doc.get("_id")
    if docid is None:
        return doc
    doc["_id"] = self.add_partition_to_str(docid)
    return doc

Append the instance's partition ID to the document's _id.

def add_partition_to_str(self, string: str) ‑> str

Expand source code

def add_partition_to_str(self, string: str) -> str:
    """Append the instance's partition ID to a string if not already present."""
    if string.startswith(f"{self.partition_id}:"):
        return string
    return f"{self.partition_id}:{string}"

Append the instance's partition ID to a string if not already present.

async def all_docs(self, keys: Iterable[str] | None = None, **kwargs) ‑> ViewResult

Expand source code

async def all_docs(self, keys: Iterable[str] | None = None, **kwargs) -> ViewResult:
    """
    Executes the built-in _all_docs view, returning all documents in the partition.

    Parameters
    ----------
    keys : Iterable[str]
        Return only documents matching these keys. Default `None`.
    kwargs
        Further `AsyncDatabase.view` parameters.

    Returns
    -------
    ViewResult
    """
    return await super().all_docs(partition=self.partition_id, keys=keys, **kwargs)

Executes the built-in _all_docs view, returning all documents in the partition.

Parameters

keys : Iterable[str]

Return only documents matching these keys. Default None.

kwargs

Further AsyncDatabase.view() parameters.

Returns

ViewResult

 

async def bulk_docs(self, docs: list[dict | Document], new_edits: bool = True) ‑> list[dict]

Expand source code

async def bulk_docs(self, docs: list[dict | Document], new_edits: bool = True) -> list[dict]:
    """See `AsyncDatabase.bulk_docs`. Prepends partition ID to document IDs."""
    return await super().bulk_docs(
        docs=[self.add_partition_to_doc(doc) for doc in docs],
        new_edits=new_edits,
    )

See AsyncDatabase.bulk_docs(). Prepends partition ID to document IDs.

async def bulk_get(self, docs: list[dict | Document], revs: bool = False) ‑> list[dict]

Expand source code

async def bulk_get(self, docs: list[dict | Document], revs: bool = False) -> list[dict]:
    """See `AsyncDatabase.bulk_get`. Prepends partition ID to document IDs."""
    return await super().bulk_get(
        docs=[self.add_partition_to_bulk_get_doc(doc) for doc in docs],
        revs=revs,
    )

See AsyncDatabase.bulk_get(). Prepends partition ID to document IDs.

async def copy(self, docid: str, destid: str, rev: str | None = None, destrev: str | None = None) ‑> tuple[str, bool, str]

Expand source code

async def copy(
    self,
    docid: str,
    destid: str,
    rev: str | None = None,
    destrev: str | None = None,
) -> tuple[str, bool, str]:
    """See `AsyncDatabase.copy`. Prepends partition ID to document IDs."""
    return await super().copy(
        docid=self.add_partition_to_str(docid),
        destid=self.add_partition_to_str(destid),
        rev=rev,
        destrev=destrev,
    )

See AsyncDatabase.copy(). Prepends partition ID to document IDs.

async def create(self, doc: dict | Document, *, batch: bool | None = None) ‑> tuple[str, bool, str]

Expand source code

async def create(
    self, doc: dict | Document, *, batch: bool | None = None
) -> tuple[str, bool, str]:
    """See `AsyncDatabase.create`. Prepends partition ID to document ID."""
    return await super().create(
        doc=self.add_partition_to_doc(doc),
        batch=batch,
    )

See AsyncDatabase.create(). Prepends partition ID to document ID.

async def delete(self, docid: str, rev: str, *, batch: bool | None = None) ‑> bool

Expand source code

async def delete(self, docid: str, rev: str, *, batch: bool | None = None) -> bool:
    """See `AsyncDatabase.delete`. Prepends partition ID to document ID."""
    return await super().delete(
        docid=self.add_partition_to_str(docid),
        rev=rev,
        batch=batch,
    )

See AsyncDatabase.delete(). Prepends partition ID to document ID.

async def delete_attachment(self, docid: str, attname: str, rev: str, *, batch: bool = False) ‑> bool

Expand source code

async def delete_attachment(
    self, docid: str, attname: str, rev: str, *, batch: bool = False
) -> bool:
    """See `AsyncDatabase.delete_attachment`. Prepends partition ID to document ID."""
    return await super().delete_attachment(
        docid=self.add_partition_to_str(docid),
        attname=attname,
        rev=rev,
        batch=batch,
    )

See AsyncDatabase.delete_attachment(). Prepends partition ID to document ID.

async def find(self,
selector: dict,
limit: int = 25,
skip: int = 0,
sort: list[dict] | None = None,
fields: list[str] | None = None,
use_index: str | list[str] | None = None,
conflicts: bool = False,
r: int = 1,
bookmark: str | None = None,
update: bool = True,
stable: bool | None = None,
execution_stats: bool = False) ‑> dict

Expand source code

async def find(
    self,
    selector: dict,
    limit: int = 25,
    skip: int = 0,
    sort: list[dict] | None = None,
    fields: list[str] | None = None,
    use_index: str | list[str] | None = None,
    conflicts: bool = False,
    r: int = 1,
    bookmark: str | None = None,
    update: bool = True,
    stable: bool | None = None,
    execution_stats: bool = False,
) -> dict:
    """
    See `AsyncDatabase.find`.
    """
    return await super().find(
        selector=selector,
        limit=limit,
        skip=skip,
        sort=sort,
        fields=fields,
        use_index=use_index,
        conflicts=conflicts,
        r=r,
        bookmark=bookmark,
        update=update,
        stable=stable,
        execution_stats=execution_stats,
        partition=self.partition_id,
    )

See AsyncDatabase.find().

async def get(self,
docid: str,
*,
attachments: bool | None = None,
att_encoding_info: bool | None = None,
atts_since: Iterable[str] | None = None,
conflicts: bool | None = None,
deleted_conflicts: bool | None = None,
latest: bool | None = None,
local_seq: bool | None = None,
meta: bool | None = None,
open_revs: Iterable[str] | None = None,
rev: str | None = None,
revs: bool | None = None,
revs_info: bool | None = None,
check: bool | None = False,
default_value: Any | None = None) ‑> Document | Any

Expand source code

async def get(
    self,
    docid: str,
    *,
    attachments: bool | None = None,
    att_encoding_info: bool | None = None,
    atts_since: Iterable[str] | None = None,
    conflicts: bool | None = None,
    deleted_conflicts: bool | None = None,
    latest: bool | None = None,
    local_seq: bool | None = None,
    meta: bool | None = None,
    open_revs: Iterable[str] | None = None,
    rev: str | None = None,
    revs: bool | None = None,
    revs_info: bool | None = None,
    check: bool | None = False,
    default_value: Any | None = None,
) -> Document | Any:
    """See `AsyncDatabase.get`. Prepends partition ID to document ID."""
    return await super().get(
        docid=self.add_partition_to_str(docid),
        attachments=attachments,
        att_encoding_info=att_encoding_info,
        atts_since=atts_since,
        conflicts=conflicts,
        deleted_conflicts=deleted_conflicts,
        latest=latest,
        local_seq=local_seq,
        meta=meta,
        open_revs=open_revs,
        rev=rev,
        revs=revs,
        revs_info=revs_info,
        check=check,
        default_value=default_value,
    )

See AsyncDatabase.get(). Prepends partition ID to document ID.

async def get_attachment(self, docid: str, attname: str, rev: str | None = None) ‑> AttachmentDocument

Expand source code

async def get_attachment(
    self, docid: str, attname: str, rev: str | None = None
) -> AttachmentDocument:
    """See `AsyncDatabase.get_attachment`. Prepends partition ID to document ID."""
    return await super().get_attachment(
        docid=self.add_partition_to_str(docid),
        attname=attname,
        rev=rev,
    )

See AsyncDatabase.get_attachment(). Prepends partition ID to document ID.

async def info(self) ‑> dict

Expand source code

async def info(self) -> dict:
    """
    Return the partition's info.

    Returns
    -------
    dict
    """
    return await super().info(partition=self.partition_id)

Return the partition's info.

Returns

dict

 

async def put_attachment(self,
docid: str,
attname: str,
path: str | None = None,
*,
content: bytes | None = None,
content_type: str | None = None,
rev: str | None = None) ‑> tuple[str, bool, str]

Expand source code

async def put_attachment(
    self,
    docid: str,
    attname: str,
    path: str | None = None,
    *,
    content: bytes | None = None,
    content_type: str | None = None,
    rev: str | None = None,
) -> tuple[str, bool, str]:
    """See `AsyncDatabase.put_attachment`. Prepends partition ID to document ID."""
    return await super().put_attachment(
        docid=self.add_partition_to_str(docid),
        attname=attname,
        content_type=content_type,
        path=path,
        content=content,
        rev=rev,
    )

See AsyncDatabase.put_attachment(). Prepends partition ID to document ID.

async def rev(self, resource: str) ‑> str | None

Expand source code

async def rev(self, resource: str) -> str | None:
    """See `AsyncDatabase.rev`. Prepends partition ID to the resource."""
    return await super().rev(self.add_partition_to_str(resource))

See AsyncBase.rev(). Prepends partition ID to the resource.

async def save(self,
doc: dict | Document,
batch: bool | None = None,
new_edits: bool | None = None,
path: str | None = None) ‑> tuple[str, bool, str]

Expand source code

async def save(
    self,
    doc: dict | Document,
    batch: bool | None = None,
    new_edits: bool | None = None,
    path: str | None = None,
) -> tuple[str, bool, str]:
    """See `AsyncDatabase.save`. Prepends partition ID to document ID."""
    return await super().save(
        doc=self.add_partition_to_doc(doc),
        batch=batch,
        new_edits=new_edits,
        path=path,
    )

See AsyncDatabase.save(). Prepends partition ID to document ID.

async def view(self,
ddoc: str,
view: str | None = None,
*,
conflicts: bool | None = None,
descending: bool | None = None,
endkey: Any | None = None,
endkey_docid: str | None = None,
group: bool | None = None,
group_level: int | None = None,
include_docs: bool | None = None,
attachments: bool | None = None,
att_encoding_info: bool | None = None,
inclusive_end: bool | None = None,
key: str | None = None,
keys: Iterable[str] | None = None,
limit: int | None = None,
reduce: bool | None = None,
skip: int | None = None,
sort: bool | None = None,
stable: bool | None = None,
startkey: Any | None = None,
startkey_docid: str | None = None,
update: str | None = None,
update_seq: bool | None = None) ‑> ViewResult

Expand source code

async def view(
    self,
    ddoc: str,
    view: str | None = None,
    *,
    conflicts: bool | None = None,
    descending: bool | None = None,
    endkey: Any | None = None,
    endkey_docid: str | None = None,
    group: bool | None = None,
    group_level: int | None = None,
    include_docs: bool | None = None,
    attachments: bool | None = None,
    att_encoding_info: bool | None = None,
    inclusive_end: bool | None = None,
    key: str | None = None,
    keys: Iterable[str] | None = None,
    limit: int | None = None,
    reduce: bool | None = None,
    skip: int | None = None,
    sort: bool | None = None,
    stable: bool | None = None,
    startkey: Any | None = None,
    startkey_docid: str | None = None,
    update: str | None = None,
    update_seq: bool | None = None,
) -> ViewResult:
    """
    See `AsyncDatabase.view`.
    """
    return await super().view(
        ddoc=ddoc,
        view=view,
        partition=self.partition_id,
        conflicts=conflicts,
        descending=descending,
        endkey=endkey,
        endkey_docid=endkey_docid,
        group=group,
        group_level=group_level,
        include_docs=include_docs,
        attachments=attachments,
        att_encoding_info=att_encoding_info,
        inclusive_end=inclusive_end,
        key=key,
        keys=keys,
        limit=limit,
        reduce=reduce,
        skip=skip,
        sort=sort,
        stable=stable,
        startkey=startkey,
        startkey_docid=startkey_docid,
        update=update,
        update_seq=update_seq,
    )

See AsyncDatabase.view().

Inherited members

• AsyncDatabase:
  • aclose
  • changes
  • changes_stream
  • check
  • compact
  • delete_index
  • design_docs
  • explain
  • get_design
  • get_partition
  • indexes
  • purge
  • put_design
  • save_index
  • security
  • server
  • update_security
  • url

class AsyncServer (url: str,
*,
port: int | None = None,
user: str | None = None,
password: str | None = None,
disable_ssl_verification: bool = False,
auth_method: str | None = None,
timeout: int | None = 300,
session: httpx.AsyncClient | None = None)

Expand source code

class AsyncServer(AsyncBase):
    """
    Async CouchDB server client. Mirrors `Server` with `async def` methods throughout.

    Use as an async context manager (recommended) or manage the lifecycle manually
    via `await client.aclose()`.

    Note: `__getitem__` is not supported on async classes — Python does not allow
    `__getitem__` to be a coroutine. Use `await client.get(name)` instead.

    Examples
    --------
    >>> async with AsyncServer("http://user:password@127.0.0.1:5984") as client:
    ...     print(await client.up())
    True
    """

    def __init__(
        self,
        url: str,
        *,
        port: int | None = None,
        user: str | None = None,
        password: str | None = None,
        disable_ssl_verification: bool = False,
        auth_method: str | None = None,
        timeout: int | None = DEFAULT_TIMEOUT,
        session: httpx.AsyncClient | None = None,
    ) -> None:
        """

        Parameters
        ----------
        url : str
            The url of the CouchDB server formatted as `scheme://user:password@host:port`. For example:

                "http://user:password@127.0.0.1:5984"
                "https://couchdb.example.com"
        port : int
            The port of the CouchDB server. Can also be supplied via the url.
        user : str
            The CouchDB admin username. Can also be supplied via the url.
        password : str
            The CouchDB admin password. Can also be supplied via the url.
        disable_ssl_verification : bool
            Controls whether to verify the server's TLS certificate. Set to `True` when connecting
            to a server with self-signed TLS certificates. Default `False`.
        auth_method : str
            Authentication method. Choices are `cookie` or `basic`.
            Default is `couchdb3.utils.DEFAULT_AUTH_METHOD`.
        timeout : int
            The default timeout for requests. Default c.f. `couchdb3.utils.DEFAULT_TIMEOUT`.
        session : httpx.AsyncClient
            A specific async client to use. Optional — if not provided, a new client will be
            initialized.
        """
        super().__init__(
            url=url,
            port=port,
            user=user,
            password=password,
            disable_ssl_verification=disable_ssl_verification,
            auth_method=auth_method,
            timeout=timeout,
            session=session,
        )

    def __repr__(self) -> str:
        """
        Returns
        -------
        str
        """
        return f"{super().__repr__()}: {self.url}"

    async def active_tasks(self) -> list[dict]:
        """
        List of running tasks, including the task type, name, status and process ID.

        Returns
        -------
        list[dict]
        """
        return (await self._get(resource="_active_tasks")).json()

    async def check_user(self, username: str, password: str) -> bool:
        """
        Checks the username/password combination by creating a temporary `AsyncServer`
        instance and performing a `check` request.

        Parameters
        ----------
        username : str
            The CouchDB user's name.
        password : str
            The CouchDB user's password.

        Returns
        -------
        bool : A boolean indicating if the username/password combination is valid.
        """
        async with AsyncServer(url=self.url, user=username, password=password) as client:
            return await client.check()

    async def save_user(
        self,
        name: str,
        *,
        user_id: str | None = None,
        derived_key: str | None = None,
        roles: list[str] | None = None,
        password: str | None = None,
        password_sha: str | None = None,
        password_scheme: str | None = None,
        salt: str | None = None,
        iterations: int | None = None,
        rev: str | None = None,
    ) -> tuple[bool, str, str]:
        """
        Create or update a user. In case of a `ConflictError`, a `HEAD` request to
        `/_users/<user_id>` will be sent to obtain the latest revision.

        Parameters
        ----------
        name : str
            User's name aka login. Immutable — you cannot rename an existing user.
        user_id : str
            The user's login with the special prefix `org.couchdb.user:`.
        derived_key : str
            PBKDF2 key derived from salt/iterations.
        roles : list[str]
            List of user roles.
        password : str
            A plaintext password — will be replaced by hashed fields before storage.
        password_sha : str
            Hashed password with salt. Used for `simple` password_scheme.
        password_scheme : str
            Password hashing scheme. May be `simple` or `pbkdf2`.
        salt : str
            Hash salt.
        iterations : int
            Number of iterations to derive key, used for `pbkdf2` password_scheme.
        rev : str
            The user's current revision. Needed when updating an existing user.

        Returns
        -------
        tuple[bool, str, str] : (success, user_id, revision)
        """
        if user_id and validate_user_id(user_id=user_id) is False:
            raise UserIDComplianceError(
                "User ID does not comply with the CouchDB requirements. "
                "See https://docs.couchdb.org/en/main/intro/security.html#why-the-org-couchdb-user-prefix."
            )
        user_id = user_id or user_name_to_id(name)
        body = {
            "_id": user_id,
            "_rev": rev,
            "derived_key": derived_key,
            "name": name,
            "roles": roles or [],
            "password": password,
            "password_sha": password_sha,
            "password_scheme": password_scheme,
            "salt": salt,
            "iterations": iterations,
            "type": "user",
        }
        try:
            response = await self._put(resource=f"_users/{user_id}", body=body)
        except ConflictError:
            body.update({"_rev": await self.rev(f"_users/{user_id}")})
            response = await self._put(resource=f"_users/{user_id}", body=body)
        data = response.json()
        return data["ok"], data["id"], data["rev"]

    async def all_dbs(
        self,
        *,
        descending: bool = False,
        endkey: str | None = None,
        limit: int | None = None,
        skip: int = 0,
        startkey: str | None = None,
    ) -> list[str]:
        """
        Get all database names.

        Parameters
        ----------
        descending : bool
            Return the databases in descending order by key. Default `False`.
        endkey : str
            Stop returning databases when the specified key is reached. Default `None`.
        limit : int
            Limit the number of the returned databases. Default `None`.
        skip : int
            Skip this number of databases before starting to return results. Default `0`.
        startkey : str
            Return databases starting with the specified key. Default `None`.

        Returns
        -------
        list[str] : A list of database names.
        """
        return (
            await self._get(
                "_all_dbs",
                query_kwargs={
                    "descending": descending,
                    "endkey": endkey,
                    "limit": limit,
                    "skip": skip,
                    "startkey": startkey,
                },
            )
        ).json()

    async def create(
        self,
        name: str,
        q: int | None = None,
        n: int | None = None,
        partitioned: bool = False,
    ) -> AsyncDatabase:
        """
        Create a database.

        Parameters
        ----------
        name : str
            The database's name.
        q : int
            Shards. Default `None` (server default: `8`).
        n : int
            Replicas. Default `None` (server default: `3`).
        partitioned : bool
            Whether to create a partitioned database. Default `False`.

        Returns
        -------
        AsyncDatabase
        """
        await self._put(resource=name, query_kwargs={"q": q, "n": n, "partitioned": partitioned})
        return await self.get(name=name)

    async def dbs_info(self, keys: list[str]) -> list[dict]:
        """
        Returns information about a list of specified databases.

        Parameters
        ----------
        keys : list[str]
            List of database names to be requested.

        Returns
        -------
        list[dict]
        """
        return (await self._post(resource="_dbs_info", body={"keys": keys})).json()

    async def get(self, name: str, check: bool = False) -> AsyncDatabase:
        """
        Get a database by name.

        Parameters
        ----------
        name : str
            The name of the database.
        check : bool
            If `True`, raise an exception if the database cannot be found. Default `False`.

        Returns
        -------
        AsyncDatabase
        """
        db = AsyncDatabase(
            name=name,
            url=self.url,
            user=self._user,
            password=self._password,
            disable_ssl_verification=self.disable_ssl_verification,
            auth_method=self.auth_method,
            session=self.session,  # shared — child sets _owns_session=False
            _server=self,
        )
        try:
            await db._head()
        except (NotFoundError, httpx.RequestError):
            if check is True:
                raise
        except CouchDBError:
            raise
        return db

    async def delete(self, resource: str | None = None) -> bool:
        """
        Delete a database.

        Parameters
        ----------
        resource : str
            The database's name.

        Returns
        -------
        bool: `True` upon successful deletion.
        """
        await self._delete(resource=resource)
        return True

    async def has_db(self, name: str) -> bool:
        """
        Check if the server contains a database with the given name.

        Note:
        The async client cannot support the `name in server` syntax (Python does not allow
        `__contains__` to be a coroutine). Use `await client.has_db(name)` instead.

        Parameters
        ----------
        name : str
            The database's name.

        Returns
        -------
        bool : `True` if the database exists, otherwise `False`.
        """
        try:
            await self._head(resource=name)
            return True
        except CouchDBError:
            return False

    async def replicate(
        self,
        source: dict | str,
        target: dict | str,
        replication_id: str | None = None,
        cancel: bool | None = None,
        continuous: bool | None = None,
        create_target: bool | None = None,
        create_target_params: dict | None = None,
        doc_ids: list[str] | None = None,
        filter_func: str | None = None,
        selector: dict | None = None,
        source_proxy: str | None = None,
        target_proxy: str | None = None,
    ) -> dict:
        """
        Request, configure, or stop, a replication operation. For more info, please refer to
        [the official documentation](https://docs.couchdb.org/en/main/api/server/common.html#replicate).

        Parameters
        ----------
        source : dict | str
            Fully qualified source database URL or an object with URL and headers.
        target : dict | str
            Fully qualified target database URL or an object with URL and headers.
        replication_id : str
            Deprecated. Ignored for one-shot replication (the `_replicate` endpoint does not accept
            a replication document ID).
        cancel : bool
            Cancels the replication.
        continuous : bool
            Configure the replication to be continuous.
        create_target : bool
            Creates the target database.
        create_target_params : dict
            Parameters for creating the target database (`q`, `n`).
        doc_ids : list[str]
            Document IDs to synchronize. Mutually exclusive with `filter_func` and `selector`.
        filter_func : str
            Name of a filter function. Mutually exclusive with `doc_ids` and `selector`.
        selector : dict
            Selector to filter documents. Mutually exclusive with `doc_ids` and `filter_func`.
        source_proxy : str
            Proxy for replication from source (`http` or `socks5`).
        target_proxy : str
            Proxy for replication to target (`http` or `socks5`).

        Returns
        -------
        dict
        """
        if (source_proxy and validate_proxy(source_proxy) is False) or (
            target_proxy and validate_proxy(target_proxy) is False
        ):
            raise ProxySchemeComplianceError("Proxy has invalid scheme.")
        if replication_id is not None:
            warnings.warn(
                "`replication_id` is deprecated and has no effect on the one-shot "
                "`_replicate` endpoint; it is ignored.",
                DeprecationWarning,
                stacklevel=2,
            )
        if sum(bool(_) for _ in [doc_ids, filter_func, selector]) > 1:
            raise CouchDBError(
                'Arguments "doc_ids", "filter_func" and "selector" are mutually exclusive.'
            )
        return (
            await self._post(
                resource="_replicate",
                body=rm_nones_from_dict(
                    {
                        "source": source,
                        "target": target,
                        "cancel": cancel,
                        "continuous": continuous,
                        "create_target": create_target,
                        "create_target_params": create_target_params,
                        "doc_ids": doc_ids,
                        "filter": filter_func,
                        "selector": selector,
                        "source_proxy": source_proxy,
                        "target_proxy": target_proxy,
                    }
                ),
            )
        ).json()

    async def membership(self) -> dict:
        """
        Displays the nodes that are part of the cluster.

        Returns
        -------
        dict : A dictionary with the following keys.

          - ``all_nodes`` (`list[str]`) — all nodes this node knows about
          - ``cluster_nodes`` (`list[str]`) — nodes that are part of the cluster
        """
        return (await self._get(resource="_membership")).json()

    async def cluster_setup(
        self,
        *,
        ensure_dbs_exist: list[str] | None = None,
    ) -> dict:
        """
        Returns the status of the node or cluster, per the cluster setup wizard.

        Parameters
        ----------
        ensure_dbs_exist : list[str]
            List of system databases to ensure exist on the node/cluster.
            Defaults to ``["_users", "_replicator"]``.

        Returns
        -------
        dict : A dictionary with a single key ``state`` whose value is one of
        ``'cluster_disabled'``, ``'single_node_disabled'``, ``'single_node_enabled'``,
        ``'cluster_enabled'``, or ``'cluster_finished'``.
        """
        return (
            await self._get(
                resource="_cluster_setup",
                query_kwargs={"ensure_dbs_exist": ensure_dbs_exist},
            )
        ).json()

    async def setup_cluster(
        self,
        action: str,
        *,
        bind_address: str | None = None,
        username: str | None = None,
        password: str | None = None,
        port: int | None = None,
        node_count: int | None = None,
        remote_node: str | None = None,
        remote_current_user: str | None = None,
        remote_current_password: str | None = None,
        host: str | None = None,
        ensure_dbs_exist: list[str] | None = None,
    ) -> dict:
        """
        Configure a node as a single (standalone) node, as part of a cluster, or finalise
        a cluster. This is a **destructive** operation — do not run against a shared or
        production CouchDB instance during testing.

        Parameters
        ----------
        action : str
            One of ``'enable_single_node'``, ``'enable_cluster'``, ``'add_node'``, or
            ``'finish_cluster'``.
        bind_address : str
            IP address to bind the current node. Use ``'0.0.0.0'`` to bind all interfaces.
            (``enable_cluster`` and ``enable_single_node`` only)
        username : str
            Server-level administrator username to create, or the remote server's
            administrator username (``add_node``).
        password : str
            Server-level administrator password to create, or the remote server's password
            (``add_node``).
        port : int
            TCP port for this node (``enable_cluster`` / ``enable_single_node``) or the
            remote node's port (``add_node``).
        node_count : int
            Total number of nodes to join into the cluster. Determines ``n`` (max 3).
            (``enable_cluster`` only)
        remote_node : str
            IP address of the remote node. (``enable_cluster`` only)
        remote_current_user : str
            Username of the admin on the remote node. (``enable_cluster`` only)
        remote_current_password : str
            Password of the admin on the remote node. (``enable_cluster`` only)
        host : str
            Remote node IP to add to the cluster. (``add_node`` only)
        ensure_dbs_exist : list[str]
            List of system databases to ensure exist. Defaults to
            ``["_users", "_replicator"]``.

        Returns
        -------
        dict : ``{"ok": true}`` on success.
        """
        return (
            await self._post(
                resource="_cluster_setup",
                body=rm_nones_from_dict(
                    {
                        "action": action,
                        "bind_address": bind_address,
                        "username": username,
                        "password": password,
                        "port": port,
                        "node_count": node_count,
                        "remote_node": remote_node,
                        "remote_current_user": remote_current_user,
                        "remote_current_password": remote_current_password,
                        "host": host,
                        "ensure_dbs_exist": ensure_dbs_exist,
                    }
                ),
            )
        ).json()

    async def node_config(
        self,
        node: str = "_local",
        section: str | None = None,
        key: str | None = None,
    ) -> dict | str:
        """
        Returns CouchDB node configuration.

        - No `section` / `key` → full configuration tree (`dict`)
        - `section` only → configuration section (`dict`)
        - `section` + `key` → single configuration value (`str` or primitive)

        The literal string ``'_local'`` (default) is an alias for the local node name.

        Parameters
        ----------
        node : str
            Node name. Default ``'_local'``.
        section : str
            Configuration section name (e.g. ``'log'``, ``'couchdb'``).
        key : str
            Configuration key within the section (e.g. ``'level'``).

        Returns
        -------
        dict | str
        """
        resource = f"_node/{node}/_config"
        if section:
            resource = f"{resource}/{section}"
            if key:
                resource = f"{resource}/{key}"
        return (await self._get(resource=resource)).json()

    async def set_node_config(
        self,
        section: str,
        key: str,
        value: str,
        node: str = "_local",
    ) -> str:
        """
        Updates a single configuration value on a node. Returns the **old** value.

        Parameters
        ----------
        section : str
            Configuration section name.
        key : str
            Configuration key name.
        value : str
            New value (must be a valid JSON string).
        node : str
            Node name. Default ``'_local'``.

        Returns
        -------
        str : The previous value of the configuration key.
        """
        return (
            await self._put(
                resource=f"_node/{node}/_config/{section}/{key}",
                body=value,
            )
        ).json()

    async def delete_node_config(
        self,
        section: str,
        key: str,
        node: str = "_local",
    ) -> str:
        """
        Deletes a single configuration value from a node. Returns the **old** value.

        Parameters
        ----------
        section : str
            Configuration section name.
        key : str
            Configuration key name.
        node : str
            Node name. Default ``'_local'``.

        Returns
        -------
        str : The deleted value.
        """
        return (
            await self._delete(
                resource=f"_node/{node}/_config/{section}/{key}",
            )
        ).json()

    async def reload_node_config(self, node: str = "_local") -> bool:
        """
        Reloads the configuration from disk. Flushes any in-memory configuration changes
        that have not been written to disk.

        Parameters
        ----------
        node : str
            Node name. Default ``'_local'``.

        Returns
        -------
        bool : ``True`` on success.
        """
        return (
            (
                await self._post(
                    resource=f"_node/{node}/_config/_reload",
                    body={},
                )
            )
            .json()
            .get("ok", False)
        )

    async def node_stats(self, node: str = "_local") -> dict:
        """
        Returns statistics for the specified node.

        Parameters
        ----------
        node : str
            Node name. Default ``'_local'``.

        Returns
        -------
        dict
        """
        return (await self._get(resource=f"_node/{node}/_stats")).json()

    async def node_system(self, node: str = "_local") -> dict:
        """
        Returns system-level statistics for the specified node.

        Parameters
        ----------
        node : str
            Node name. Default ``'_local'``.

        Returns
        -------
        dict
        """
        return (await self._get(resource=f"_node/{node}/_system")).json()

    async def up(self, raise_exception: bool = False) -> bool:
        """
        Check if the server is up.

        Parameters
        ----------
        raise_exception : bool
            If `True`, exceptions will be raised instead of returning `False`.

        Returns
        -------
        bool : `True` if the server is up.
        """
        try:
            response = await self._get(resource="_up")
            return "status" in response.json() and response.json()["status"] == "ok"
        except Exception:
            if raise_exception:
                raise
            return False

Async CouchDB server client. Mirrors Server with async def methods throughout.

Use as an async context manager (recommended) or manage the lifecycle manually via await client.aclose().

Note: __getitem__ is not supported on async classes — Python does not allow __getitem__ to be a coroutine. Use await client.get(name) instead.

Examples

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

Parameters

url : str

The url of the CouchDB server formatted as scheme://user:password@host:port. For example:

"http://user:password@127.0.0.1:5984"
"https://couchdb.example.com"

port : int

The port of the CouchDB server. Can also be supplied via the url.

user : str

The CouchDB admin username. Can also be supplied via the url.

password : str

The CouchDB admin password. Can also be supplied via the url.

disable_ssl_verification : bool

Controls whether to verify the server's TLS certificate. Set to True when connecting to a server with self-signed TLS certificates. Default False.

auth_method : str

Authentication method. Choices are cookie or basic. Default is DEFAULT_AUTH_METHOD.

timeout : int

The default timeout for requests. Default c.f. DEFAULT_TIMEOUT.

session : httpx.AsyncClient

A specific async client to use. Optional — if not provided, a new client will be initialized.

Ancestors

• AsyncBase

Methods

async def active_tasks(self) ‑> list[dict]

Expand source code

async def active_tasks(self) -> list[dict]:
    """
    List of running tasks, including the task type, name, status and process ID.

    Returns
    -------
    list[dict]
    """
    return (await self._get(resource="_active_tasks")).json()

List of running tasks, including the task type, name, status and process ID.

Returns

list[dict]

 

async def all_dbs(self,
*,
descending: bool = False,
endkey: str | None = None,
limit: int | None = None,
skip: int = 0,
startkey: str | None = None) ‑> list[str]

Expand source code

async def all_dbs(
    self,
    *,
    descending: bool = False,
    endkey: str | None = None,
    limit: int | None = None,
    skip: int = 0,
    startkey: str | None = None,
) -> list[str]:
    """
    Get all database names.

    Parameters
    ----------
    descending : bool
        Return the databases in descending order by key. Default `False`.
    endkey : str
        Stop returning databases when the specified key is reached. Default `None`.
    limit : int
        Limit the number of the returned databases. Default `None`.
    skip : int
        Skip this number of databases before starting to return results. Default `0`.
    startkey : str
        Return databases starting with the specified key. Default `None`.

    Returns
    -------
    list[str] : A list of database names.
    """
    return (
        await self._get(
            "_all_dbs",
            query_kwargs={
                "descending": descending,
                "endkey": endkey,
                "limit": limit,
                "skip": skip,
                "startkey": startkey,
            },
        )
    ).json()

Get all database names.

Parameters

descending : bool

Return the databases in descending order by key. Default False.

endkey : str

Stop returning databases when the specified key is reached. Default None.

limit : int

Limit the number of the returned databases. Default None.

skip : int

Skip this number of databases before starting to return results. Default 0.

startkey : str

Return databases starting with the specified key. Default None.

Returns

list[str] : A list of database names.

async def check_user(self, username: str, password: str) ‑> bool

Expand source code

async def check_user(self, username: str, password: str) -> bool:
    """
    Checks the username/password combination by creating a temporary `AsyncServer`
    instance and performing a `check` request.

    Parameters
    ----------
    username : str
        The CouchDB user's name.
    password : str
        The CouchDB user's password.

    Returns
    -------
    bool : A boolean indicating if the username/password combination is valid.
    """
    async with AsyncServer(url=self.url, user=username, password=password) as client:
        return await client.check()

Checks the username/password combination by creating a temporary AsyncServer instance and performing a check request.

Parameters

username : str

The CouchDB user's name.

password : str

The CouchDB user's password.

Returns

bool : A boolean indicating if the username/password combination is valid.

async def cluster_setup(self, *, ensure_dbs_exist: list[str] | None = None) ‑> dict

Expand source code

async def cluster_setup(
    self,
    *,
    ensure_dbs_exist: list[str] | None = None,
) -> dict:
    """
    Returns the status of the node or cluster, per the cluster setup wizard.

    Parameters
    ----------
    ensure_dbs_exist : list[str]
        List of system databases to ensure exist on the node/cluster.
        Defaults to ``["_users", "_replicator"]``.

    Returns
    -------
    dict : A dictionary with a single key ``state`` whose value is one of
    ``'cluster_disabled'``, ``'single_node_disabled'``, ``'single_node_enabled'``,
    ``'cluster_enabled'``, or ``'cluster_finished'``.
    """
    return (
        await self._get(
            resource="_cluster_setup",
            query_kwargs={"ensure_dbs_exist": ensure_dbs_exist},
        )
    ).json()

Returns the status of the node or cluster, per the cluster setup wizard.

Parameters

ensure_dbs_exist : list[str]

List of system databases to ensure exist on the node/cluster. Defaults to ["_users", "_replicator"].

Returns

dict : A dictionary with a single key state whose value is one of

 

'cluster_disabled', 'single_node_disabled', 'single_node_enabled', 'cluster_enabled', or 'cluster_finished'.

async def create(self,
name: str,
q: int | None = None,
n: int | None = None,
partitioned: bool = False) ‑> AsyncDatabase

Expand source code

async def create(
    self,
    name: str,
    q: int | None = None,
    n: int | None = None,
    partitioned: bool = False,
) -> AsyncDatabase:
    """
    Create a database.

    Parameters
    ----------
    name : str
        The database's name.
    q : int
        Shards. Default `None` (server default: `8`).
    n : int
        Replicas. Default `None` (server default: `3`).
    partitioned : bool
        Whether to create a partitioned database. Default `False`.

    Returns
    -------
    AsyncDatabase
    """
    await self._put(resource=name, query_kwargs={"q": q, "n": n, "partitioned": partitioned})
    return await self.get(name=name)

Create a database.

Parameters

name : str

The database's name.

q : int

Shards. Default None (server default: 8).

n : int

Replicas. Default None (server default: 3).

partitioned : bool

Whether to create a partitioned database. Default False.

Returns

AsyncDatabase

 

async def dbs_info(self, keys: list[str]) ‑> list[dict]

Expand source code

async def dbs_info(self, keys: list[str]) -> list[dict]:
    """
    Returns information about a list of specified databases.

    Parameters
    ----------
    keys : list[str]
        List of database names to be requested.

    Returns
    -------
    list[dict]
    """
    return (await self._post(resource="_dbs_info", body={"keys": keys})).json()

Returns information about a list of specified databases.

Parameters

keys : list[str]

List of database names to be requested.

Returns

list[dict]

 

async def delete(self, resource: str | None = None) ‑> bool

Expand source code

async def delete(self, resource: str | None = None) -> bool:
    """
    Delete a database.

    Parameters
    ----------
    resource : str
        The database's name.

    Returns
    -------
    bool: `True` upon successful deletion.
    """
    await self._delete(resource=resource)
    return True

Delete a database.

Parameters

resource : str

The database's name.

Returns

bool: True upon successful deletion.

async def delete_node_config(self, section: str, key: str, node: str = '_local') ‑> str

Expand source code

async def delete_node_config(
    self,
    section: str,
    key: str,
    node: str = "_local",
) -> str:
    """
    Deletes a single configuration value from a node. Returns the **old** value.

    Parameters
    ----------
    section : str
        Configuration section name.
    key : str
        Configuration key name.
    node : str
        Node name. Default ``'_local'``.

    Returns
    -------
    str : The deleted value.
    """
    return (
        await self._delete(
            resource=f"_node/{node}/_config/{section}/{key}",
        )
    ).json()

Deletes a single configuration value from a node. Returns the old value.

Parameters

section : str

Configuration section name.

key : str

Configuration key name.

node : str

Node name. Default '_local'.

Returns

str : The deleted value.

async def get(self, name: str, check: bool = False) ‑> AsyncDatabase

Expand source code

async def get(self, name: str, check: bool = False) -> AsyncDatabase:
    """
    Get a database by name.

    Parameters
    ----------
    name : str
        The name of the database.
    check : bool
        If `True`, raise an exception if the database cannot be found. Default `False`.

    Returns
    -------
    AsyncDatabase
    """
    db = AsyncDatabase(
        name=name,
        url=self.url,
        user=self._user,
        password=self._password,
        disable_ssl_verification=self.disable_ssl_verification,
        auth_method=self.auth_method,
        session=self.session,  # shared — child sets _owns_session=False
        _server=self,
    )
    try:
        await db._head()
    except (NotFoundError, httpx.RequestError):
        if check is True:
            raise
    except CouchDBError:
        raise
    return db

Get a database by name.

Parameters

name : str

The name of the database.

check : bool

If True, raise an exception if the database cannot be found. Default False.

Returns

AsyncDatabase

 

async def has_db(self, name: str) ‑> bool

Expand source code

async def has_db(self, name: str) -> bool:
    """
    Check if the server contains a database with the given name.

    Note:
    The async client cannot support the `name in server` syntax (Python does not allow
    `__contains__` to be a coroutine). Use `await client.has_db(name)` instead.

    Parameters
    ----------
    name : str
        The database's name.

    Returns
    -------
    bool : `True` if the database exists, otherwise `False`.
    """
    try:
        await self._head(resource=name)
        return True
    except CouchDBError:
        return False

Check if the server contains a database with the given name.

Note: The async client cannot support the name in server syntax (Python does not allow __contains__ to be a coroutine). Use await client.has_db(name) instead.

Parameters

name : str

The database's name.

Returns

bool : True if the database exists, otherwise False.

async def membership(self) ‑> dict

Expand source code

async def membership(self) -> dict:
    """
    Displays the nodes that are part of the cluster.

    Returns
    -------
    dict : A dictionary with the following keys.

      - ``all_nodes`` (`list[str]`) — all nodes this node knows about
      - ``cluster_nodes`` (`list[str]`) — nodes that are part of the cluster
    """
    return (await self._get(resource="_membership")).json()

Displays the nodes that are part of the cluster.

Returns

dict : A dictionary with the following keys.

• all_nodes (list[str]) — all nodes this node knows about
• cluster_nodes (list[str]) — nodes that are part of the cluster

async def node_config(self, node: str = '_local', section: str | None = None, key: str | None = None) ‑> dict | str

Expand source code

async def node_config(
    self,
    node: str = "_local",
    section: str | None = None,
    key: str | None = None,
) -> dict | str:
    """
    Returns CouchDB node configuration.

    - No `section` / `key` → full configuration tree (`dict`)
    - `section` only → configuration section (`dict`)
    - `section` + `key` → single configuration value (`str` or primitive)

    The literal string ``'_local'`` (default) is an alias for the local node name.

    Parameters
    ----------
    node : str
        Node name. Default ``'_local'``.
    section : str
        Configuration section name (e.g. ``'log'``, ``'couchdb'``).
    key : str
        Configuration key within the section (e.g. ``'level'``).

    Returns
    -------
    dict | str
    """
    resource = f"_node/{node}/_config"
    if section:
        resource = f"{resource}/{section}"
        if key:
            resource = f"{resource}/{key}"
    return (await self._get(resource=resource)).json()

Returns CouchDB node configuration.

• No section / key → full configuration tree (dict)
• section only → configuration section (dict)
• section + key → single configuration value (str or primitive)

The literal string '_local' (default) is an alias for the local node name.

Parameters

node : str

Node name. Default '_local'.

section : str

Configuration section name (e.g. 'log', 'couchdb').

key : str

Configuration key within the section (e.g. 'level').

Returns

dict | str

 

async def node_stats(self, node: str = '_local') ‑> dict

Expand source code

async def node_stats(self, node: str = "_local") -> dict:
    """
    Returns statistics for the specified node.

    Parameters
    ----------
    node : str
        Node name. Default ``'_local'``.

    Returns
    -------
    dict
    """
    return (await self._get(resource=f"_node/{node}/_stats")).json()

Returns statistics for the specified node.

Parameters

node : str

Node name. Default '_local'.

Returns

dict

 

async def node_system(self, node: str = '_local') ‑> dict

Expand source code

async def node_system(self, node: str = "_local") -> dict:
    """
    Returns system-level statistics for the specified node.

    Parameters
    ----------
    node : str
        Node name. Default ``'_local'``.

    Returns
    -------
    dict
    """
    return (await self._get(resource=f"_node/{node}/_system")).json()

Returns system-level statistics for the specified node.

Parameters

node : str

Node name. Default '_local'.

Returns

dict

 

async def reload_node_config(self, node: str = '_local') ‑> bool

Expand source code

async def reload_node_config(self, node: str = "_local") -> bool:
    """
    Reloads the configuration from disk. Flushes any in-memory configuration changes
    that have not been written to disk.

    Parameters
    ----------
    node : str
        Node name. Default ``'_local'``.

    Returns
    -------
    bool : ``True`` on success.
    """
    return (
        (
            await self._post(
                resource=f"_node/{node}/_config/_reload",
                body={},
            )
        )
        .json()
        .get("ok", False)
    )

Reloads the configuration from disk. Flushes any in-memory configuration changes that have not been written to disk.

Parameters

node : str

Node name. Default '_local'.

Returns

bool : True on success.

async def replicate(self,
source: dict | str,
target: dict | str,
replication_id: str | None = None,
cancel: bool | None = None,
continuous: bool | None = None,
create_target: bool | None = None,
create_target_params: dict | None = None,
doc_ids: list[str] | None = None,
filter_func: str | None = None,
selector: dict | None = None,
source_proxy: str | None = None,
target_proxy: str | None = None) ‑> dict

Expand source code

async def replicate(
    self,
    source: dict | str,
    target: dict | str,
    replication_id: str | None = None,
    cancel: bool | None = None,
    continuous: bool | None = None,
    create_target: bool | None = None,
    create_target_params: dict | None = None,
    doc_ids: list[str] | None = None,
    filter_func: str | None = None,
    selector: dict | None = None,
    source_proxy: str | None = None,
    target_proxy: str | None = None,
) -> dict:
    """
    Request, configure, or stop, a replication operation. For more info, please refer to
    [the official documentation](https://docs.couchdb.org/en/main/api/server/common.html#replicate).

    Parameters
    ----------
    source : dict | str
        Fully qualified source database URL or an object with URL and headers.
    target : dict | str
        Fully qualified target database URL or an object with URL and headers.
    replication_id : str
        Deprecated. Ignored for one-shot replication (the `_replicate` endpoint does not accept
        a replication document ID).
    cancel : bool
        Cancels the replication.
    continuous : bool
        Configure the replication to be continuous.
    create_target : bool
        Creates the target database.
    create_target_params : dict
        Parameters for creating the target database (`q`, `n`).
    doc_ids : list[str]
        Document IDs to synchronize. Mutually exclusive with `filter_func` and `selector`.
    filter_func : str
        Name of a filter function. Mutually exclusive with `doc_ids` and `selector`.
    selector : dict
        Selector to filter documents. Mutually exclusive with `doc_ids` and `filter_func`.
    source_proxy : str
        Proxy for replication from source (`http` or `socks5`).
    target_proxy : str
        Proxy for replication to target (`http` or `socks5`).

    Returns
    -------
    dict
    """
    if (source_proxy and validate_proxy(source_proxy) is False) or (
        target_proxy and validate_proxy(target_proxy) is False
    ):
        raise ProxySchemeComplianceError("Proxy has invalid scheme.")
    if replication_id is not None:
        warnings.warn(
            "`replication_id` is deprecated and has no effect on the one-shot "
            "`_replicate` endpoint; it is ignored.",
            DeprecationWarning,
            stacklevel=2,
        )
    if sum(bool(_) for _ in [doc_ids, filter_func, selector]) > 1:
        raise CouchDBError(
            'Arguments "doc_ids", "filter_func" and "selector" are mutually exclusive.'
        )
    return (
        await self._post(
            resource="_replicate",
            body=rm_nones_from_dict(
                {
                    "source": source,
                    "target": target,
                    "cancel": cancel,
                    "continuous": continuous,
                    "create_target": create_target,
                    "create_target_params": create_target_params,
                    "doc_ids": doc_ids,
                    "filter": filter_func,
                    "selector": selector,
                    "source_proxy": source_proxy,
                    "target_proxy": target_proxy,
                }
            ),
        )
    ).json()

Request, configure, or stop, a replication operation. For more info, please refer to the official documentation.

Parameters

source : dict | str

Fully qualified source database URL or an object with URL and headers.

target : dict | str

Fully qualified target database URL or an object with URL and headers.

replication_id : str

Deprecated. Ignored for one-shot replication (the _replicate endpoint does not accept a replication document ID).

cancel : bool

Cancels the replication.

continuous : bool

Configure the replication to be continuous.

create_target : bool

Creates the target database.

create_target_params : dict

Parameters for creating the target database (q, n).

doc_ids : list[str]

Document IDs to synchronize. Mutually exclusive with filter_func and selector.

filter_func : str

Name of a filter function. Mutually exclusive with doc_ids and selector.

selector : dict

Selector to filter documents. Mutually exclusive with doc_ids and filter_func.

source_proxy : str

Proxy for replication from source (http or socks5).

target_proxy : str

Proxy for replication to target (http or socks5).

Returns

dict

 

async def save_user(self,
name: str,
*,
user_id: str | None = None,
derived_key: str | None = None,
roles: list[str] | None = None,
password: str | None = None,
password_sha: str | None = None,
password_scheme: str | None = None,
salt: str | None = None,
iterations: int | None = None,
rev: str | None = None) ‑> tuple[bool, str, str]

Expand source code

async def save_user(
    self,
    name: str,
    *,
    user_id: str | None = None,
    derived_key: str | None = None,
    roles: list[str] | None = None,
    password: str | None = None,
    password_sha: str | None = None,
    password_scheme: str | None = None,
    salt: str | None = None,
    iterations: int | None = None,
    rev: str | None = None,
) -> tuple[bool, str, str]:
    """
    Create or update a user. In case of a `ConflictError`, a `HEAD` request to
    `/_users/<user_id>` will be sent to obtain the latest revision.

    Parameters
    ----------
    name : str
        User's name aka login. Immutable — you cannot rename an existing user.
    user_id : str
        The user's login with the special prefix `org.couchdb.user:`.
    derived_key : str
        PBKDF2 key derived from salt/iterations.
    roles : list[str]
        List of user roles.
    password : str
        A plaintext password — will be replaced by hashed fields before storage.
    password_sha : str
        Hashed password with salt. Used for `simple` password_scheme.
    password_scheme : str
        Password hashing scheme. May be `simple` or `pbkdf2`.
    salt : str
        Hash salt.
    iterations : int
        Number of iterations to derive key, used for `pbkdf2` password_scheme.
    rev : str
        The user's current revision. Needed when updating an existing user.

    Returns
    -------
    tuple[bool, str, str] : (success, user_id, revision)
    """
    if user_id and validate_user_id(user_id=user_id) is False:
        raise UserIDComplianceError(
            "User ID does not comply with the CouchDB requirements. "
            "See https://docs.couchdb.org/en/main/intro/security.html#why-the-org-couchdb-user-prefix."
        )
    user_id = user_id or user_name_to_id(name)
    body = {
        "_id": user_id,
        "_rev": rev,
        "derived_key": derived_key,
        "name": name,
        "roles": roles or [],
        "password": password,
        "password_sha": password_sha,
        "password_scheme": password_scheme,
        "salt": salt,
        "iterations": iterations,
        "type": "user",
    }
    try:
        response = await self._put(resource=f"_users/{user_id}", body=body)
    except ConflictError:
        body.update({"_rev": await self.rev(f"_users/{user_id}")})
        response = await self._put(resource=f"_users/{user_id}", body=body)
    data = response.json()
    return data["ok"], data["id"], data["rev"]

Create or update a user. In case of a ConflictError, a HEAD request to /_users/<user_id> will be sent to obtain the latest revision.

Parameters

name : str

User's name aka login. Immutable — you cannot rename an existing user.

user_id : str

The user's login with the special prefix org.couchdb.user:.

derived_key : str

PBKDF2 key derived from salt/iterations.

roles : list[str]

List of user roles.

password : str

A plaintext password — will be replaced by hashed fields before storage.

password_sha : str

Hashed password with salt. Used for simple password_scheme.

password_scheme : str

Password hashing scheme. May be simple or pbkdf2.

salt : str

Hash salt.

iterations : int

Number of iterations to derive key, used for pbkdf2 password_scheme.

rev : str

The user's current revision. Needed when updating an existing user.

Returns

tuple[bool, str, str] : (success, user_id, revision)

 

async def set_node_config(self, section: str, key: str, value: str, node: str = '_local') ‑> str

Expand source code

async def set_node_config(
    self,
    section: str,
    key: str,
    value: str,
    node: str = "_local",
) -> str:
    """
    Updates a single configuration value on a node. Returns the **old** value.

    Parameters
    ----------
    section : str
        Configuration section name.
    key : str
        Configuration key name.
    value : str
        New value (must be a valid JSON string).
    node : str
        Node name. Default ``'_local'``.

    Returns
    -------
    str : The previous value of the configuration key.
    """
    return (
        await self._put(
            resource=f"_node/{node}/_config/{section}/{key}",
            body=value,
        )
    ).json()

Updates a single configuration value on a node. Returns the old value.

Parameters

section : str

Configuration section name.

key : str

Configuration key name.

value : str

New value (must be a valid JSON string).

node : str

Node name. Default '_local'.

Returns

str : The previous value of the configuration key.

async def setup_cluster(self,
action: str,
*,
bind_address: str | None = None,
username: str | None = None,
password: str | None = None,
port: int | None = None,
node_count: int | None = None,
remote_node: str | None = None,
remote_current_user: str | None = None,
remote_current_password: str | None = None,
host: str | None = None,
ensure_dbs_exist: list[str] | None = None) ‑> dict

Expand source code

async def setup_cluster(
    self,
    action: str,
    *,
    bind_address: str | None = None,
    username: str | None = None,
    password: str | None = None,
    port: int | None = None,
    node_count: int | None = None,
    remote_node: str | None = None,
    remote_current_user: str | None = None,
    remote_current_password: str | None = None,
    host: str | None = None,
    ensure_dbs_exist: list[str] | None = None,
) -> dict:
    """
    Configure a node as a single (standalone) node, as part of a cluster, or finalise
    a cluster. This is a **destructive** operation — do not run against a shared or
    production CouchDB instance during testing.

    Parameters
    ----------
    action : str
        One of ``'enable_single_node'``, ``'enable_cluster'``, ``'add_node'``, or
        ``'finish_cluster'``.
    bind_address : str
        IP address to bind the current node. Use ``'0.0.0.0'`` to bind all interfaces.
        (``enable_cluster`` and ``enable_single_node`` only)
    username : str
        Server-level administrator username to create, or the remote server's
        administrator username (``add_node``).
    password : str
        Server-level administrator password to create, or the remote server's password
        (``add_node``).
    port : int
        TCP port for this node (``enable_cluster`` / ``enable_single_node``) or the
        remote node's port (``add_node``).
    node_count : int
        Total number of nodes to join into the cluster. Determines ``n`` (max 3).
        (``enable_cluster`` only)
    remote_node : str
        IP address of the remote node. (``enable_cluster`` only)
    remote_current_user : str
        Username of the admin on the remote node. (``enable_cluster`` only)
    remote_current_password : str
        Password of the admin on the remote node. (``enable_cluster`` only)
    host : str
        Remote node IP to add to the cluster. (``add_node`` only)
    ensure_dbs_exist : list[str]
        List of system databases to ensure exist. Defaults to
        ``["_users", "_replicator"]``.

    Returns
    -------
    dict : ``{"ok": true}`` on success.
    """
    return (
        await self._post(
            resource="_cluster_setup",
            body=rm_nones_from_dict(
                {
                    "action": action,
                    "bind_address": bind_address,
                    "username": username,
                    "password": password,
                    "port": port,
                    "node_count": node_count,
                    "remote_node": remote_node,
                    "remote_current_user": remote_current_user,
                    "remote_current_password": remote_current_password,
                    "host": host,
                    "ensure_dbs_exist": ensure_dbs_exist,
                }
            ),
        )
    ).json()

Configure a node as a single (standalone) node, as part of a cluster, or finalise a cluster. This is a destructive operation — do not run against a shared or production CouchDB instance during testing.

Parameters

action : str

One of 'enable_single_node', 'enable_cluster', 'add_node', or 'finish_cluster'.

bind_address : str

IP address to bind the current node. Use '0.0.0.0' to bind all interfaces. (enable_cluster and enable_single_node only)

username : str

Server-level administrator username to create, or the remote server's administrator username (add_node).

password : str

Server-level administrator password to create, or the remote server's password (add_node).

port : int

TCP port for this node (enable_cluster / enable_single_node) or the remote node's port (add_node).

node_count : int

Total number of nodes to join into the cluster. Determines n (max 3). (enable_cluster only)

remote_node : str

IP address of the remote node. (enable_cluster only)

remote_current_user : str

Username of the admin on the remote node. (enable_cluster only)

remote_current_password : str

Password of the admin on the remote node. (enable_cluster only)

host : str

Remote node IP to add to the cluster. (add_node only)

ensure_dbs_exist : list[str]

List of system databases to ensure exist. Defaults to ["_users", "_replicator"].

Returns

dict : {"ok": true} on success.

async def up(self, raise_exception: bool = False) ‑> bool

Expand source code

async def up(self, raise_exception: bool = False) -> bool:
    """
    Check if the server is up.

    Parameters
    ----------
    raise_exception : bool
        If `True`, exceptions will be raised instead of returning `False`.

    Returns
    -------
    bool : `True` if the server is up.
    """
    try:
        response = await self._get(resource="_up")
        return "status" in response.json() and response.json()["status"] == "ok"
    except Exception:
        if raise_exception:
            raise
        return False

Check if the server is up.

Parameters

raise_exception : bool

If True, exceptions will be raised instead of returning False.

Returns

bool : True if the server is up.

Inherited members

• AsyncBase:
  • aclose
  • check
  • info
  • rev
  • url