Module couchdb3.sync

Synchronous CouchDB client subpackage.

from couchdb3.sync import Server, Database, Partition

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

Sub-modules

couchdb3.sync.base

couchdb3.sync.database

couchdb3.sync.server

Classes

class Database (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.Client | None = None)

Expand source code

class Database(Base):
    """
    Abstract Couchdb database
    """

    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.Client | 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.Client
            A specific client to use. Optional - if not provided, a new client will be initialized.
        _server : Server
            The owning `Server` instance. Set internally by `Server.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 a `Database` 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 `Server` instance this database was obtained from, or `None` if the database
        was constructed directly (i.e. not via `Server.get()`).

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

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

    def __getitem__(self, item) -> Document:
        return self.get(docid=item, check=True)

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

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

    def all_docs(
        self,
        partition: str | None = None,
        keys: Iterable[str] | None = None,
        **kwargs,
    ) -> ViewResult:
        """
        Executes the built-in _all_docs view, returning all the 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 in the argument. Default is `None`.
        kwargs
            Further `couchdb3.sync.Database.view` parameters.

        Returns
        -------
        ViewResult

        Examples
        --------
        >>> db.all_docs()
        >>> db.all_docs(include_docs=True)
        >>> for row in db.all_docs(include_docs=True).rows:
        ...     print(row.id, row.doc)
        >>> db.all_docs(keys=["id-1", "id-2"], include_docs=True)
        """
        return self.view(
            f"_partition/{partition}/_all_docs" if partition else "_all_docs",
            keys=keys,
            **kwargs,
        )

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

    def bulk_docs(self, docs: list[dict | Document], new_edits: bool = True) -> list[dict]:
        """
        The bulk document API allows you to create and update multiple documents at the same time within a single
        request. The basic operation is similar to creating or updating a single document, except that you batch the
        document structure and information.

        When creating new documents the document ID (`_id`) is optional.

        For updating existing documents, you must provide the document ID, revision information (`_rev`), and new
        document values.

        In case of batch deleting documents all fields as document ID, revision information and deletion status
        (`_deleted`) are required.

        Parameters
        ----------
        docs : List[Union[Dict, Document]]
             List of documents objects
        new_edits : bool
            If `False`, prevents the database from assigning them new revision IDs. Default `True`.

        Returns
        -------
        List[Dict] : A list of dictionaries containing the following keys.

          - `id` the document's id
          - `ok` operation status
          - `rev` the document's revision
        """
        return self._post(resource="_bulk_docs", body={"docs": docs, "new_edits": new_edits}).json()

    def bulk_get(
        self,
        docs: list[dict | Document],
        revs: bool = False,
    ) -> list[dict]:
        """
        This method can be called to query several documents in bulk. It is well suited for fetching a specific
        revision of documents, as replicators do for example, or for getting revision history.

        Parameters
        ----------
        docs : List[Union[Dict, Document]]
            List of document objects, with `id`, and optionally `rev` and `atts_since`.
        revs : bool
             Give the revisions history.

        Returns
        -------
        List[Dict] : An array of results for each requested document/rev pair.

          - `id` key lists the requested
          document ID,
          - `docs` contains a single-item array of objects, each of which has either an `error` key and value describing
          the error, or `ok` key and associated value of the requested document, with the additional _revisions property
          that lists the parent revisions if `revs=true`.

        Examples
        --------
        >>> results = db.bulk_get(docs=[{"id": "id-1"}, {"id": "id-2"}])
        >>> for item in results:
        ...     print(item["id"], item["docs"][0]["ok"])
        """
        return (
            self._post(
                resource="_bulk_get",
                body={"docs": [extract_document_id_and_rev(_) for _ in docs]},
                query_kwargs={"revs": revs},
            )
            .json()
            .get("results", [])
        )

    def compact(self, ddoc: str | None = None) -> bool:
        """
        Request compaction of the database. For more info, please refer to
        [the official documentation](https://docs.couchdb.org/en/main/api/database/compact.html#db-compact).

        If the `ddoc` parameter is provided, it will compact the view indexes associated with the specified design
        document.

        Parameters
        ----------
        ddoc : str
            A design document name.

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

    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. Copying a document is only possible within the same
        database. For more info, please refer to
        [the official documentation](https://docs.couchdb.org/en/main/api/document/common.html#copy--db-docid).

        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] : A tuple consisting of the id, success message & revision.
        """
        destination = destid
        if destrev:
            destination += f"?rev={destrev}"
        data = self._request(
            method="COPY",
            resource=docid,
            headers={
                "Destination": destination,
            },
            query_kwargs={"rev": rev},
        ).json()
        return data["id"], data["ok"], data["rev"]

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

        Parameters
        ----------
        doc : Union[Dict, couchdb3.document.Document]
            A dictionary or a `couchdb3.document.Document` instance to be created.
        batch : bool
            Stores document in batch mode. Default `None`.

        Returns
        -------
        Tuple[str, bool, str] : A tuple consisting of the id, success message & revision.
        """
        data = self._post(body=doc, query_kwargs={"batch": "ok" if batch is True else None}).json()
        return data["id"], data["ok"], data["rev"]

    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. If not known, one can use `Database.rev` with the given `docid`.
        batch : bool
            Stores document in batch mode. Default `None`.

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

    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. If not known, one can use `Database.rev` with the given `docid`.
        batch : bool
            Stores document in batch mode. Default `None`.

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

    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 `Database.find`.

        Parameters
        ----------
        selector : Dict
            JSON object describing criteria used to select documents. More information provided in CouchDB's section on
            [selector syntax](https://docs.couchdb.org/en/main/api/database/find.html#find-selectors).
        limit : int
            Maximum number of results returned. Default is `25`.
        skip : int
            Skip the first `n` results, where `n` is the value specified. Default is `0`.
        sort : Dict
             JSON array following CouchDB's [sort syntax]
             (https://docs.couchdb.org/en/main/api/database/find.html#find-sort). Default is `None`.
        fields : List[str]
            Dictionary specifying which fields of each object should be returned. If it is omitted, the entire object
            is returned. More information provided in CouchDB's [section on filtering fields]
            (https://docs.couchdb.org/en/main/api/database/find.html#find-filter).
        use_index : Union[str, List[str]]
            Instruct a query to use a specific index. Specified either as `"<design_document>"` or
            `["<design_document>", "<index_name>"]`. Default is `None`.
        conflicts : bool
            Include conflicted documents if `True`. Intended use is to easily find conflicted documents,
            without an index or view. Default is `False`.
        r : int
            Read quorum needed for the result. This defaults to 1, in which case the document found in the index is
            returned. If set to a higher value, each document is read from at least that many replicas before it is
            returned in the results. This is likely to take more time than using only the document stored locally with
            the index. Default is `None`.
        bookmark : str
            A string that enables you to specify which page of results you require. Used for paging through result
            sets. Every query returns an opaque string under the `bookmark` key that can then be passed back in a query
            to get the next page of results. If any part of the selector query changes between requests, the results
            are undefined. Default is `None`.
        update: bool
            Whether to update the index prior to returning the result. Default is `True`.
        stable : bool
            Whether the view results should be returned from a “stable” set of shards. Default is `None`.
        execution_stats : bool
            Include [execution statistics](https://docs.couchdb.org/en/main/api/database/find.html#find-statistics) in
            the query response. Default is `False`.

        Returns
        -------
        Dict: A dictionary containing the following keys.

          - dbname (`str`) – Name of database
          - index (`Dict`) – Index used to fulfill the query
          - selector (`Dict`) – Query selector used
          - opts (`Dict`) – Query options used
          - limit (`int`) – Limit parameter used
          - skip (`int`) – Skip parameter used
          - fields (`List`) – Fields to be returned by the query
          - range (`Dict`) – Range parameters passed to the underlying view

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

    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. More information provided in CouchDB's section on
            [selector syntax](https://docs.couchdb.org/en/main/api/database/find.html#find-selectors).
        limit : int
            Maximum number of results returned. Default is `25`.
        skip : int
            Skip the first `n` results, where `n` is the value specified. Default is `0`.
        sort : Dict
             JSON array following CouchDB's [sort syntax]
             (https://docs.couchdb.org/en/main/api/database/find.html#find-sort). Default is `None`.
        fields : List[str]
            Dictionary specifying which fields of each object should be returned. If it is omitted, the entire object
            is returned. More information provided in CouchDB's [section on filtering fields]
            (https://docs.couchdb.org/en/main/api/database/find.html#find-filter).
        use_index : Union[str, List[str]]
            Instruct a query to use a specific index. Specified either as `"<design_document>"` or
            `["<design_document>", "<index_name>"]`. Default is `None`.
        conflicts : bool
            Include conflicted documents if `True`. Intended use is to easily find conflicted documents,
            without an index or view. Default is `False`.
        r : int
            Read quorum needed for the result. This defaults to 1, in which case the document found in the index is
            returned. If set to a higher value, each document is read from at least that many replicas before it is
            returned in the results. This is likely to take more time than using only the document stored locally with
            the index. Default is `None`.
        bookmark : str
            A string that enables you to specify which page of results you require. Used for paging through result
            sets. Every query returns an opaque string under the `bookmark` key that can then be passed back in a query
            to get the next page of results. If any part of the selector query changes between requests, the results
            are undefined. Default is `None`.
        update: bool
            Whether to update the index prior to returning the result. Default is `True`.
        stable : bool
            Whether the view results should be returned from a “stable” set of shards. Default is `None`.
        execution_stats : bool
            Include [execution statistics](https://docs.couchdb.org/en/main/api/database/find.html#find-statistics) in
            the query response. Default is `False`.
        partition: str
            An optional partition ID. Only valid for partitioned databases. (Default `None`.)

        Returns
        -------
        Dict: A dictionary containing the following keys.

          - `bookmark`
          - `docs`
          - `warning`

        Examples
        --------
        >>> db.save_index({"fields": ["type", "name"]}, ddoc="my-ddoc", name="type-name-idx")
        >>> result = db.find({"type": {"$eq": "post"}}, fields=["_id", "name"], limit=10)
        >>> for doc in result["docs"]:
        ...     print(doc)
        """
        return 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()

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

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

          - total_rows (`int`) – Number of indexes
          - indexes (`List[Dict]`) – Array of index definitions
        """
        return self._get(resource="_index").json()

    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 attachments bodies in response. Default `None`.
        att_encoding_info : bool
            Includes encoding information in attachment stubs if the particular attachment is compressed.
            Default `None`.
        atts_since : Iterable[str]
            Includes attachments only since specified revisions. Doesn’t includes attachments for specified revisions.
            Default `None`.
        conflicts : bool
            Includes information about conflicts in document. Default `None`.
        deleted_conflicts : bool
            Includes information about deleted conflicted revisions. Default `None`.
        latest : bool
            Forces retrieving latest “leaf” revision, no matter what rev was requested. Default `None`.
        local_seq : bool
            Includes last update sequence for the document. Default `None`.
        meta : bool
            Acts same as specifying all conflicts, deleted_conflicts and revs_info query parameters. Default `None`.
        open_revs : Iterable[str]
            Retrieves documents of specified leaf revisions. Additionally, it accepts value as all to return all 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 `docid` cannot be found in the database. Default `False`.
        default_value : Any
            The default value to return if `check=False` and the `docid` is not in the database. Default `None`.

        Returns
        -------
        `couchdb3.document.Document`

        Examples
        --------
        >>> doc = db.get("mydoc-id")          # returns Document or None
        >>> doc = db["mydoc-id"]              # subscript shorthand; raises KeyError if not found
        >>> doc = db.get("missing-id")        # returns None
        >>> doc = db.get("missing-id", check=True)  # raises CouchDBError if not found
        """
        try:
            return Document(
                **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

    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.

        Returns
        -------
        AttachmentDocument : A `couchdb3.document.AttachmentDocument` instance.
        """
        response = 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,
        )

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

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

        Returns
        -------
        Document : A `couchdb3.document.Document` object containing the design document's content.
        """
        return self.get(docid=f"_design/{ddoc}", **kwargs)

    def purge(self, data: dict) -> dict:
        """
        Purge permanently the given pairs of `(id,rev)`. When deleting a (revisions of a) document, the document is
        marked as `_deleted=true` as opposed to being completely purged. For more info, please refer to
        [the official documentation](https://docs.couchdb.org/en/main/api/database/misc.html#db-purge).

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

        Returns
        -------

        """
        return self._post(resource="_purge", body=data).json()

    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]:
        """
        Uploads the supplied content as an attachment to the specified document.

        Parameters
        ----------
        docid : str
            The document's id.
        attname : str
            The attachment's name.
        path : str
            The path ot the local file to be uploaded.
            Precisely one of the arguments `path` or `content` must be supplied.
        content : bytes
            The content to be uploaded.
            Precisely one of the arguments `path` or `content` must be supplied.
        content_type : str
            The attachment's content-type (mime-type).
            Must be provided when passing the `content` argument.
        rev : str
            The document's current revision. Must be supplied for existing documents.

        Returns
        -------
        Tuple[str, bool, str] : A tuple consisting of the following elements.

          - the document's id ( `str`)
          - the operation status (`bool`)
          - the revision ( `str`)
        """
        if (not content and not path) or (content and path):
            raise ValueError(
                'Precisely one of the arguments "attdata" and  "attloc" 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 = file.read()
        response = 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"]

    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. For more info, please refer to
        [the official documentation](https://docs.couchdb.org/en/latest/api/ddoc/common.html#put--db-_design-ddoc).

        Parameters
        ----------
        ddoc : str
            The design document's name.
        rev : str
            The design document's revision in case of an update.
        language : str
            Defines [Query Server](https://docs.couchdb.org/en/latest/query-server/index.html#query-server) to process
            design document functions.
        options : Dict
            View’s default options.
        filters : Dict
            [Filter functions](https://docs.couchdb.org/en/latest/ddocs/ddocs.html#filterfun) definition.
        updates : Dict
            [Update functions](https://docs.couchdb.org/en/latest/ddocs/ddocs.html#updatefun) definition.
        validate_doc_update : str
            [Validate document update](https://docs.couchdb.org/en/latest/ddocs/ddocs.html#vdufun) function source.
        views : Dict
            [View functions](https://docs.couchdb.org/en/latest/ddocs/ddocs.html#viewfun) definition.
        autoupdate : bool
            Indicates whether to automatically build indexes defined in this design document.
        partitioned : bool
            Set to `True` for a partitioned design.
        kwargs
        Further `Database.save` parameters.

        Returns
        -------
        Tuple[str, bool, str] : The document's id ( `str`), the operation status (`bool`) and the revision ( `str`).

        Examples
        --------
        >>> db.put_design("my-ddoc", views={
        ...     "my-view": {
        ...         "map": "function(doc) { if (doc.type === 'post') emit(doc._id, null); }"
        ...     }
        ... })
        """
        if partitioned:
            options = {**(options or {}), "partitioned": partitioned}
        return 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,
        )

    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 the existing document.

        Parameters
        ----------
        doc : Union[Dict, couchdb3.document.Document]
            A dictionary or a `couchdb3.document.Document` instance containing a valid identifier (`doc["_id"]`)
            as well as revision number (`doc["_rev"]`) if need be.
        batch : bool
            Store document in batch mode. Default `None`.
        new_edits : bool
            Prevents insertion of a conflicting document. If false, a well-formed _rev must be included in the document.
            `new_edits=False` is used by the replicator to insert documents into the target database even if that leads
            to the creation of conflicts.
        path : str
            Database path, e.g `_design`. Default `None`.

        Returns
        -------
        Tuple[str, bool, str] : The document's id ( `str`), the operation status (`bool`) and the revision ( `str`).
        """
        batch = "ok" if batch else None
        data = self._put(
            resource="{}/{}".format(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"]

    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 in which the index will be created. By default, each index will be created in
            its own design document. Indexes can be grouped into design documents for efficiency. However, a change to
            one index in a design document will invalidate all other indexes in the same document (similar to views).
        name : str
            Name of the index. If no name is provided, a name will be generated automatically.
        index_type : str
            Can be `json`  or `text`. Defaults to `json`. Geospatial indexes will be supported in the future. Optional
            Text indexes are supported via a third party library
        partitioned : bool
             Determines whether a JSON index is partitioned or global. The default value of `partitioned` is the
             `partitioned` property of the database. To create a global index on a partitioned database, specify `False`
             for the `"partitioned"` field. If you specify `True` for the `"partitioned"` field on an unpartitioned
             database, an error occurs.

        Returns
        -------
        Tuple[str, str, str]: A tuple consisting of the following elements.

          - result (`str`) – Flag to show whether the index was created or one already exists. Can be `"created"` or
          `"exists"`.
          - id (`str`) – Id of the design document the index was created in.
          - name (`str`) – Name of the index created.

        Examples
        --------
        >>> result, id, name = db.save_index(
        ...     {"fields": ["type", "name"]},
        ...     ddoc="my-ddoc",
        ...     name="type-name-idx",
        ... )
        """
        data = 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"]

    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 self._delete(resource=f"_index/{ddoc}/{index_type}/{name}").json().get("ok")

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

        Returns
        -------
        SecurityDocument : A `couchdb3.document.SecurityDocument` object.
        """
        data = self._get(resource="_security").json()
        return SecurityDocument(**data)

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

        Parameters
        ----------
        admins : Union[Dict, SecurityDocumentElement]
            Object with two fields as `names` and `roles`. [See the official
            documentation](https://docs.couchdb.org/en/main/api/database/security.html#db-security) for more info.
        members : Union[Dict, SecurityDocumentElement]
            Object with two fields as `names` and `roles`. [See the official
            documentation](https://docs.couchdb.org/en/main/api/database/security.html#db-security) for more info.

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

    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, c.f [the official
        documentation](https://docs.couchdb.org/en/main/api/ddoc/views.html#db-design-design-doc-view-view-name).

        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. Ignored if `include_docs` isn’t `True`. Default is `None`.
        descending : bool
            Return the documents in descending order by key. Default is `None`.
        endkey : Any
             Stop returning records when the specified key is reached. Default is `None`
        endkey_docid: str
            Stop returning records when the specified document ID is reached. Ignored if `endkey` is not set. Default
            is `None`.
        group: bool
            Group the results using the reduce function to a group or single row. Implies `reduce` is `true` and the
            maximum `group_level`. Default is `None`.
        group_level : int
             Specify the group level to be used. Implies group is true. Default is `None`.
        include_docs : bool
            Include the associated document with each row. Default is `None`.
        attachments : bool
            Include the Base64-encoded content of attachments in the documents that are included if `include_docs` is
            `True`. Ignored if `include_docs` isn’t `True`. Default is `None`.
        att_encoding_info : bool
            Include encoding information in attachment stubs if `include_docs` is `True` and the particular attachment
            is compressed. Ignored if `include_docs` isn’t `True`. Default is `False`.
        inclusive_end : bool
            Specifies whether the specified end key should be included in the result. Default is `None`.
        key : str
            Return only documents that match the specified key. 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 to the specified number.  Default is `None`.
        reduce : bool
            Use the reduction function. Default is `True` when a reduce function is defined. Default is `None`.
        skip : int
            Skip this number of records before starting to return the results. Default is `None`.
        sort : bool
            Sort returned rows (see Sorting Returned Rows). Setting this to `False` offers a performance boost. The
            `total_rows` and `offset` fields are not available when this is set to `False`. Default is `None`.
        stable : bool
            Whether or not the view results should be returned from a stable set of shards. Default is `None`.
        startkey : Any
            Return records starting with the specified key. Default is `None`
        startkey_docid : str
            Return records starting with the specified document ID. Ignored if `startkey` is not set. Default is `None`
        update : str
            Whether or not the view in question should be updated prior to responding to the user. Supported values:

            - `true`
            - `false`
            - `lazy`

            Default is `None`.
        update_seq : bool
             Whether to include in the response an `update_seq` value indicating the sequence id of the database the
             view reflects. Default is `False`.

        Returns
        -------
        `view.ViewResult`

        Examples
        --------
        >>> import couchdb3
        >>> db = couchdb3.Server("http://localhost:5984").get("mydb")
        >>> db.view("myddoc", "myview")
        ViewResult: {
            'total_rows': 1,
            'offset': 0,
            'rows': [
                {
                    'id': 'doc-test-view',
                    'key': 'doc-test-view',
                    'value': None
                }
            ]
        }
        >>> result = db.view("my-ddoc", "my-view", include_docs=True, limit=10)
        >>> for row in result.rows:
        ...     print(row.id, row.key, row.value, row.doc)
        """
        path = partitioned_db_resource_parser(
            resource="_design",
            partition=partition,
        )
        return ViewResult(
            **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()
        )

    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 self._post(
                resource="_changes",
                body={"doc_ids": doc_ids},
                query_kwargs=query_kwargs,
            ).json()
        if selector is not None:
            query_kwargs["filter"] = "_selector"
            return self._post(
                resource="_changes",
                body={"selector": selector},
                query_kwargs=query_kwargs,
            ).json()
        return self._get(resource="_changes", query_kwargs=query_kwargs).json()

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

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

        Returns
        -------
        `Partition`
        """
        return Partition(
            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,
            _database=self,
        )

Abstract Couchdb database

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.Client

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

_server : Server

The owning Server instance. Set internally by Server.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 a Database directly.

Ancestors

• Base

Subclasses

• Partition

Instance variables

prop server

Expand source code

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

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

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

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

Read-only. Setting this attribute raises AttributeError.

Returns

Server | None

 

Methods

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

Expand source code

def all_docs(
    self,
    partition: str | None = None,
    keys: Iterable[str] | None = None,
    **kwargs,
) -> ViewResult:
    """
    Executes the built-in _all_docs view, returning all the 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 in the argument. Default is `None`.
    kwargs
        Further `couchdb3.sync.Database.view` parameters.

    Returns
    -------
    ViewResult

    Examples
    --------
    >>> db.all_docs()
    >>> db.all_docs(include_docs=True)
    >>> for row in db.all_docs(include_docs=True).rows:
    ...     print(row.id, row.doc)
    >>> db.all_docs(keys=["id-1", "id-2"], include_docs=True)
    """
    return self.view(
        f"_partition/{partition}/_all_docs" if partition else "_all_docs",
        keys=keys,
        **kwargs,
    )

Executes the built-in _all_docs view, returning all the 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 in the argument. Default is None.

kwargs

Further Database.view() parameters.

Returns

ViewResult

 

Examples

>>> db.all_docs()
>>> db.all_docs(include_docs=True)
>>> for row in db.all_docs(include_docs=True).rows:
...     print(row.id, row.doc)
>>> db.all_docs(keys=["id-1", "id-2"], include_docs=True)

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

Expand source code

def bulk_docs(self, docs: list[dict | Document], new_edits: bool = True) -> list[dict]:
    """
    The bulk document API allows you to create and update multiple documents at the same time within a single
    request. The basic operation is similar to creating or updating a single document, except that you batch the
    document structure and information.

    When creating new documents the document ID (`_id`) is optional.

    For updating existing documents, you must provide the document ID, revision information (`_rev`), and new
    document values.

    In case of batch deleting documents all fields as document ID, revision information and deletion status
    (`_deleted`) are required.

    Parameters
    ----------
    docs : List[Union[Dict, Document]]
         List of documents objects
    new_edits : bool
        If `False`, prevents the database from assigning them new revision IDs. Default `True`.

    Returns
    -------
    List[Dict] : A list of dictionaries containing the following keys.

      - `id` the document's id
      - `ok` operation status
      - `rev` the document's revision
    """
    return self._post(resource="_bulk_docs", body={"docs": docs, "new_edits": new_edits}).json()

The bulk document API allows you to create and update multiple documents at the same time within a single request. The basic operation is similar to creating or updating a single document, except that you batch the document structure and information.

When creating new documents the document ID (_id) is optional.

For updating existing documents, you must provide the document ID, revision information (_rev), and new document values.

In case of batch deleting documents all fields as document ID, revision information and deletion status (_deleted) are required.

Parameters

docs : List[Union[Dict, Document]]

List of documents objects

new_edits : bool

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

Returns

List[Dict] : A list of dictionaries containing the following keys.

• id the document's id
• ok operation status
• rev the document's revision

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

Expand source code

def bulk_get(
    self,
    docs: list[dict | Document],
    revs: bool = False,
) -> list[dict]:
    """
    This method can be called to query several documents in bulk. It is well suited for fetching a specific
    revision of documents, as replicators do for example, or for getting revision history.

    Parameters
    ----------
    docs : List[Union[Dict, Document]]
        List of document objects, with `id`, and optionally `rev` and `atts_since`.
    revs : bool
         Give the revisions history.

    Returns
    -------
    List[Dict] : An array of results for each requested document/rev pair.

      - `id` key lists the requested
      document ID,
      - `docs` contains a single-item array of objects, each of which has either an `error` key and value describing
      the error, or `ok` key and associated value of the requested document, with the additional _revisions property
      that lists the parent revisions if `revs=true`.

    Examples
    --------
    >>> results = db.bulk_get(docs=[{"id": "id-1"}, {"id": "id-2"}])
    >>> for item in results:
    ...     print(item["id"], item["docs"][0]["ok"])
    """
    return (
        self._post(
            resource="_bulk_get",
            body={"docs": [extract_document_id_and_rev(_) for _ in docs]},
            query_kwargs={"revs": revs},
        )
        .json()
        .get("results", [])
    )

This method can be called to query several documents in bulk. It is well suited for fetching a specific revision of documents, as replicators do for example, or for getting revision history.

Parameters

docs : List[Union[Dict, Document]]

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

revs : bool

Give the revisions history.

Returns

List[Dict] : An array of results for each requested document/rev pair.

• id key lists the requested document ID,
• docs contains a single-item array of objects, each of which has either an error key and value describing the error, or ok key and associated value of the requested document, with the additional _revisions property that lists the parent revisions if revs=true.

Examples

>>> results = db.bulk_get(docs=[{"id": "id-1"}, {"id": "id-2"}])
>>> for item in results:
...     print(item["id"], item["docs"][0]["ok"])

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

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 self._post(
            resource="_changes",
            body={"doc_ids": doc_ids},
            query_kwargs=query_kwargs,
        ).json()
    if selector is not None:
        query_kwargs["filter"] = "_selector"
        return self._post(
            resource="_changes",
            body={"selector": selector},
            query_kwargs=query_kwargs,
        ).json()
    return 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.

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

Expand source code

def compact(self, ddoc: str | None = None) -> bool:
    """
    Request compaction of the database. For more info, please refer to
    [the official documentation](https://docs.couchdb.org/en/main/api/database/compact.html#db-compact).

    If the `ddoc` parameter is provided, it will compact the view indexes associated with the specified design
    document.

    Parameters
    ----------
    ddoc : str
        A design document name.

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

Request compaction of the database. For more info, please refer to the official documentation.

If the ddoc parameter is provided, it will compact the view indexes associated with the specified design document.

Parameters

ddoc : str

A design document name.

Returns

bool: True upon compaction request successfully sent.

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

Expand source code

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. Copying a document is only possible within the same
    database. For more info, please refer to
    [the official documentation](https://docs.couchdb.org/en/main/api/document/common.html#copy--db-docid).

    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] : A tuple consisting of the id, success message & revision.
    """
    destination = destid
    if destrev:
        destination += f"?rev={destrev}"
    data = 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. Copying a document is only possible within the same database. For more info, please refer to the official documentation.

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] : A tuple consisting of the id, success message & revision.

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

Expand source code

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

    Parameters
    ----------
    doc : Union[Dict, couchdb3.document.Document]
        A dictionary or a `couchdb3.document.Document` instance to be created.
    batch : bool
        Stores document in batch mode. Default `None`.

    Returns
    -------
    Tuple[str, bool, str] : A tuple consisting of the id, success message & revision.
    """
    data = 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.

Parameters

doc : Union[Dict, Document]

A dictionary or a Document instance to be created.

batch : bool

Stores document in batch mode. Default None.

Returns

Tuple[str, bool, str] : A tuple consisting of the id, success message & revision.

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

Expand source code

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. If not known, one can use `Database.rev` with the given `docid`.
    batch : bool
        Stores document in batch mode. Default `None`.

    Returns
    -------
    bool : `True` upon successful deletion.
    """
    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. If not known, one can use Base.rev() with the given docid.

batch : bool

Stores document in batch mode. Default None.

Returns

bool : True upon successful deletion.

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

Expand source code

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. If not known, one can use `Database.rev` with the given `docid`.
    batch : bool
        Stores document in batch mode. Default `None`.

    Returns
    -------
    bool : `True` upon successful deletion.
    """
    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. If not known, one can use Base.rev() with the given docid.

batch : bool

Stores document in batch mode. Default None.

Returns

bool : True upon successful deletion.

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

Expand source code

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 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.

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

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

 

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

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 `Database.find`.

    Parameters
    ----------
    selector : Dict
        JSON object describing criteria used to select documents. More information provided in CouchDB's section on
        [selector syntax](https://docs.couchdb.org/en/main/api/database/find.html#find-selectors).
    limit : int
        Maximum number of results returned. Default is `25`.
    skip : int
        Skip the first `n` results, where `n` is the value specified. Default is `0`.
    sort : Dict
         JSON array following CouchDB's [sort syntax]
         (https://docs.couchdb.org/en/main/api/database/find.html#find-sort). Default is `None`.
    fields : List[str]
        Dictionary specifying which fields of each object should be returned. If it is omitted, the entire object
        is returned. More information provided in CouchDB's [section on filtering fields]
        (https://docs.couchdb.org/en/main/api/database/find.html#find-filter).
    use_index : Union[str, List[str]]
        Instruct a query to use a specific index. Specified either as `"<design_document>"` or
        `["<design_document>", "<index_name>"]`. Default is `None`.
    conflicts : bool
        Include conflicted documents if `True`. Intended use is to easily find conflicted documents,
        without an index or view. Default is `False`.
    r : int
        Read quorum needed for the result. This defaults to 1, in which case the document found in the index is
        returned. If set to a higher value, each document is read from at least that many replicas before it is
        returned in the results. This is likely to take more time than using only the document stored locally with
        the index. Default is `None`.
    bookmark : str
        A string that enables you to specify which page of results you require. Used for paging through result
        sets. Every query returns an opaque string under the `bookmark` key that can then be passed back in a query
        to get the next page of results. If any part of the selector query changes between requests, the results
        are undefined. Default is `None`.
    update: bool
        Whether to update the index prior to returning the result. Default is `True`.
    stable : bool
        Whether the view results should be returned from a “stable” set of shards. Default is `None`.
    execution_stats : bool
        Include [execution statistics](https://docs.couchdb.org/en/main/api/database/find.html#find-statistics) in
        the query response. Default is `False`.

    Returns
    -------
    Dict: A dictionary containing the following keys.

      - dbname (`str`) – Name of database
      - index (`Dict`) – Index used to fulfill the query
      - selector (`Dict`) – Query selector used
      - opts (`Dict`) – Query options used
      - limit (`int`) – Limit parameter used
      - skip (`int`) – Skip parameter used
      - fields (`List`) – Fields to be returned by the query
      - range (`Dict`) – Range parameters passed to the underlying view

    """
    return 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 Database.find().

Parameters

selector : Dict

JSON object describing criteria used to select documents. More information provided in CouchDB's section on selector syntax.

limit : int

Maximum number of results returned. Default is 25.

skip : int

Skip the first n results, where n is the value specified. Default is 0.

sort : Dict

JSON array following CouchDB's [sort syntax] (https://docs.couchdb.org/en/main/api/database/find.html#find-sort). Default is None.

fields : List[str]

Dictionary specifying which fields of each object should be returned. If it is omitted, the entire object is returned. More information provided in CouchDB's [section on filtering fields] (https://docs.couchdb.org/en/main/api/database/find.html#find-filter).

use_index : Union[str, List[str]]

Instruct a query to use a specific index. Specified either as "<design_document>" or ["<design_document>", "<index_name>"]. Default is None.

conflicts : bool

Include conflicted documents if True. Intended use is to easily find conflicted documents, without an index or view. Default is False.

r : int

Read quorum needed for the result. This defaults to 1, in which case the document found in the index is returned. If set to a higher value, each document is read from at least that many replicas before it is returned in the results. This is likely to take more time than using only the document stored locally with the index. Default is None.

bookmark : str

A string that enables you to specify which page of results you require. Used for paging through result sets. Every query returns an opaque string under the bookmark key that can then be passed back in a query to get the next page of results. If any part of the selector query changes between requests, the results are undefined. Default is None.

update : bool

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

stable : bool

Whether the view results should be returned from a “stable” set of shards. Default is None.

execution_stats : bool

Include execution statistics in the query response. Default is False.

Returns

Dict: A dictionary containing the following keys.

• dbname (str) – Name of database
• index (Dict) – Index used to fulfill the query
• selector (Dict) – Query selector used
• opts (Dict) – Query options used
• limit (int) – Limit parameter used
• skip (int) – Skip parameter used
• fields (List) – Fields to be returned by the query
• range (Dict) – Range parameters passed to the underlying view

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

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. More information provided in CouchDB's section on
        [selector syntax](https://docs.couchdb.org/en/main/api/database/find.html#find-selectors).
    limit : int
        Maximum number of results returned. Default is `25`.
    skip : int
        Skip the first `n` results, where `n` is the value specified. Default is `0`.
    sort : Dict
         JSON array following CouchDB's [sort syntax]
         (https://docs.couchdb.org/en/main/api/database/find.html#find-sort). Default is `None`.
    fields : List[str]
        Dictionary specifying which fields of each object should be returned. If it is omitted, the entire object
        is returned. More information provided in CouchDB's [section on filtering fields]
        (https://docs.couchdb.org/en/main/api/database/find.html#find-filter).
    use_index : Union[str, List[str]]
        Instruct a query to use a specific index. Specified either as `"<design_document>"` or
        `["<design_document>", "<index_name>"]`. Default is `None`.
    conflicts : bool
        Include conflicted documents if `True`. Intended use is to easily find conflicted documents,
        without an index or view. Default is `False`.
    r : int
        Read quorum needed for the result. This defaults to 1, in which case the document found in the index is
        returned. If set to a higher value, each document is read from at least that many replicas before it is
        returned in the results. This is likely to take more time than using only the document stored locally with
        the index. Default is `None`.
    bookmark : str
        A string that enables you to specify which page of results you require. Used for paging through result
        sets. Every query returns an opaque string under the `bookmark` key that can then be passed back in a query
        to get the next page of results. If any part of the selector query changes between requests, the results
        are undefined. Default is `None`.
    update: bool
        Whether to update the index prior to returning the result. Default is `True`.
    stable : bool
        Whether the view results should be returned from a “stable” set of shards. Default is `None`.
    execution_stats : bool
        Include [execution statistics](https://docs.couchdb.org/en/main/api/database/find.html#find-statistics) in
        the query response. Default is `False`.
    partition: str
        An optional partition ID. Only valid for partitioned databases. (Default `None`.)

    Returns
    -------
    Dict: A dictionary containing the following keys.

      - `bookmark`
      - `docs`
      - `warning`

    Examples
    --------
    >>> db.save_index({"fields": ["type", "name"]}, ddoc="my-ddoc", name="type-name-idx")
    >>> result = db.find({"type": {"$eq": "post"}}, fields=["_id", "name"], limit=10)
    >>> for doc in result["docs"]:
    ...     print(doc)
    """
    return 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. More information provided in CouchDB's section on selector syntax.

limit : int

Maximum number of results returned. Default is 25.

skip : int

Skip the first n results, where n is the value specified. Default is 0.

sort : Dict

JSON array following CouchDB's [sort syntax] (https://docs.couchdb.org/en/main/api/database/find.html#find-sort). Default is None.

fields : List[str]

Dictionary specifying which fields of each object should be returned. If it is omitted, the entire object is returned. More information provided in CouchDB's [section on filtering fields] (https://docs.couchdb.org/en/main/api/database/find.html#find-filter).

use_index : Union[str, List[str]]

Instruct a query to use a specific index. Specified either as "<design_document>" or ["<design_document>", "<index_name>"]. Default is None.

conflicts : bool

Include conflicted documents if True. Intended use is to easily find conflicted documents, without an index or view. Default is False.

r : int

Read quorum needed for the result. This defaults to 1, in which case the document found in the index is returned. If set to a higher value, each document is read from at least that many replicas before it is returned in the results. This is likely to take more time than using only the document stored locally with the index. Default is None.

bookmark : str

A string that enables you to specify which page of results you require. Used for paging through result sets. Every query returns an opaque string under the bookmark key that can then be passed back in a query to get the next page of results. If any part of the selector query changes between requests, the results are undefined. Default is None.

update : bool

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

stable : bool

Whether the view results should be returned from a “stable” set of shards. Default is None.

execution_stats : bool

Include execution statistics in the query response. Default is False.

partition : str

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

Returns

Dict: A dictionary containing the following keys.

• bookmark
• docs
• warning

Examples

>>> db.save_index({"fields": ["type", "name"]}, ddoc="my-ddoc", name="type-name-idx")
>>> result = db.find({"type": {"$eq": "post"}}, fields=["_id", "name"], limit=10)
>>> for doc in result["docs"]:
...     print(doc)

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

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 attachments bodies in response. Default `None`.
    att_encoding_info : bool
        Includes encoding information in attachment stubs if the particular attachment is compressed.
        Default `None`.
    atts_since : Iterable[str]
        Includes attachments only since specified revisions. Doesn’t includes attachments for specified revisions.
        Default `None`.
    conflicts : bool
        Includes information about conflicts in document. Default `None`.
    deleted_conflicts : bool
        Includes information about deleted conflicted revisions. Default `None`.
    latest : bool
        Forces retrieving latest “leaf” revision, no matter what rev was requested. Default `None`.
    local_seq : bool
        Includes last update sequence for the document. Default `None`.
    meta : bool
        Acts same as specifying all conflicts, deleted_conflicts and revs_info query parameters. Default `None`.
    open_revs : Iterable[str]
        Retrieves documents of specified leaf revisions. Additionally, it accepts value as all to return all 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 `docid` cannot be found in the database. Default `False`.
    default_value : Any
        The default value to return if `check=False` and the `docid` is not in the database. Default `None`.

    Returns
    -------
    `couchdb3.document.Document`

    Examples
    --------
    >>> doc = db.get("mydoc-id")          # returns Document or None
    >>> doc = db["mydoc-id"]              # subscript shorthand; raises KeyError if not found
    >>> doc = db.get("missing-id")        # returns None
    >>> doc = db.get("missing-id", check=True)  # raises CouchDBError if not found
    """
    try:
        return Document(
            **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 attachments bodies in response. Default None.

att_encoding_info : bool

Includes encoding information in attachment stubs if the particular attachment is compressed. Default None.

atts_since : Iterable[str]

Includes attachments only since specified revisions. Doesn’t includes attachments for specified revisions. Default None.

conflicts : bool

Includes information about conflicts in document. Default None.

deleted_conflicts : bool

Includes information about deleted conflicted revisions. Default None.

latest : bool

Forces retrieving latest “leaf” revision, no matter what rev was requested. Default None.

local_seq : bool

Includes last update sequence for the document. Default None.

meta : bool

Acts same as specifying all conflicts, deleted_conflicts and revs_info query parameters. Default None.

open_revs : Iterable[str]

Retrieves documents of specified leaf revisions. Additionally, it accepts value as all to return all 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 docid cannot be found in the database. Default False.

default_value : Any

The default value to return if check=False and the docid is not in the database. Default None.

Returns

Document

Examples

>>> doc = db.get("mydoc-id")          # returns Document or None
>>> doc = db["mydoc-id"]              # subscript shorthand; raises KeyError if not found
>>> doc = db.get("missing-id")        # returns None
>>> doc = db.get("missing-id", check=True)  # raises CouchDBError if not found

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

Expand source code

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.

    Returns
    -------
    AttachmentDocument : A `couchdb3.document.AttachmentDocument` instance.
    """
    response = 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.

Returns

AttachmentDocument : A AttachmentDocument instance.

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

Expand source code

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

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

    Returns
    -------
    Document : A `couchdb3.document.Document` object containing the design document's content.
    """
    return self.get(docid=f"_design/{ddoc}", **kwargs)

Get a design document.

Parameters

ddoc : str

The design document's name.

kwargs

Further Database.get() parameters.

Returns

Document : A Document object containing the design document's content.

def get_partition(self, partition_id: str) ‑> Partition

Expand source code

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

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

    Returns
    -------
    `Partition`
    """
    return Partition(
        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,
        _database=self,
    )

Get a given partition.

Parameters

partition_id : str

The partition's ID.

Returns

Partition

def indexes(self) ‑> dict

Expand source code

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

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

      - total_rows (`int`) – Number of indexes
      - indexes (`List[Dict]`) – Array of index definitions
    """
    return self._get(resource="_index").json()

Get a list of all indexes in the database.

Returns

Dict : A dictionary with the following keys.

• total_rows (int) – Number of indexes
• indexes (List[Dict]) – Array of index definitions

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

Expand source code

def purge(self, data: dict) -> dict:
    """
    Purge permanently the given pairs of `(id,rev)`. When deleting a (revisions of a) document, the document is
    marked as `_deleted=true` as opposed to being completely purged. For more info, please refer to
    [the official documentation](https://docs.couchdb.org/en/main/api/database/misc.html#db-purge).

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

    Returns
    -------

    """
    return self._post(resource="_purge", body=data).json()

Purge permanently the given pairs of (id,rev). When deleting a (revisions of a) document, the document is marked as _deleted=true as opposed to being completely purged. For more info, please refer to the official documentation.

Parameters

data : Dict

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

Returns

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

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]:
    """
    Uploads the supplied content as an attachment to the specified document.

    Parameters
    ----------
    docid : str
        The document's id.
    attname : str
        The attachment's name.
    path : str
        The path ot the local file to be uploaded.
        Precisely one of the arguments `path` or `content` must be supplied.
    content : bytes
        The content to be uploaded.
        Precisely one of the arguments `path` or `content` must be supplied.
    content_type : str
        The attachment's content-type (mime-type).
        Must be provided when passing the `content` argument.
    rev : str
        The document's current revision. Must be supplied for existing documents.

    Returns
    -------
    Tuple[str, bool, str] : A tuple consisting of the following elements.

      - the document's id ( `str`)
      - the operation status (`bool`)
      - the revision ( `str`)
    """
    if (not content and not path) or (content and path):
        raise ValueError(
            'Precisely one of the arguments "attdata" and  "attloc" 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 = file.read()
    response = 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"]

Uploads the supplied content as an attachment to the specified document.

Parameters

docid : str

The document's id.

attname : str

The attachment's name.

path : str

The path ot the local file to be uploaded. Precisely one of the arguments path or content must be supplied.

content : bytes

The content to be uploaded. Precisely one of the arguments path or content must be supplied.

content_type : str

The attachment's content-type (mime-type). Must be provided when passing the content argument.

rev : str

The document's current revision. Must be supplied for existing documents.

Returns

Tuple[str, bool, str] : A tuple consisting of the following elements.

• the document's id ( str)
• the operation status (bool)
• the revision ( str)

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

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. For more info, please refer to
    [the official documentation](https://docs.couchdb.org/en/latest/api/ddoc/common.html#put--db-_design-ddoc).

    Parameters
    ----------
    ddoc : str
        The design document's name.
    rev : str
        The design document's revision in case of an update.
    language : str
        Defines [Query Server](https://docs.couchdb.org/en/latest/query-server/index.html#query-server) to process
        design document functions.
    options : Dict
        View’s default options.
    filters : Dict
        [Filter functions](https://docs.couchdb.org/en/latest/ddocs/ddocs.html#filterfun) definition.
    updates : Dict
        [Update functions](https://docs.couchdb.org/en/latest/ddocs/ddocs.html#updatefun) definition.
    validate_doc_update : str
        [Validate document update](https://docs.couchdb.org/en/latest/ddocs/ddocs.html#vdufun) function source.
    views : Dict
        [View functions](https://docs.couchdb.org/en/latest/ddocs/ddocs.html#viewfun) definition.
    autoupdate : bool
        Indicates whether to automatically build indexes defined in this design document.
    partitioned : bool
        Set to `True` for a partitioned design.
    kwargs
    Further `Database.save` parameters.

    Returns
    -------
    Tuple[str, bool, str] : The document's id ( `str`), the operation status (`bool`) and the revision ( `str`).

    Examples
    --------
    >>> db.put_design("my-ddoc", views={
    ...     "my-view": {
    ...         "map": "function(doc) { if (doc.type === 'post') emit(doc._id, null); }"
    ...     }
    ... })
    """
    if partitioned:
        options = {**(options or {}), "partitioned": partitioned}
    return 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. For more info, please refer to the official documentation.

Parameters

ddoc : str

The design document's name.

rev : str

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

language : str

Defines 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 defined in this design document.

partitioned : bool

Set to True for a partitioned design.

kwargs

 

Further Database.save() parameters.

Returns

Tuple[str, bool, str] : The document's id ( str), the operation status (bool) and the revision ( str).

Examples

>>> db.put_design("my-ddoc", views={
...     "my-view": {
...         "map": "function(doc) { if (doc.type === 'post') emit(doc._id, null); }"
...     }
... })

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

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 the existing document.

    Parameters
    ----------
    doc : Union[Dict, couchdb3.document.Document]
        A dictionary or a `couchdb3.document.Document` instance containing a valid identifier (`doc["_id"]`)
        as well as revision number (`doc["_rev"]`) if need be.
    batch : bool
        Store document in batch mode. Default `None`.
    new_edits : bool
        Prevents insertion of a conflicting document. If false, a well-formed _rev must be included in the document.
        `new_edits=False` is used by the replicator to insert documents into the target database even if that leads
        to the creation of conflicts.
    path : str
        Database path, e.g `_design`. Default `None`.

    Returns
    -------
    Tuple[str, bool, str] : The document's id ( `str`), the operation status (`bool`) and the revision ( `str`).
    """
    batch = "ok" if batch else None
    data = self._put(
        resource="{}/{}".format(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 the existing document.

Parameters

doc : Union[Dict, Document]

A dictionary or a Document instance containing a valid identifier (doc["_id"]) as well as revision number (doc["_rev"]) if need be.

batch : bool

Store document in batch mode. Default None.

new_edits : bool

Prevents insertion of a conflicting document. If false, a well-formed _rev must be included in the document. new_edits=False is used by the replicator to insert documents into the target database even if that leads to the creation of conflicts.

path : str

Database path, e.g _design. Default None.

Returns

Tuple[str, bool, str] : The document's id ( str), the operation status (bool) and the revision ( str).

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

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 in which the index will be created. By default, each index will be created in
        its own design document. Indexes can be grouped into design documents for efficiency. However, a change to
        one index in a design document will invalidate all other indexes in the same document (similar to views).
    name : str
        Name of the index. If no name is provided, a name will be generated automatically.
    index_type : str
        Can be `json`  or `text`. Defaults to `json`. Geospatial indexes will be supported in the future. Optional
        Text indexes are supported via a third party library
    partitioned : bool
         Determines whether a JSON index is partitioned or global. The default value of `partitioned` is the
         `partitioned` property of the database. To create a global index on a partitioned database, specify `False`
         for the `"partitioned"` field. If you specify `True` for the `"partitioned"` field on an unpartitioned
         database, an error occurs.

    Returns
    -------
    Tuple[str, str, str]: A tuple consisting of the following elements.

      - result (`str`) – Flag to show whether the index was created or one already exists. Can be `"created"` or
      `"exists"`.
      - id (`str`) – Id of the design document the index was created in.
      - name (`str`) – Name of the index created.

    Examples
    --------
    >>> result, id, name = db.save_index(
    ...     {"fields": ["type", "name"]},
    ...     ddoc="my-ddoc",
    ...     name="type-name-idx",
    ... )
    """
    data = 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 in which the index will be created. By default, each index will be created in its own design document. Indexes can be grouped into design documents for efficiency. However, a change to one index in a design document will invalidate all other indexes in the same document (similar to views).

name : str

Name of the index. If no name is provided, a name will be generated automatically.

index_type : str

Can be json or text. Defaults to json. Geospatial indexes will be supported in the future. Optional Text indexes are supported via a third party library

partitioned : bool

Determines whether a JSON index is partitioned or global. The default value of partitioned is the partitioned property of the database. To create a global index on a partitioned database, specify False for the "partitioned" field. If you specify True for the "partitioned" field on an unpartitioned database, an error occurs.

Returns

Tuple[str, str, str]: A tuple consisting of the following elements.

• result (str) – Flag to show whether the index was created or one already exists. Can be "created" or "exists".
• id (str) – Id of the design document the index was created in.
• name (str) – Name of the index created.

Examples

>>> result, id, name = db.save_index(
...     {"fields": ["type", "name"]},
...     ddoc="my-ddoc",
...     name="type-name-idx",
... )

def security(self) ‑> SecurityDocument

Expand source code

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

    Returns
    -------
    SecurityDocument : A `couchdb3.document.SecurityDocument` object.
    """
    data = self._get(resource="_security").json()
    return SecurityDocument(**data)

Returns the current security object from the specified database.

Returns

SecurityDocument : A SecurityDocument object.

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

Expand source code

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

    Parameters
    ----------
    admins : Union[Dict, SecurityDocumentElement]
        Object with two fields as `names` and `roles`. [See the official
        documentation](https://docs.couchdb.org/en/main/api/database/security.html#db-security) for more info.
    members : Union[Dict, SecurityDocumentElement]
        Object with two fields as `names` and `roles`. [See the official
        documentation](https://docs.couchdb.org/en/main/api/database/security.html#db-security) for more info.

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

Update database security.

Parameters

admins : Union[Dict, SecurityDocumentElement]

Object with two fields as names and roles. See the official documentation for more info.

members : Union[Dict, SecurityDocumentElement]

Object with two fields as names and roles. See the official documentation for more info.

Returns

bool : Operation status.

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

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, c.f [the official
    documentation](https://docs.couchdb.org/en/main/api/ddoc/views.html#db-design-design-doc-view-view-name).

    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. Ignored if `include_docs` isn’t `True`. Default is `None`.
    descending : bool
        Return the documents in descending order by key. Default is `None`.
    endkey : Any
         Stop returning records when the specified key is reached. Default is `None`
    endkey_docid: str
        Stop returning records when the specified document ID is reached. Ignored if `endkey` is not set. Default
        is `None`.
    group: bool
        Group the results using the reduce function to a group or single row. Implies `reduce` is `true` and the
        maximum `group_level`. Default is `None`.
    group_level : int
         Specify the group level to be used. Implies group is true. Default is `None`.
    include_docs : bool
        Include the associated document with each row. Default is `None`.
    attachments : bool
        Include the Base64-encoded content of attachments in the documents that are included if `include_docs` is
        `True`. Ignored if `include_docs` isn’t `True`. Default is `None`.
    att_encoding_info : bool
        Include encoding information in attachment stubs if `include_docs` is `True` and the particular attachment
        is compressed. Ignored if `include_docs` isn’t `True`. Default is `False`.
    inclusive_end : bool
        Specifies whether the specified end key should be included in the result. Default is `None`.
    key : str
        Return only documents that match the specified key. 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 to the specified number.  Default is `None`.
    reduce : bool
        Use the reduction function. Default is `True` when a reduce function is defined. Default is `None`.
    skip : int
        Skip this number of records before starting to return the results. Default is `None`.
    sort : bool
        Sort returned rows (see Sorting Returned Rows). Setting this to `False` offers a performance boost. The
        `total_rows` and `offset` fields are not available when this is set to `False`. Default is `None`.
    stable : bool
        Whether or not the view results should be returned from a stable set of shards. Default is `None`.
    startkey : Any
        Return records starting with the specified key. Default is `None`
    startkey_docid : str
        Return records starting with the specified document ID. Ignored if `startkey` is not set. Default is `None`
    update : str
        Whether or not the view in question should be updated prior to responding to the user. Supported values:

        - `true`
        - `false`
        - `lazy`

        Default is `None`.
    update_seq : bool
         Whether to include in the response an `update_seq` value indicating the sequence id of the database the
         view reflects. Default is `False`.

    Returns
    -------
    `view.ViewResult`

    Examples
    --------
    >>> import couchdb3
    >>> db = couchdb3.Server("http://localhost:5984").get("mydb")
    >>> db.view("myddoc", "myview")
    ViewResult: {
        'total_rows': 1,
        'offset': 0,
        'rows': [
            {
                'id': 'doc-test-view',
                'key': 'doc-test-view',
                'value': None
            }
        ]
    }
    >>> result = db.view("my-ddoc", "my-view", include_docs=True, limit=10)
    >>> for row in result.rows:
    ...     print(row.id, row.key, row.value, row.doc)
    """
    path = partitioned_db_resource_parser(
        resource="_design",
        partition=partition,
    )
    return ViewResult(
        **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, c.f the official documentation.

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. Ignored if include_docs isn’t True. Default is None.

descending : bool

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

endkey : Any

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

endkey_docid : str

Stop returning records when the specified document ID is reached. Ignored if endkey is not set. Default is None.

group : bool

Group the results using the reduce function to a group or single row. Implies reduce is true and the maximum group_level. Default is None.

group_level : int

Specify the group level to be used. Implies group is true. Default is None.

include_docs : bool

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

attachments : bool

Include the Base64-encoded content of attachments in the documents that are included if include_docs is True. Ignored if include_docs isn’t True. Default is None.

att_encoding_info : bool

Include encoding information in attachment stubs if include_docs is True and the particular attachment is compressed. Ignored if include_docs isn’t True. Default is False.

inclusive_end : bool

Specifies whether the specified end key should be included in the result. Default is None.

key : str

Return only documents that match the specified key. 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 to the specified number. Default is None.

reduce : bool

Use the reduction function. Default is True when a reduce function is defined. Default is None.

skip : int

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

sort : bool

Sort returned rows (see Sorting Returned Rows). Setting this to False offers a performance boost. The total_rows and offset fields are not available when this is set to False. Default is None.

stable : bool

Whether or not the view results should be returned from a stable set of shards. Default is None.

startkey : Any

Return records starting with the specified key. Default is None

startkey_docid : str

Return records starting with the specified document ID. Ignored if startkey is not set. Default is None

update : str

Whether or not the view in question should be updated prior to responding to the user. Supported values:

• true
• false
• lazy

Default is None.

update_seq : bool

Whether to include in the response an update_seq value indicating the sequence id of the database the view reflects. Default is False.

Returns

view.ViewResult

Examples

>>> import couchdb3
>>> db = couchdb3.Server("http://localhost:5984").get("mydb")
>>> db.view("myddoc", "myview")
ViewResult: {
    'total_rows': 1,
    'offset': 0,
    'rows': [
        {
            'id': 'doc-test-view',
            'key': 'doc-test-view',
            'value': None
        }
    ]
}
>>> result = db.view("my-ddoc", "my-view", include_docs=True, limit=10)
>>> for row in result.rows:
...     print(row.id, row.key, row.value, row.doc)

Inherited members

• Base:
  • check
  • info
  • rev
  • url

class Partition (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.Client | None = None)

Expand source code

class Partition(Database):
    """
    Abstract Couchdb partition
    """

    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.Client | 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 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.Client
            A specific client to use. Optional - if not provided, a new client will be initialized.
        _database : Database
            The owning `Database` instance. Set internally by `Database.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 a `Partition`
            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 `Database` instance this partition was obtained from, or `None` if the partition
        was constructed directly (i.e. not via `Database.get_partition()`).

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

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

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

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

        Parameters
        ----------
        keys : Iterable[str]
            Return only documents where the key matches one of the keys specified in the argument. Default is `None`.
        kwargs
            Further `couchdb3.sync.Database.view` parameters.

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

    # noinspection PyMethodOverriding
    def info(
        self,
    ) -> dict:
        """
        Return the partition's info by sending a `GET` request to `/self.root`.

        Returns
        -------
        Dict: A dictionary containing the server's or database's info.
        """
        return super().info(partition=self.partition_id)

    # noinspection PyMethodOverriding
    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 `Database.find`.
        """
        return 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,
        )

    # noinspection PyMethodOverriding
    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:
        """
        Executes the specified view function from the specified design document, c.f [the official
        documentation](https://docs.couchdb.org/en/main/api/ddoc/views.html#db-design-design-doc-view-view-name).

        Parameters
        ----------
        ddoc : str
            The corresponding design document's id.
        view : str
            The view's id.
        conflicts : bool
            Include conflicts information in response. Ignored if `include_docs` isn’t `True`. Default is `None`.
        descending : bool
            Return the documents in descending order by key. Default is `None`.
        endkey : Any
             Stop returning records when the specified key is reached. Default is `None`
        endkey_docid: str
            Stop returning records when the specified document ID is reached. Ignored if `endkey` is not set. Default
            is `None`.
        group: bool
            Group the results using the reduce function to a group or single row. Implies `reduce` is `true` and the
            maximum `group_level`. Default is `None`.
        group_level : int
             Specify the group level to be used. Implies group is true. Default is `None`.
        include_docs : bool
            Include the associated document with each row. Default is `None`.
        attachments : bool
            Include the Base64-encoded content of attachments in the documents that are included if `include_docs` is
            `True`. Ignored if `include_docs` isn’t `True`. Default is `None`.
        att_encoding_info : bool
            Include encoding information in attachment stubs if `include_docs` is `True` and the particular attachment
            is compressed. Ignored if `include_docs` isn’t `True`. Default is `False`.
        inclusive_end : bool
            Specifies whether the specified end key should be included in the result. Default is `None`.
        key : str
            Return only documents that match the specified key. 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 to the specified number.  Default is `None`.
        reduce : bool
            Use the reduction function. Default is `True` when a reduce function is defined. Default is `None`.
        skip : int
            Skip this number of records before starting to return the results. Default is `None`.
        sort : bool
            Sort returned rows (see Sorting Returned Rows). Setting this to `False` offers a performance boost. The
            `total_rows` and `offset` fields are not available when this is set to `False`. Default is `None`.
        stable : bool
            Whether or not the view results should be returned from a stable set of shards. Default is `None`.
        startkey : Any
            Return records starting with the specified key. Default is `None`
        startkey_docid : Any
            Return records starting with the specified document ID. Ignored if `startkey` is not set. Default is `None`
        update : str
            Whether or not the view in question should be updated prior to responding to the user. Supported values:

            - `true`
            - `false`
            - `lazy`

            Default is `None`.
        update_seq : bool
             Whether to include in the response an `update_seq` value indicating the sequence id of the database the
             view reflects. Default is `False`.

        Returns
        -------
        `view.ViewResult`
        """
        return 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,
        )

    def bulk_docs(self, docs: list[dict | Document], new_edits: bool = True) -> list[dict]:
        """
        See `Database.bulk_docs`.

        Note:
        Appends the partition's ID to the documents' ID.
        """
        return super().bulk_docs(
            docs=[self.add_partition_to_doc(doc) for doc in docs],
            new_edits=new_edits,
        )

    def bulk_get(
        self,
        docs: list[dict | Document],
        revs: bool = False,
    ) -> list[dict]:
        """
        See `Database.bulk_get`.

        Note:
        Appends the partition's ID to the documents' ID.
        """
        return super().bulk_get(
            docs=[self.add_partition_to_bulk_get_doc(doc) for doc in docs],
            revs=revs,
        )

    def copy(
        self,
        docid: str,
        destid: str,
        rev: str | None = None,
        destrev: str | None = None,
    ) -> tuple[str, bool, str]:
        """
        See `Database.copy`.

        Note:
        Appends the partition's ID to the document's ID.
        """
        return super().copy(
            docid=self.add_partition_to_str(docid),
            destid=self.add_partition_to_str(destid),
            rev=rev,
            destrev=destrev,
        )

    def create(
        self,
        doc: dict | Document,
        *,
        batch: bool | None = None,
    ) -> tuple[str, bool, str]:
        """
        See `Database.create`.

        Note:
        Appends the partition's ID to the document's ID.
        """
        return super().create(
            doc=self.add_partition_to_doc(doc),
            batch=batch,
        )

    def delete(
        self,
        docid: str,
        rev: str,
        *,
        batch: bool | None = None,
    ) -> bool:
        """
        See `Database.delete`.

        Note:
        Appends the partition's ID to the document's ID.
        """
        return super().delete(
            docid=self.add_partition_to_str(docid),
            rev=rev,
            batch=batch,
        )

    def delete_attachment(
        self,
        docid: str,
        attname: str,
        rev: str,
        *,
        batch: bool = False,
    ) -> bool:
        """
        See `Database.delete_attachment`.

        Note:
        Appends the partition's ID to the document's ID.
        """
        return super().delete_attachment(
            docid=self.add_partition_to_str(docid),
            attname=attname,
            rev=rev,
            batch=batch,
        )

    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 `Database.get`.

        Note:
        Appends the partition's ID to the document's ID.
        """
        return 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,
        )

    def get_attachment(
        self,
        docid: str,
        attname: str,
        rev: str | None = None,
    ) -> AttachmentDocument:
        """
        See `Database.get_attachment`.

        Note:
        Appends the partition's ID to the document's ID.
        """
        return super().get_attachment(
            docid=self.add_partition_to_str(docid),
            attname=attname,
            rev=rev,
        )

    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 `Database.put_attachment`.

        Note:
        Appends the partition's ID to the document's ID.
        """
        return super().put_attachment(
            docid=self.add_partition_to_str(docid),
            attname=attname,
            content_type=content_type,
            path=path,
            content=content,
            rev=rev,
        )

    def rev(self, resource: str) -> str | None:
        """
        See `Database.rev`.
        """
        return super().rev(self.add_partition_to_str(resource))

    def save(
        self,
        doc: dict | Document,
        batch: bool | None = None,
        new_edits: bool | None = None,
        path: str | None = None,
    ) -> tuple[str, bool, str]:
        """
        See `Database.save`.

        Note:
        Appends the partition's ID to the document's ID.
        """
        return super().save(
            doc=self.add_partition_to_doc(doc),
            batch=batch,
            new_edits=new_edits,
            path=path,
        )

    def __contains__(self, item):
        return super().__contains__(self.add_partition_to_str(item))

    def add_partition_to_str(self, string: str) -> str:
        """
        Append the instance's partition ID to a string.
        """
        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

Abstract Couchdb partition

Parameters

partition_id : str

The partition's ID.

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.Client

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

_database : Database

The owning Database instance. Set internally by Database.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 a Partition directly.

Ancestors

• Database
• Base

Instance variables

prop database

Expand source code

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

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

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

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

Read-only. Setting this attribute raises AttributeError.

Returns

Database | 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 string.startswith(f"{self.partition_id}:"):
        return string
    return f"{self.partition_id}:{string}"

Append the instance's partition ID to a string.

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

Expand source code

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

    Parameters
    ----------
    keys : Iterable[str]
        Return only documents where the key matches one of the keys specified in the argument. Default is `None`.
    kwargs
        Further `couchdb3.sync.Database.view` parameters.

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

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

Parameters

keys : Iterable[str]

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

kwargs

Further Database.view() parameters.

Returns

ViewResult

 

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

Expand source code

def bulk_docs(self, docs: list[dict | Document], new_edits: bool = True) -> list[dict]:
    """
    See `Database.bulk_docs`.

    Note:
    Appends the partition's ID to the documents' ID.
    """
    return super().bulk_docs(
        docs=[self.add_partition_to_doc(doc) for doc in docs],
        new_edits=new_edits,
    )

See Database.bulk_docs().

Note: Appends the partition's ID to the documents' ID.

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

Expand source code

def bulk_get(
    self,
    docs: list[dict | Document],
    revs: bool = False,
) -> list[dict]:
    """
    See `Database.bulk_get`.

    Note:
    Appends the partition's ID to the documents' ID.
    """
    return super().bulk_get(
        docs=[self.add_partition_to_bulk_get_doc(doc) for doc in docs],
        revs=revs,
    )

See Database.bulk_get().

Note: Appends the partition's ID to the documents' ID.

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

Expand source code

def copy(
    self,
    docid: str,
    destid: str,
    rev: str | None = None,
    destrev: str | None = None,
) -> tuple[str, bool, str]:
    """
    See `Database.copy`.

    Note:
    Appends the partition's ID to the document's ID.
    """
    return super().copy(
        docid=self.add_partition_to_str(docid),
        destid=self.add_partition_to_str(destid),
        rev=rev,
        destrev=destrev,
    )

See Database.copy().

Note: Appends the partition's ID to the document's ID.

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

Expand source code

def create(
    self,
    doc: dict | Document,
    *,
    batch: bool | None = None,
) -> tuple[str, bool, str]:
    """
    See `Database.create`.

    Note:
    Appends the partition's ID to the document's ID.
    """
    return super().create(
        doc=self.add_partition_to_doc(doc),
        batch=batch,
    )

See Database.create().

Note: Appends the partition's ID to the document's ID.

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

Expand source code

def delete(
    self,
    docid: str,
    rev: str,
    *,
    batch: bool | None = None,
) -> bool:
    """
    See `Database.delete`.

    Note:
    Appends the partition's ID to the document's ID.
    """
    return super().delete(
        docid=self.add_partition_to_str(docid),
        rev=rev,
        batch=batch,
    )

See Database.delete().

Note: Appends the partition's ID to the document's ID.

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

Expand source code

def delete_attachment(
    self,
    docid: str,
    attname: str,
    rev: str,
    *,
    batch: bool = False,
) -> bool:
    """
    See `Database.delete_attachment`.

    Note:
    Appends the partition's ID to the document's ID.
    """
    return super().delete_attachment(
        docid=self.add_partition_to_str(docid),
        attname=attname,
        rev=rev,
        batch=batch,
    )

See Database.delete_attachment().

Note: Appends the partition's ID to the document's ID.

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

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 `Database.find`.
    """
    return 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 Database.find().

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

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 `Database.get`.

    Note:
    Appends the partition's ID to the document's ID.
    """
    return 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 Database.get().

Note: Appends the partition's ID to the document's ID.

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

Expand source code

def get_attachment(
    self,
    docid: str,
    attname: str,
    rev: str | None = None,
) -> AttachmentDocument:
    """
    See `Database.get_attachment`.

    Note:
    Appends the partition's ID to the document's ID.
    """
    return super().get_attachment(
        docid=self.add_partition_to_str(docid),
        attname=attname,
        rev=rev,
    )

See Database.get_attachment().

Note: Appends the partition's ID to the document's ID.

def info(self) ‑> dict

Expand source code

def info(
    self,
) -> dict:
    """
    Return the partition's info by sending a `GET` request to `/self.root`.

    Returns
    -------
    Dict: A dictionary containing the server's or database's info.
    """
    return super().info(partition=self.partition_id)

Return the partition's info by sending a GET request to /self.root.

Returns

Dict: A dictionary containing the server's or database's info.

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

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 `Database.put_attachment`.

    Note:
    Appends the partition's ID to the document's ID.
    """
    return super().put_attachment(
        docid=self.add_partition_to_str(docid),
        attname=attname,
        content_type=content_type,
        path=path,
        content=content,
        rev=rev,
    )

See Database.put_attachment().

Note: Appends the partition's ID to the document's ID.

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

Expand source code

def rev(self, resource: str) -> str | None:
    """
    See `Database.rev`.
    """
    return super().rev(self.add_partition_to_str(resource))

See Base.rev().

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

def save(
    self,
    doc: dict | Document,
    batch: bool | None = None,
    new_edits: bool | None = None,
    path: str | None = None,
) -> tuple[str, bool, str]:
    """
    See `Database.save`.

    Note:
    Appends the partition's ID to the document's ID.
    """
    return super().save(
        doc=self.add_partition_to_doc(doc),
        batch=batch,
        new_edits=new_edits,
        path=path,
    )

See Database.save().

Note: Appends the partition's ID to the document's ID.

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

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:
    """
    Executes the specified view function from the specified design document, c.f [the official
    documentation](https://docs.couchdb.org/en/main/api/ddoc/views.html#db-design-design-doc-view-view-name).

    Parameters
    ----------
    ddoc : str
        The corresponding design document's id.
    view : str
        The view's id.
    conflicts : bool
        Include conflicts information in response. Ignored if `include_docs` isn’t `True`. Default is `None`.
    descending : bool
        Return the documents in descending order by key. Default is `None`.
    endkey : Any
         Stop returning records when the specified key is reached. Default is `None`
    endkey_docid: str
        Stop returning records when the specified document ID is reached. Ignored if `endkey` is not set. Default
        is `None`.
    group: bool
        Group the results using the reduce function to a group or single row. Implies `reduce` is `true` and the
        maximum `group_level`. Default is `None`.
    group_level : int
         Specify the group level to be used. Implies group is true. Default is `None`.
    include_docs : bool
        Include the associated document with each row. Default is `None`.
    attachments : bool
        Include the Base64-encoded content of attachments in the documents that are included if `include_docs` is
        `True`. Ignored if `include_docs` isn’t `True`. Default is `None`.
    att_encoding_info : bool
        Include encoding information in attachment stubs if `include_docs` is `True` and the particular attachment
        is compressed. Ignored if `include_docs` isn’t `True`. Default is `False`.
    inclusive_end : bool
        Specifies whether the specified end key should be included in the result. Default is `None`.
    key : str
        Return only documents that match the specified key. 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 to the specified number.  Default is `None`.
    reduce : bool
        Use the reduction function. Default is `True` when a reduce function is defined. Default is `None`.
    skip : int
        Skip this number of records before starting to return the results. Default is `None`.
    sort : bool
        Sort returned rows (see Sorting Returned Rows). Setting this to `False` offers a performance boost. The
        `total_rows` and `offset` fields are not available when this is set to `False`. Default is `None`.
    stable : bool
        Whether or not the view results should be returned from a stable set of shards. Default is `None`.
    startkey : Any
        Return records starting with the specified key. Default is `None`
    startkey_docid : Any
        Return records starting with the specified document ID. Ignored if `startkey` is not set. Default is `None`
    update : str
        Whether or not the view in question should be updated prior to responding to the user. Supported values:

        - `true`
        - `false`
        - `lazy`

        Default is `None`.
    update_seq : bool
         Whether to include in the response an `update_seq` value indicating the sequence id of the database the
         view reflects. Default is `False`.

    Returns
    -------
    `view.ViewResult`
    """
    return 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,
    )

Executes the specified view function from the specified design document, c.f the official documentation.

Parameters

ddoc : str

The corresponding design document's id.

view : str

The view's id.

conflicts : bool

Include conflicts information in response. Ignored if include_docs isn’t True. Default is None.

descending : bool

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

endkey : Any

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

endkey_docid : str

Stop returning records when the specified document ID is reached. Ignored if endkey is not set. Default is None.

group : bool

Group the results using the reduce function to a group or single row. Implies reduce is true and the maximum group_level. Default is None.

group_level : int

Specify the group level to be used. Implies group is true. Default is None.

include_docs : bool

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

attachments : bool

Include the Base64-encoded content of attachments in the documents that are included if include_docs is True. Ignored if include_docs isn’t True. Default is None.

att_encoding_info : bool

Include encoding information in attachment stubs if include_docs is True and the particular attachment is compressed. Ignored if include_docs isn’t True. Default is False.

inclusive_end : bool

Specifies whether the specified end key should be included in the result. Default is None.

key : str

Return only documents that match the specified key. 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 to the specified number. Default is None.

reduce : bool

Use the reduction function. Default is True when a reduce function is defined. Default is None.

skip : int

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

sort : bool

Sort returned rows (see Sorting Returned Rows). Setting this to False offers a performance boost. The total_rows and offset fields are not available when this is set to False. Default is None.

stable : bool

Whether or not the view results should be returned from a stable set of shards. Default is None.

startkey : Any

Return records starting with the specified key. Default is None

startkey_docid : Any

Return records starting with the specified document ID. Ignored if startkey is not set. Default is None

update : str

Whether or not the view in question should be updated prior to responding to the user. Supported values:

• true
• false
• lazy

Default is None.

update_seq : bool

Whether to include in the response an update_seq value indicating the sequence id of the database the view reflects. Default is False.

Returns

view.ViewResult

Inherited members

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

class Server (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.Client | None = None)

Expand source code

class Server(Base):
    """
    Abstract Couchdb client
    """

    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.Client | 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.Client
            A specific 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 __getitem__(self, item) -> Database:
        return self.get(item, check=True)

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

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

    def active_tasks(
        self,
    ) -> list[dict]:
        """
        List of running tasks, including the task type, name, status and process ID. The result is a JSON array of the
        currently running tasks, with each task being described with a single object. Depending on operation type set
        of response object fields might be different.

        Returns
        -------
        List[Dict]
        """
        return self._get(resource="_active_tasks").json()

    def check_user(self, username: str, password: str) -> bool:
        """
        Checks the username/password combination by creating a `Server` instance and performing a `Server.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.
        """
        return Server(url=self.url, user=username, password=password).check()

    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 e.g. you cannot rename an existing user - you have to create new one.
        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. CouchDB doesn’t provide any built-in roles, so you’re free to define your own depending
            on your needs. However, you cannot set system roles like `_admin` there. Also, only administrators may
            assign roles to users - by default all users have no roles.
        password : str
            A plaintext password can be provided, but will be replaced by hashed fields before the document is actually
            stored.
        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. Used for both `simple` and `pbkdf2` `password_scheme` options.
        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]: A tuple consisting of the following elements.

          - the success status (`bool`)
          - the user ID ( `str`)
          - the current revision ( `str`)
        """
        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 = self._put(resource=f"_users/{user_id}", body=body)
        except ConflictError:
            body.update({"_rev": self.rev(f"_users/{user_id}")})
            response = self._put(resource=f"_users/{user_id}", body=body)
        data = response.json()
        return data["ok"], data["id"], data["rev"]

    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 to the specified number. Default `None`.
        skip : int
            Skip this number of databases before starting to return the results. Default `0`.
        startkey : str
            Return databases starting with the specified key. Default `None`.

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

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

        Parameters
        ----------
        name : str
            The database's name.
        q : int
            Shards, aka the number of range partitions. Default `None` (i.e. server default will be used: `8`, unless
            overridden in the `cluster config`).
        n : int
            Replicas. The number of copies of the database in the cluster. Default `None` (i.e. server default will be
            used: `3`, unless overridden in the `cluster config`).
        partitioned : bool
            Whether to create a partitioned database. Default `False`.

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

    def dbs_info(self, keys: list[str]) -> list[dict]:
        """
        Returns information of a list of the specified databases in the CouchDB instance.

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

        Returns
        -------
        List[Dict] : A list dictionaries containing the corresponding database info.
        """
        return self._post(resource="_dbs_info", body={"keys": keys}).json()

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

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

        Returns
        -------
        couchdb3.sync.Database

        """
        db = Database(
            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,
            _server=self,
        )
        try:
            db._head()
        except (NotFoundError, httpx.RequestError):
            if check is True:
                raise
        except CouchDBError:
            raise
        return db

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

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

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

    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 : Union[Dict, str]
            Fully qualified source database URL or an object which contains the full URL of the source database with
            additional parameters like headers. Eg:

                "http://example.com/source_db_name"

            or

                {“url”:”url in here”, “headers”: {“header1”:”value1”, …}}

        target : Union[Dict, str]
            Fully qualified target database URL or an object which contains the full URL of the source database with
            additional parameters like headers. Eg:

                "http://example.com/target_db_name"

            or

                {“url”:”url in here”, “headers”: {“header1”:”value1”, …}}

        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. Required administrator’s privileges on target server.
        create_target_params : Dict
            An object that contains parameters to be used when creating the target database. Can include the standard
            `q` and `n` parameters.
        doc_ids : List[str]
            Array of document IDs to be synchronized. `doc_ids`, `filter` and `selector` are mutually exclusive.
        filter_func : str
             The name of a [filter function](https://docs.couchdb.org/en/main/ddocs/ddocs.html#filterfun).
             `doc_ids`, `filter` and `selector` are mutually exclusive.
        selector : Dict
            A [selector](https://docs.couchdb.org/en/main/api/database/find.html#find-selectors) to filter documents
            for synchronization. Has the same behavior as the
            [selector objects](https://docs.couchdb.org/en/main/replication/replicator.html#selectorobj) in replication
            documents. `doc_ids`, `filter` and `selector` are mutually exclusive.
        source_proxy : str
            Address of a proxy server through which replication from the source should occur (protocol can be `"http”`
            or `“socks5”`).
        target_proxy : str
            Address of a proxy server through which replication to the target should occur (protocol can be `"http”`
            or `“socks5”`).

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

          - history (`list`) - Replication history
          - ok (`bool`) - Replication status
          - replication_id_version (`int`) – Replication protocol version
          - session_id (`str`) – Unique session ID
          - source_last_seq (`int`) – Last sequence number read from source database
        """
        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 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()

    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 self._get(resource="_membership").json()

    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 self._get(
            resource="_cluster_setup",
            query_kwargs={"ensure_dbs_exist": ensure_dbs_exist},
        ).json()

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

    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 self._get(resource=resource).json()

    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 self._put(
            resource=f"_node/{node}/_config/{section}/{key}",
            body=value,
        ).json()

    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 self._delete(
            resource=f"_node/{node}/_config/{section}/{key}",
        ).json()

    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 (
            self._post(
                resource=f"_node/{node}/_config/_reload",
                body={},
            )
            .json()
            .get("ok", False)
        )

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

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

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

    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 self._get(resource=f"_node/{node}/_system").json()

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

        Parameters
        ----------
        raise_exception : bool
            If `True`, exceptions encountered when check server will be raised.

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

Abstract Couchdb client

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.Client

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

Ancestors

• Base

Methods

def active_tasks(self) ‑> list[dict]

Expand source code

def active_tasks(
    self,
) -> list[dict]:
    """
    List of running tasks, including the task type, name, status and process ID. The result is a JSON array of the
    currently running tasks, with each task being described with a single object. Depending on operation type set
    of response object fields might be different.

    Returns
    -------
    List[Dict]
    """
    return self._get(resource="_active_tasks").json()

List of running tasks, including the task type, name, status and process ID. The result is a JSON array of the currently running tasks, with each task being described with a single object. Depending on operation type set of response object fields might be different.

Returns

List[Dict]

 

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

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 to the specified number. Default `None`.
    skip : int
        Skip this number of databases before starting to return the results. Default `0`.
    startkey : str
        Return databases starting with the specified key. Default `None`.

    Returns
    -------
    List[str] : A list of database names.
    """
    return 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 to the specified number. Default None.

skip : int

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

startkey : str

Return databases starting with the specified key. Default None.

Returns

List[str] : A list of database names.

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

Expand source code

def check_user(self, username: str, password: str) -> bool:
    """
    Checks the username/password combination by creating a `Server` instance and performing a `Server.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.
    """
    return Server(url=self.url, user=username, password=password).check()

Checks the username/password combination by creating a Server instance and performing a Base.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.

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

Expand source code

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 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'.

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

Expand source code

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

    Parameters
    ----------
    name : str
        The database's name.
    q : int
        Shards, aka the number of range partitions. Default `None` (i.e. server default will be used: `8`, unless
        overridden in the `cluster config`).
    n : int
        Replicas. The number of copies of the database in the cluster. Default `None` (i.e. server default will be
        used: `3`, unless overridden in the `cluster config`).
    partitioned : bool
        Whether to create a partitioned database. Default `False`.

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

Create a database.

Parameters

name : str

The database's name.

q : int

Shards, aka the number of range partitions. Default None (i.e. server default will be used: 8, unless overridden in the cluster config).

n : int

Replicas. The number of copies of the database in the cluster. Default None (i.e. server default will be used: 3, unless overridden in the cluster config).

partitioned : bool

Whether to create a partitioned database. Default False.

Returns

Database

 

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

Expand source code

def dbs_info(self, keys: list[str]) -> list[dict]:
    """
    Returns information of a list of the specified databases in the CouchDB instance.

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

    Returns
    -------
    List[Dict] : A list dictionaries containing the corresponding database info.
    """
    return self._post(resource="_dbs_info", body={"keys": keys}).json()

Returns information of a list of the specified databases in the CouchDB instance.

Parameters

keys : List[str]

List of database names to be requested

Returns

List[Dict] : A list dictionaries containing the corresponding database info.

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

Expand source code

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

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

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

Delete a database.

Parameters

resource : str

The database's name.

Returns

bool: True upon successful deletion.

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

Expand source code

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 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.

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

Expand source code

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

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

    Returns
    -------
    couchdb3.sync.Database

    """
    db = Database(
        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,
        _server=self,
    )
    try:
        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 database name cannot be found in the server. Default False.

Returns

Database

 

def membership(self) ‑> dict

Expand source code

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

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

Expand source code

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

 

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

Expand source code

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

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

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

Returns statistics for the specified node.

Parameters

node : str

Node name. Default '_local'.

Returns

dict

 

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

Expand source code

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

 

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

Expand source code

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

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

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 : Union[Dict, str]
        Fully qualified source database URL or an object which contains the full URL of the source database with
        additional parameters like headers. Eg:

            "http://example.com/source_db_name"

        or

            {“url”:”url in here”, “headers”: {“header1”:”value1”, …}}

    target : Union[Dict, str]
        Fully qualified target database URL or an object which contains the full URL of the source database with
        additional parameters like headers. Eg:

            "http://example.com/target_db_name"

        or

            {“url”:”url in here”, “headers”: {“header1”:”value1”, …}}

    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. Required administrator’s privileges on target server.
    create_target_params : Dict
        An object that contains parameters to be used when creating the target database. Can include the standard
        `q` and `n` parameters.
    doc_ids : List[str]
        Array of document IDs to be synchronized. `doc_ids`, `filter` and `selector` are mutually exclusive.
    filter_func : str
         The name of a [filter function](https://docs.couchdb.org/en/main/ddocs/ddocs.html#filterfun).
         `doc_ids`, `filter` and `selector` are mutually exclusive.
    selector : Dict
        A [selector](https://docs.couchdb.org/en/main/api/database/find.html#find-selectors) to filter documents
        for synchronization. Has the same behavior as the
        [selector objects](https://docs.couchdb.org/en/main/replication/replicator.html#selectorobj) in replication
        documents. `doc_ids`, `filter` and `selector` are mutually exclusive.
    source_proxy : str
        Address of a proxy server through which replication from the source should occur (protocol can be `"http”`
        or `“socks5”`).
    target_proxy : str
        Address of a proxy server through which replication to the target should occur (protocol can be `"http”`
        or `“socks5”`).

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

      - history (`list`) - Replication history
      - ok (`bool`) - Replication status
      - replication_id_version (`int`) – Replication protocol version
      - session_id (`str`) – Unique session ID
      - source_last_seq (`int`) – Last sequence number read from source database
    """
    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 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 : Union[Dict, str]

Fully qualified source database URL or an object which contains the full URL of the source database with additional parameters like headers. Eg:

"http://example.com/source_db_name"

or

{“url”:”url in here”, “headers”: {“header1”:”value1”, …}}

target : Union[Dict, str]

Fully qualified target database URL or an object which contains the full URL of the source database with additional parameters like headers. Eg:

"http://example.com/target_db_name"

or

{“url”:”url in here”, “headers”: {“header1”:”value1”, …}}

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. Required administrator’s privileges on target server.

create_target_params : Dict

An object that contains parameters to be used when creating the target database. Can include the standard q and n parameters.

doc_ids : List[str]

Array of document IDs to be synchronized. doc_ids, filter and selector are mutually exclusive.

filter_func : str

The name of a filter function. doc_ids, filter and selector are mutually exclusive.

selector : Dict

A selector to filter documents for synchronization. Has the same behavior as the selector objects in replication documents. doc_ids, filter and selector are mutually exclusive.

source_proxy : str

Address of a proxy server through which replication from the source should occur (protocol can be "http” or “socks5”).

target_proxy : str

Address of a proxy server through which replication to the target should occur (protocol can be "http” or “socks5”).

Returns

Dict : A dictionary with the following keys.

• history (list) - Replication history
• ok (bool) - Replication status
• replication_id_version (int) – Replication protocol version
• session_id (str) – Unique session ID
• source_last_seq (int) – Last sequence number read from source database

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

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 e.g. you cannot rename an existing user - you have to create new one.
    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. CouchDB doesn’t provide any built-in roles, so you’re free to define your own depending
        on your needs. However, you cannot set system roles like `_admin` there. Also, only administrators may
        assign roles to users - by default all users have no roles.
    password : str
        A plaintext password can be provided, but will be replaced by hashed fields before the document is actually
        stored.
    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. Used for both `simple` and `pbkdf2` `password_scheme` options.
    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]: A tuple consisting of the following elements.

      - the success status (`bool`)
      - the user ID ( `str`)
      - the current revision ( `str`)
    """
    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 = self._put(resource=f"_users/{user_id}", body=body)
    except ConflictError:
        body.update({"_rev": self.rev(f"_users/{user_id}")})
        response = 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 e.g. you cannot rename an existing user - you have to create new one.

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. CouchDB doesn’t provide any built-in roles, so you’re free to define your own depending on your needs. However, you cannot set system roles like _admin there. Also, only administrators may assign roles to users - by default all users have no roles.

password : str

A plaintext password can be provided, but will be replaced by hashed fields before the document is actually stored.

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. Used for both simple and pbkdf2 password_scheme options.

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]: A tuple consisting of the following elements.

• the success status (bool)
• the user ID ( str)
• the current revision ( str)

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

Expand source code

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 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.

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

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 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.

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

Expand source code

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

    Parameters
    ----------
    raise_exception : bool
        If `True`, exceptions encountered when check server will be raised.

    Returns
    -------
    bool : `True` if the server is up.
    """
    try:
        response = 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 encountered when check server will be raised.

Returns

bool : True if the server is up.

Inherited members

• Base:
  • check
  • info
  • rev
  • url