Module astrapy.cursors

Classes

class AsyncCommandCursor (address: str, items: list[T])

A (asynchronous) cursor over the results of a Data API command (as opposed to a cursor over data as one would get with a find method).

Asynchronous command cursors are iterated over, e.g. with an async for loop.

Generally command cursors are not supposed to be instantiated directly, rather they are obtained by invoking methods on a collection/database (such as the database list_collections method).

Expand source code

class AsyncCommandCursor(Generic[T]):
    """
    A (asynchronous) cursor over the results of a Data API command
    (as opposed to a cursor over data as one would get with a `find` method).

    Asynchronous command cursors are iterated over, e.g. with an async for loop.

    Generally command cursors are not supposed to be instantiated directly,
    rather they are obtained by invoking methods on a collection/database
    (such as the database `list_collections` method).
    """

    def __init__(self, address: str, items: list[T]) -> None:
        self._address = address
        self.items = items
        self.iterable = items.__iter__()
        self._alive = True

    def __repr__(self) -> str:
        return f'{self.__class__.__name__}("{self.address}", ' f"{self.state})"

    def __aiter__(self) -> AsyncCommandCursor[T]:
        self._ensure_alive()
        return self

    async def __anext__(self) -> T:
        try:
            item = self.iterable.__next__()
            return item
        except StopIteration:
            self._alive = False
            raise StopAsyncIteration

    @property
    def state(self) -> str:
        """
        The current state of this cursor, which can be:
            - "alive": the cursor has still the potential to return items.
            - "exhausted": the cursor has finished and won't return documents
        """

        return "alive" if self._alive else "exhausted"

    @property
    def address(self) -> str:
        """
        The API endpoint used by this cursor when issuing
        requests to the database.
        """

        return self._address

    @property
    def alive(self) -> bool:
        """
        Whether the cursor has the potential to yield more data.
        """

        return self._alive

    @property
    def cursor_id(self) -> int:
        """
        An integer uniquely identifying this cursor.
        """

        return id(self)

    def _ensure_alive(self) -> None:
        if not self._alive:
            raise CursorIsStartedException(
                text="Cursor is closed.",
                cursor_state=self.state,
            )

    def close(self) -> None:
        """
        Stop/kill the cursor, regardless of its status.
        """

        self._alive = False

Ancestors

• typing.Generic

Instance variables

var address : str

The API endpoint used by this cursor when issuing requests to the database.

Expand source code

@property
def address(self) -> str:
    """
    The API endpoint used by this cursor when issuing
    requests to the database.
    """

    return self._address

var alive : bool

Whether the cursor has the potential to yield more data.

Expand source code

@property
def alive(self) -> bool:
    """
    Whether the cursor has the potential to yield more data.
    """

    return self._alive

var cursor_id : int

An integer uniquely identifying this cursor.

Expand source code

@property
def cursor_id(self) -> int:
    """
    An integer uniquely identifying this cursor.
    """

    return id(self)

var state : str

The current state of this cursor, which can be: - "alive": the cursor has still the potential to return items. - "exhausted": the cursor has finished and won't return documents

Expand source code

@property
def state(self) -> str:
    """
    The current state of this cursor, which can be:
        - "alive": the cursor has still the potential to return items.
        - "exhausted": the cursor has finished and won't return documents
    """

    return "alive" if self._alive else "exhausted"

Methods

def close(self) ‑> None

Stop/kill the cursor, regardless of its status.

Expand source code

def close(self) -> None:
    """
    Stop/kill the cursor, regardless of its status.
    """

    self._alive = False

class AsyncCursor (collection: AsyncCollection, filter: dict[str, Any] | None, projection: ProjectionType | None, max_time_ms: int | None, overall_max_time_ms: int | None)

Represents a (asynchronous) cursor over documents in a collection. An asynchronous cursor is iterated over, e.g. with a for loop, and keeps track of its progress.

Generally cursors are not supposed to be instantiated directly, rather they are obtained by invoking the find method on a collection.

Attributes

collection

the collection to find documents in filter: a predicate expressed as a dictionary according to the Data API filter syntax. Examples are: {} {"name": "John"} {"price": {"$le": 100}} {"$and": [{"name": "John"}, {"price": {"$le": 100}}]} See the Data API documentation for the full set of operators. projection: used to select a subset of fields in the document being returned. The projection can be: an iterable over the field names to return; a dictionary {field_name: True} to positively select certain fields; or a dictionary {field_name: False} if one wants to discard some fields from the response. The default is to return the whole documents. max_time_ms: a timeout, in milliseconds, for each single one of the underlying HTTP requests used to fetch documents as the cursor is iterated over.

Note

When not specifying sorting criteria at all (by vector or otherwise), the cursor can scroll through an arbitrary number of documents as the Data API and the client periodically exchange new chunks of documents. It should be noted that the behavior of the cursor in the case documents have been added/removed after the cursor was started depends on database internals and it is not guaranteed, nor excluded, that such "real-time" changes in the data would be picked up by the cursor.

Expand source code

class AsyncCursor(BaseCursor):
    """
    Represents a (asynchronous) cursor over documents in a collection.
    An asynchronous cursor is iterated over, e.g. with a for loop,
    and keeps track of its progress.

    Generally cursors are not supposed to be instantiated directly,
    rather they are obtained by invoking the `find` method on a collection.

    Attributes:
        collection: the collection to find documents in
            filter: a predicate expressed as a dictionary according to the
                Data API filter syntax. Examples are:
                    {}
                    {"name": "John"}
                    {"price": {"$le": 100}}
                    {"$and": [{"name": "John"}, {"price": {"$le": 100}}]}
                See the Data API documentation for the full set of operators.
            projection: used to select a subset of fields in the document being
                returned. The projection can be: an iterable over the field names
                to return; a dictionary {field_name: True} to positively select
                certain fields; or a dictionary {field_name: False} if one wants
                to discard some fields from the response.
                The default is to return the whole documents.
            max_time_ms: a timeout, in milliseconds, for each single one
                of the underlying HTTP requests used to fetch documents as the
                cursor is iterated over.

    Note:
        When not specifying sorting criteria at all (by vector or otherwise),
        the cursor can scroll through an arbitrary number of documents as
        the Data API and the client periodically exchange new chunks of documents.
        It should be noted that the behavior of the cursor in the case documents
        have been added/removed after the cursor was started depends on database
        internals and it is not guaranteed, nor excluded, that such "real-time"
        changes in the data would be picked up by the cursor.
    """

    def __init__(
        self,
        collection: AsyncCollection,
        filter: dict[str, Any] | None,
        projection: ProjectionType | None,
        max_time_ms: int | None,
        overall_max_time_ms: int | None,
    ) -> None:
        self._collection: AsyncCollection = collection
        self._filter = filter
        self._projection = projection
        self._overall_max_time_ms = overall_max_time_ms
        if overall_max_time_ms is not None and max_time_ms is not None:
            self._max_time_ms = min(max_time_ms, overall_max_time_ms)
        else:
            self._max_time_ms = max_time_ms
        self._limit: int | None = None
        self._skip: int | None = None
        self._include_similarity: bool | None = None
        self._include_sort_vector: bool | None = None
        self._sort: dict[str, Any] | None = None
        self._state = CursorState.IDLE
        self._consumed = 0
        #
        self._iterator: _AsyncLookAheadIterator | None = None
        self._api_response_status: dict[str, Any] | None = None

    def __aiter__(self) -> AsyncCursor:
        self._ensure_alive()
        if self._iterator is None:
            self._iterator = self._create_iterator()
            self._state = CursorState.STARTED
        return self

    async def __anext__(self) -> DocumentType:
        if not self.alive:
            # keep raising once exhausted:
            raise StopAsyncIteration
        if self._iterator is None:
            self._iterator = self._create_iterator()
            self._state = CursorState.STARTED
        # check for overall timing out
        if self._overall_max_time_ms is not None:
            _elapsed = time.time() - self._started_time_s  # type: ignore[operator]
            if _elapsed > (self._overall_max_time_ms / 1000.0):
                raise DataAPITimeoutException(
                    text="Cursor timed out.",
                    timeout_type="generic",
                    endpoint=None,
                    raw_payload=None,
                )
        try:
            next_item = await self._iterator.__anext__()
            self._consumed = self._consumed + 1
            return next_item
        except StopAsyncIteration:
            self.close()
            raise

    async def get_sort_vector(self) -> list[float] | None:
        """
        Return the vector used in this ANN search, if applicable.
        If this is not an ANN search, or it was invoked without the
        `include_sort_vector` parameter, return None.

        Invoking this method on a pristine cursor will trigger an API call
        to get the first page of results.
        """

        if self._iterator is None:
            self._iterator = self._create_iterator()
            self._state = CursorState.STARTED
        await self._iterator.preread()
        if self._api_response_status:
            return self._api_response_status.get("sortVector")
        else:
            return None

    def _item_at_index(self, index: int) -> DocumentType:
        finder_cursor = self._to_sync().skip(index).limit(1)
        items = list(finder_cursor)
        if items:
            return items[0]
        else:
            raise IndexError("no such item for AsyncCursor instance")

    def _create_iterator(self) -> _AsyncLookAheadIterator:
        self._ensure_idle()
        self._ensure_alive()
        _options_0 = {
            k: v
            for k, v in {
                "limit": self._limit,
                "skip": self._skip,
                "includeSimilarity": self._include_similarity,
                "includeSortVector": self._include_sort_vector,
            }.items()
            if v is not None
        }
        _projection = normalize_optional_projection(self._projection)
        _sort = self._sort or {}
        _filter = self._filter or {}
        f0_payload = {
            "find": {
                k: v
                for k, v in {
                    "filter": _filter,
                    "projection": _projection,
                    "options": _options_0,
                    "sort": _sort,
                }.items()
                if v
            }
        }

        async def _find_iterator() -> AsyncIterator[DocumentType]:
            resp_0 = await self._collection.command(
                body=f0_payload,
                max_time_ms=self._max_time_ms,
            )
            self._api_response_status = resp_0.get("status")
            if "nextPageState" not in resp_0.get("data", {}):
                raise DataAPIFaultyResponseException(
                    text="Faulty response from find API command (no 'nextPageState').",
                    raw_response=resp_0,
                )
            next_page_state = resp_0["data"]["nextPageState"]
            if "documents" not in resp_0.get("data", {}):
                raise DataAPIFaultyResponseException(
                    text="Faulty response from find API command (no 'documents').",
                    raw_response=resp_0,
                )
            for doc in resp_0["data"]["documents"]:
                yield doc
            while next_page_state is not None:
                _options_n = {**_options_0, **{"pageState": next_page_state}}
                fn_payload = {
                    "find": {
                        k: v
                        for k, v in {
                            "filter": _filter,
                            "projection": _projection,
                            "options": _options_n,
                            "sort": _sort,
                        }.items()
                        if v
                    }
                }
                resp_n = await self._collection.command(
                    body=fn_payload,
                    max_time_ms=self._max_time_ms,
                )
                self._api_response_status = resp_n.get("status")
                if "nextPageState" not in resp_n.get("data", {}):
                    raise DataAPIFaultyResponseException(
                        text="Faulty response from find API command (no 'nextPageState').",
                        raw_response=resp_n,
                    )
                next_page_state = resp_n["data"]["nextPageState"]
                if "documents" not in resp_n.get("data", {}):
                    raise DataAPIFaultyResponseException(
                        text="Faulty response from find API command (no 'documents').",
                        raw_response=resp_n,
                    )
                for doc in resp_n["data"]["documents"]:
                    yield doc

        logger.info(f"creating iterator on '{self._collection.name}'")
        iterator = _find_iterator()
        logger.info(f"finished creating iterator on '{self._collection.name}'")
        self._started_time_s = time.time()
        return _AsyncLookAheadIterator(iterator)

    def _to_sync(
        self: AsyncCursor,
        *,
        limit: int | None = None,
        skip: int | None = None,
        include_similarity: bool | None = None,
        include_sort_vector: bool | None = None,
        sort: dict[str, Any] | None = None,
    ) -> Cursor:
        new_cursor = Cursor(
            collection=self._collection.to_sync(),
            filter=self._filter,
            projection=self._projection,
            max_time_ms=self._max_time_ms,
            overall_max_time_ms=self._overall_max_time_ms,
        )
        # Cursor treated as mutable within this function scope:
        new_cursor._limit = limit if limit is not None else self._limit
        new_cursor._skip = skip if skip is not None else self._skip
        new_cursor._include_similarity = (
            include_similarity
            if include_similarity is not None
            else self._include_similarity
        )
        new_cursor._include_sort_vector = (
            include_sort_vector
            if include_sort_vector is not None
            else self._include_sort_vector
        )
        new_cursor._sort = sort if sort is not None else self._sort
        return new_cursor

    @property
    def collection(self) -> AsyncCollection:
        """
        The (asynchronous) collection this cursor is targeting.
        """

        the_warning = deprecation.DeprecatedWarning(
            "the 'collection' property of Cursor objects",
            deprecated_in="1.5.1",
            removed_in="2.0.0",
            details="Please use the 'data_source' property",
        )
        warnings.warn(the_warning, stacklevel=2)

        return self.data_source

    @property
    def data_source(self) -> AsyncCollection:
        """
        The (asynchronous) collection this cursor is targeting.
        """

        return self._collection

    async def distinct(self, key: str, max_time_ms: int | None = None) -> list[Any]:
        """
        Compute a list of unique values for a specific field across all
        documents the cursor iterates through.

        Invoking this method has no effect on the cursor state, i.e.
        the position of the cursor is unchanged.

        Args:
            key: the name of the field whose value is inspected across documents.
                Keys can use dot-notation to descend to deeper document levels.
                Example of acceptable `key` values:
                    "field"
                    "field.subfield"
                    "field.3"
                    "field.3.subfield"
                if lists are encountered and no numeric index is specified,
                all items in the list are visited.
            max_time_ms: a timeout, in milliseconds, for the operation.

        Note:
            this operation works at client-side by scrolling through all
            documents matching the cursor parameters (such as `filter`).
            Please be aware of this fact, especially for a very large
            amount of documents, for this may have implications on latency,
            network traffic and possibly billing.
        """

        _item_hashes = set()
        distinct_items = []

        _extractor = _create_document_key_extractor(key)
        _key = _reduce_distinct_key_to_safe(key)

        d_cursor = self._copy(
            projection={_key: True},
            overall_max_time_ms=max_time_ms,
        )
        logger.info(f"running distinct() on '{self._collection.name}'")
        async for document in d_cursor:
            for item in _extractor(document):
                _item_hash = _hash_document(item)
                if _item_hash not in _item_hashes:
                    _item_hashes.add(_item_hash)
                    distinct_items.append(item)

        logger.info(f"finished running distinct() on '{self._collection.name}'")
        return distinct_items

Ancestors

• BaseCursor

Instance variables

var collection : AsyncCollection

The (asynchronous) collection this cursor is targeting.

Expand source code

@property
def collection(self) -> AsyncCollection:
    """
    The (asynchronous) collection this cursor is targeting.
    """

    the_warning = deprecation.DeprecatedWarning(
        "the 'collection' property of Cursor objects",
        deprecated_in="1.5.1",
        removed_in="2.0.0",
        details="Please use the 'data_source' property",
    )
    warnings.warn(the_warning, stacklevel=2)

    return self.data_source

var data_source : AsyncCollection

The (asynchronous) collection this cursor is targeting.

Expand source code

@property
def data_source(self) -> AsyncCollection:
    """
    The (asynchronous) collection this cursor is targeting.
    """

    return self._collection

Methods

async def distinct(self, key: str, max_time_ms: int | None = None) ‑> list[typing.Any]

Compute a list of unique values for a specific field across all documents the cursor iterates through.

Invoking this method has no effect on the cursor state, i.e. the position of the cursor is unchanged.

Args

key

the name of the field whose value is inspected across documents. Keys can use dot-notation to descend to deeper document levels. Example of acceptable key values: "field" "field.subfield" "field.3" "field.3.subfield" if lists are encountered and no numeric index is specified, all items in the list are visited.

max_time_ms

a timeout, in milliseconds, for the operation.

Note

this operation works at client-side by scrolling through all documents matching the cursor parameters (such as filter). Please be aware of this fact, especially for a very large amount of documents, for this may have implications on latency, network traffic and possibly billing.

Expand source code

async def distinct(self, key: str, max_time_ms: int | None = None) -> list[Any]:
    """
    Compute a list of unique values for a specific field across all
    documents the cursor iterates through.

    Invoking this method has no effect on the cursor state, i.e.
    the position of the cursor is unchanged.

    Args:
        key: the name of the field whose value is inspected across documents.
            Keys can use dot-notation to descend to deeper document levels.
            Example of acceptable `key` values:
                "field"
                "field.subfield"
                "field.3"
                "field.3.subfield"
            if lists are encountered and no numeric index is specified,
            all items in the list are visited.
        max_time_ms: a timeout, in milliseconds, for the operation.

    Note:
        this operation works at client-side by scrolling through all
        documents matching the cursor parameters (such as `filter`).
        Please be aware of this fact, especially for a very large
        amount of documents, for this may have implications on latency,
        network traffic and possibly billing.
    """

    _item_hashes = set()
    distinct_items = []

    _extractor = _create_document_key_extractor(key)
    _key = _reduce_distinct_key_to_safe(key)

    d_cursor = self._copy(
        projection={_key: True},
        overall_max_time_ms=max_time_ms,
    )
    logger.info(f"running distinct() on '{self._collection.name}'")
    async for document in d_cursor:
        for item in _extractor(document):
            _item_hash = _hash_document(item)
            if _item_hash not in _item_hashes:
                _item_hashes.add(_item_hash)
                distinct_items.append(item)

    logger.info(f"finished running distinct() on '{self._collection.name}'")
    return distinct_items

async def get_sort_vector(self) ‑> list[float] | None

Return the vector used in this ANN search, if applicable. If this is not an ANN search, or it was invoked without the include_sort_vector parameter, return None.

Invoking this method on a pristine cursor will trigger an API call to get the first page of results.

Expand source code

async def get_sort_vector(self) -> list[float] | None:
    """
    Return the vector used in this ANN search, if applicable.
    If this is not an ANN search, or it was invoked without the
    `include_sort_vector` parameter, return None.

    Invoking this method on a pristine cursor will trigger an API call
    to get the first page of results.
    """

    if self._iterator is None:
        self._iterator = self._create_iterator()
        self._state = CursorState.STARTED
    await self._iterator.preread()
    if self._api_response_status:
        return self._api_response_status.get("sortVector")
    else:
        return None

Inherited members

• BaseCursor:
  • address
  • alive
  • clone
  • close
  • consumed
  • cursor_id
  • include_similarity
  • include_sort_vector
  • limit
  • retrieved
  • rewind
  • skip
  • sort
  • state

class BaseCursor (collection: Collection | AsyncCollection, filter: dict[str, Any] | None, projection: ProjectionType | None, max_time_ms: int | None, overall_max_time_ms: int | None)

Represents a generic Cursor over query results, regardless of whether synchronous or asynchronous. It cannot be instantiated.

See classes Cursor and AsyncCursor for more information.

Expand source code

class BaseCursor:
    """
    Represents a generic Cursor over query results, regardless of whether
    synchronous or asynchronous. It cannot be instantiated.

    See classes Cursor and AsyncCursor for more information.
    """

    _collection: Collection | AsyncCollection
    _filter: dict[str, Any] | None
    _projection: ProjectionType | None
    _max_time_ms: int | None
    _overall_max_time_ms: int | None
    _started_time_s: float | None
    _limit: int | None
    _skip: int | None
    _include_similarity: bool | None
    _include_sort_vector: bool | None
    _sort: dict[str, Any] | None
    _state: CursorState
    _consumed: int
    _iterator: _LookAheadIterator | _AsyncLookAheadIterator | None = None
    _api_response_status: dict[str, Any] | None

    def __init__(
        self,
        collection: Collection | AsyncCollection,
        filter: dict[str, Any] | None,
        projection: ProjectionType | None,
        max_time_ms: int | None,
        overall_max_time_ms: int | None,
    ) -> None:
        raise NotImplementedError

    def __repr__(self) -> str:
        return (
            f'{self.__class__.__name__}("{self._collection.name}", '
            f"{self.state}, "
            f"consumed so far: {self.consumed})"
        )

    def _item_at_index(self, index: int) -> DocumentType:
        # deferred to subclasses
        raise NotImplementedError

    def _ensure_alive(self) -> None:
        if not self.alive:
            raise CursorIsStartedException(
                text="Cursor not alive.",
                cursor_state=self.state,
            )

    def _ensure_idle(self) -> None:
        if self._state != CursorState.IDLE:
            raise CursorIsStartedException(
                text="Cursor started already.",
                cursor_state=self.state,
            )

    def _copy(
        self: BC,
        *,
        projection: ProjectionType | None = None,
        max_time_ms: int | None = None,
        overall_max_time_ms: int | None = None,
        limit: int | None = None,
        skip: int | None = None,
        include_similarity: bool | None = None,
        include_sort_vector: bool | None = None,
        sort: dict[str, Any] | None = None,
    ) -> BC:
        new_cursor = self.__class__(
            collection=self._collection,
            filter=self._filter,
            projection=projection or self._projection,
            max_time_ms=max_time_ms or self._max_time_ms,
            overall_max_time_ms=overall_max_time_ms or self._overall_max_time_ms,
        )
        # Cursor treated as mutable within this function scope:
        new_cursor._limit = limit if limit is not None else self._limit
        new_cursor._skip = skip if skip is not None else self._skip
        new_cursor._include_similarity = (
            include_similarity
            if include_similarity is not None
            else self._include_similarity
        )
        new_cursor._include_sort_vector = (
            include_sort_vector
            if include_sort_vector is not None
            else self._include_sort_vector
        )
        new_cursor._sort = sort if sort is not None else self._sort
        return new_cursor

    @property
    def state(self) -> str:
        """
        The current state of this cursor, which can be one of
        the astrapy.cursors.CursorState enum.
        """

        return self._state.value

    @property
    def address(self) -> str:
        """
        The API endpoint used by this cursor when issuing
        requests to the database.
        """

        return self._collection._api_commander.full_path

    @property
    def alive(self) -> bool:
        """
        Whether the cursor has the potential to yield more data.
        """

        return self._state != CursorState.CLOSED

    def clone(self: BC) -> BC:
        """
        Clone the cursor into a new, fresh one.

        Returns:
            a copy of this cursor, reset to its pristine state,
            i.e. fully un-consumed.
        """

        return self._copy()

    def close(self) -> None:
        """
        Stop/kill the cursor, regardless of its status.
        """

        self._state = CursorState.CLOSED

    @property
    def cursor_id(self) -> int:
        """
        An integer uniquely identifying this cursor.
        """

        return id(self)

    def limit(self: BC, limit: int | None) -> BC:
        """
        Set a new `limit` value for this cursor.

        Args:
            limit: the new value to set

        Returns:
            this cursor itself.
        """

        self._ensure_idle()
        self._ensure_alive()
        self._limit = limit if limit != 0 else None
        return self

    def include_similarity(self: BC, include_similarity: bool | None) -> BC:
        """
        Set a new `include_similarity` value for this cursor.

        Args:
            include_similarity: the new value to set

        Returns:
            this cursor itself.
        """

        self._ensure_idle()
        self._ensure_alive()
        self._include_similarity = include_similarity
        return self

    def include_sort_vector(self: BC, include_sort_vector: bool | None) -> BC:
        """
        Set a new `include_sort_vector` value for this cursor.

        Args:
            include_sort_vector: the new value to set

        Returns:
            this cursor itself.
        """

        self._ensure_idle()
        self._ensure_alive()
        self._include_sort_vector = include_sort_vector
        return self

    @property
    def retrieved(self) -> int:
        """
        The number of documents consumed so far (by the code consuming the cursor).
        """

        the_warning = deprecation.DeprecatedWarning(
            "the 'retrieved' property of Cursor objects",
            deprecated_in="1.5.1",
            removed_in="2.0.0",
            details="Please use the 'consumed' property",
        )
        warnings.warn(the_warning, stacklevel=2)

        return self.consumed

    @property
    def consumed(self) -> int:
        """
        The number of documents consumed so far (by the code consuming the cursor).
        """

        return self._consumed

    def rewind(self: BC) -> BC:
        """
        Reset the cursor to its pristine state, i.e. fully unconsumed.

        Returns:
            this cursor itself.
        """

        self._state = CursorState.IDLE
        self._consumed = 0
        self._iterator = None
        return self

    def skip(self: BC, skip: int | None) -> BC:
        """
        Set a new `skip` value for this cursor.

        Args:
            skip: the new value to set

        Returns:
            this cursor itself.

        Note:
            This parameter can be used only in conjunction with an explicit
            `sort` criterion of the ascending/descending type (i.e. it cannot
            be used when not sorting, nor with vector-based ANN search).
        """
        self._ensure_idle()
        self._ensure_alive()
        self._skip = skip
        return self

    def sort(
        self: BC,
        sort: dict[str, Any] | None,
    ) -> BC:
        """
        Set a new `sort` value for this cursor.

        Args:
            sort: the new sorting prescription to set

        Returns:
            this cursor itself.

        Note:
            Some combinations of arguments impose an implicit upper bound on the
            number of documents that are returned by the Data API. More specifically:
            (a) Vector ANN searches cannot return more than a number of documents
            that at the time of writing is set to 1000 items.
            (b) When using a sort criterion of the ascending/descending type,
            the Data API will return a smaller number of documents, set to 20
            at the time of writing, and stop there. The returned documents are
            the top results across the whole collection according to the requested
            criterion.
            These provisions should be kept in mind even when subsequently running
            a command such as `.distinct()` on a cursor.
        """

        self._ensure_idle()
        self._ensure_alive()
        self._sort = sort
        return self

Subclasses

• AsyncCursor
• Cursor

Instance variables

var address : str

The API endpoint used by this cursor when issuing requests to the database.

Expand source code

@property
def address(self) -> str:
    """
    The API endpoint used by this cursor when issuing
    requests to the database.
    """

    return self._collection._api_commander.full_path

var alive : bool

Whether the cursor has the potential to yield more data.

Expand source code

@property
def alive(self) -> bool:
    """
    Whether the cursor has the potential to yield more data.
    """

    return self._state != CursorState.CLOSED

var consumed : int

The number of documents consumed so far (by the code consuming the cursor).

Expand source code

@property
def consumed(self) -> int:
    """
    The number of documents consumed so far (by the code consuming the cursor).
    """

    return self._consumed

var cursor_id : int

An integer uniquely identifying this cursor.

Expand source code

@property
def cursor_id(self) -> int:
    """
    An integer uniquely identifying this cursor.
    """

    return id(self)

var retrieved : int

The number of documents consumed so far (by the code consuming the cursor).

Expand source code

@property
def retrieved(self) -> int:
    """
    The number of documents consumed so far (by the code consuming the cursor).
    """

    the_warning = deprecation.DeprecatedWarning(
        "the 'retrieved' property of Cursor objects",
        deprecated_in="1.5.1",
        removed_in="2.0.0",
        details="Please use the 'consumed' property",
    )
    warnings.warn(the_warning, stacklevel=2)

    return self.consumed

var state : str

The current state of this cursor, which can be one of the astrapy.cursors.CursorState enum.

Expand source code

@property
def state(self) -> str:
    """
    The current state of this cursor, which can be one of
    the astrapy.cursors.CursorState enum.
    """

    return self._state.value

Methods

def clone(self: BC) ‑> ~BC

Clone the cursor into a new, fresh one.

Returns

a copy of this cursor, reset to its pristine state, i.e. fully un-consumed.

Expand source code

def clone(self: BC) -> BC:
    """
    Clone the cursor into a new, fresh one.

    Returns:
        a copy of this cursor, reset to its pristine state,
        i.e. fully un-consumed.
    """

    return self._copy()

def close(self) ‑> None

Stop/kill the cursor, regardless of its status.

Expand source code

def close(self) -> None:
    """
    Stop/kill the cursor, regardless of its status.
    """

    self._state = CursorState.CLOSED

def include_similarity(self: BC, include_similarity: bool | None) ‑> ~BC

Set a new include_similarity value for this cursor.

Args

include_similarity

the new value to set

Returns

this cursor itself.

Expand source code

def include_similarity(self: BC, include_similarity: bool | None) -> BC:
    """
    Set a new `include_similarity` value for this cursor.

    Args:
        include_similarity: the new value to set

    Returns:
        this cursor itself.
    """

    self._ensure_idle()
    self._ensure_alive()
    self._include_similarity = include_similarity
    return self

def include_sort_vector(self: BC, include_sort_vector: bool | None) ‑> ~BC

Set a new include_sort_vector value for this cursor.

Args

include_sort_vector

the new value to set

Returns

this cursor itself.

Expand source code

def include_sort_vector(self: BC, include_sort_vector: bool | None) -> BC:
    """
    Set a new `include_sort_vector` value for this cursor.

    Args:
        include_sort_vector: the new value to set

    Returns:
        this cursor itself.
    """

    self._ensure_idle()
    self._ensure_alive()
    self._include_sort_vector = include_sort_vector
    return self

def limit(self: BC, limit: int | None) ‑> ~BC

Set a new limit value for this cursor.

Args

limit

the new value to set

Returns

this cursor itself.

Expand source code

def limit(self: BC, limit: int | None) -> BC:
    """
    Set a new `limit` value for this cursor.

    Args:
        limit: the new value to set

    Returns:
        this cursor itself.
    """

    self._ensure_idle()
    self._ensure_alive()
    self._limit = limit if limit != 0 else None
    return self

def rewind(self: BC) ‑> ~BC

Reset the cursor to its pristine state, i.e. fully unconsumed.

Returns

this cursor itself.

Expand source code

def rewind(self: BC) -> BC:
    """
    Reset the cursor to its pristine state, i.e. fully unconsumed.

    Returns:
        this cursor itself.
    """

    self._state = CursorState.IDLE
    self._consumed = 0
    self._iterator = None
    return self

def skip(self: BC, skip: int | None) ‑> ~BC

Set a new skip value for this cursor.

Args

skip

the new value to set

Returns

this cursor itself.

Note

This parameter can be used only in conjunction with an explicit sort criterion of the ascending/descending type (i.e. it cannot be used when not sorting, nor with vector-based ANN search).

Expand source code

def skip(self: BC, skip: int | None) -> BC:
    """
    Set a new `skip` value for this cursor.

    Args:
        skip: the new value to set

    Returns:
        this cursor itself.

    Note:
        This parameter can be used only in conjunction with an explicit
        `sort` criterion of the ascending/descending type (i.e. it cannot
        be used when not sorting, nor with vector-based ANN search).
    """
    self._ensure_idle()
    self._ensure_alive()
    self._skip = skip
    return self

def sort(self: BC, sort: dict[str, Any] | None) ‑> ~BC

Set a new sort value for this cursor.

Args

sort

the new sorting prescription to set

Returns

this cursor itself.

Note

Some combinations of arguments impose an implicit upper bound on the number of documents that are returned by the Data API. More specifically: (a) Vector ANN searches cannot return more than a number of documents that at the time of writing is set to 1000 items. (b) When using a sort criterion of the ascending/descending type, the Data API will return a smaller number of documents, set to 20 at the time of writing, and stop there. The returned documents are the top results across the whole collection according to the requested criterion. These provisions should be kept in mind even when subsequently running a command such as .distinct() on a cursor.

Expand source code

def sort(
    self: BC,
    sort: dict[str, Any] | None,
) -> BC:
    """
    Set a new `sort` value for this cursor.

    Args:
        sort: the new sorting prescription to set

    Returns:
        this cursor itself.

    Note:
        Some combinations of arguments impose an implicit upper bound on the
        number of documents that are returned by the Data API. More specifically:
        (a) Vector ANN searches cannot return more than a number of documents
        that at the time of writing is set to 1000 items.
        (b) When using a sort criterion of the ascending/descending type,
        the Data API will return a smaller number of documents, set to 20
        at the time of writing, and stop there. The returned documents are
        the top results across the whole collection according to the requested
        criterion.
        These provisions should be kept in mind even when subsequently running
        a command such as `.distinct()` on a cursor.
    """

    self._ensure_idle()
    self._ensure_alive()
    self._sort = sort
    return self

class CommandCursor (address: str, items: list[T])

A (synchronous) cursor over the results of a Data API command (as opposed to a cursor over data as one would get with a find method).

Command cursors are iterated over, e.g. with a for loop.

Generally command cursors are not supposed to be instantiated directly, rather they are obtained by invoking methods on a collection/database (such as the database list_collections method).

Expand source code

class CommandCursor(Generic[T]):
    """
    A (synchronous) cursor over the results of a Data API command
    (as opposed to a cursor over data as one would get with a `find` method).

    Command cursors are iterated over, e.g. with a for loop.

    Generally command cursors are not supposed to be instantiated directly,
    rather they are obtained by invoking methods on a collection/database
    (such as the database `list_collections` method).
    """

    def __init__(self, address: str, items: list[T]) -> None:
        self._address = address
        self.items = items
        self.iterable = items.__iter__()
        self._alive = True

    def __repr__(self) -> str:
        return f'{self.__class__.__name__}("{self.address}", ' f"{self.state})"

    def __iter__(self) -> CommandCursor[T]:
        self._ensure_alive()
        return self

    def __next__(self) -> T:
        try:
            item = self.iterable.__next__()
            return item
        except StopIteration:
            self._alive = False
            raise

    @property
    def state(self) -> str:
        """
        The current state of this cursor, which can be:
            - "alive": the cursor has still the potential to return items.
            - "exhausted": the cursor has finished and won't return documents
        """

        return "alive" if self._alive else "exhausted"

    @property
    def address(self) -> str:
        """
        The API endpoint used by this cursor when issuing
        requests to the database.
        """

        return self._address

    @property
    def alive(self) -> bool:
        """
        Whether the cursor has the potential to yield more data.
        """

        return self._alive

    @property
    def cursor_id(self) -> int:
        """
        An integer uniquely identifying this cursor.
        """

        return id(self)

    def _ensure_alive(self) -> None:
        if not self._alive:
            raise CursorIsStartedException(
                text="Cursor is closed.",
                cursor_state=self.state,
            )

    def close(self) -> None:
        """
        Stop/kill the cursor, regardless of its status.
        """

        self._alive = False

Ancestors

• typing.Generic

Instance variables

var address : str

The API endpoint used by this cursor when issuing requests to the database.

Expand source code

@property
def address(self) -> str:
    """
    The API endpoint used by this cursor when issuing
    requests to the database.
    """

    return self._address

var alive : bool

Whether the cursor has the potential to yield more data.

Expand source code

@property
def alive(self) -> bool:
    """
    Whether the cursor has the potential to yield more data.
    """

    return self._alive

var cursor_id : int

An integer uniquely identifying this cursor.

Expand source code

@property
def cursor_id(self) -> int:
    """
    An integer uniquely identifying this cursor.
    """

    return id(self)

var state : str

The current state of this cursor, which can be: - "alive": the cursor has still the potential to return items. - "exhausted": the cursor has finished and won't return documents

Expand source code

@property
def state(self) -> str:
    """
    The current state of this cursor, which can be:
        - "alive": the cursor has still the potential to return items.
        - "exhausted": the cursor has finished and won't return documents
    """

    return "alive" if self._alive else "exhausted"

Methods

def close(self) ‑> None

Stop/kill the cursor, regardless of its status.

Expand source code

def close(self) -> None:
    """
    Stop/kill the cursor, regardless of its status.
    """

    self._alive = False

class Cursor (collection: Collection, filter: dict[str, Any] | None, projection: ProjectionType | None, max_time_ms: int | None, overall_max_time_ms: int | None)

Represents a (synchronous) cursor over documents in a collection. A cursor is iterated over, e.g. with a for loop, and keeps track of its progress.

Generally cursors are not supposed to be instantiated directly, rather they are obtained by invoking the find method on a collection.

Attributes

collection

the collection to find documents in filter: a predicate expressed as a dictionary according to the Data API filter syntax. Examples are: {} {"name": "John"} {"price": {"$le": 100}} {"$and": [{"name": "John"}, {"price": {"$le": 100}}]} See the Data API documentation for the full set of operators. projection: used to select a subset of fields in the document being returned. The projection can be: an iterable over the field names to return; a dictionary {field_name: True} to positively select certain fields; or a dictionary {field_name: False} if one wants to discard some fields from the response. The default is to return the whole documents. max_time_ms: a timeout, in milliseconds, for each single one of the underlying HTTP requests used to fetch documents as the cursor is iterated over.

Note

When not specifying sorting criteria at all (by vector or otherwise), the cursor can scroll through an arbitrary number of documents as the Data API and the client periodically exchange new chunks of documents. It should be noted that the behavior of the cursor in the case documents have been added/removed after the cursor was started depends on database internals and it is not guaranteed, nor excluded, that such "real-time" changes in the data would be picked up by the cursor.

Expand source code

class Cursor(BaseCursor):
    """
    Represents a (synchronous) cursor over documents in a collection.
    A cursor is iterated over, e.g. with a for loop, and keeps track of
    its progress.

    Generally cursors are not supposed to be instantiated directly,
    rather they are obtained by invoking the `find` method on a collection.

    Attributes:
        collection: the collection to find documents in
            filter: a predicate expressed as a dictionary according to the
                Data API filter syntax. Examples are:
                    {}
                    {"name": "John"}
                    {"price": {"$le": 100}}
                    {"$and": [{"name": "John"}, {"price": {"$le": 100}}]}
                See the Data API documentation for the full set of operators.
            projection: used to select a subset of fields in the document being
                returned. The projection can be: an iterable over the field names
                to return; a dictionary {field_name: True} to positively select
                certain fields; or a dictionary {field_name: False} if one wants
                to discard some fields from the response.
                The default is to return the whole documents.
            max_time_ms: a timeout, in milliseconds, for each single one
                of the underlying HTTP requests used to fetch documents as the
                cursor is iterated over.

    Note:
        When not specifying sorting criteria at all (by vector or otherwise),
        the cursor can scroll through an arbitrary number of documents as
        the Data API and the client periodically exchange new chunks of documents.
        It should be noted that the behavior of the cursor in the case documents
        have been added/removed after the cursor was started depends on database
        internals and it is not guaranteed, nor excluded, that such "real-time"
        changes in the data would be picked up by the cursor.
    """

    def __init__(
        self,
        collection: Collection,
        filter: dict[str, Any] | None,
        projection: ProjectionType | None,
        max_time_ms: int | None,
        overall_max_time_ms: int | None,
    ) -> None:
        self._collection: Collection = collection
        self._filter = filter
        self._projection = projection
        self._overall_max_time_ms = overall_max_time_ms
        if overall_max_time_ms is not None and max_time_ms is not None:
            self._max_time_ms = min(max_time_ms, overall_max_time_ms)
        else:
            self._max_time_ms = max_time_ms
        self._limit: int | None = None
        self._skip: int | None = None
        self._include_similarity: bool | None = None
        self._include_sort_vector: bool | None = None
        self._sort: dict[str, Any] | None = None
        self._state = CursorState.IDLE
        self._consumed = 0
        #
        self._iterator: _LookAheadIterator | None = None
        self._api_response_status: dict[str, Any] | None = None

    def __iter__(self) -> Cursor:
        self._ensure_alive()
        if self._iterator is None:
            self._iterator = self._create_iterator()
            self._state = CursorState.STARTED
        return self

    def __next__(self) -> DocumentType:
        if not self.alive:
            # keep raising once exhausted:
            raise StopIteration
        if self._iterator is None:
            self._iterator = self._create_iterator()
            self._state = CursorState.STARTED
        # check for overall timing out
        if self._overall_max_time_ms is not None:
            _elapsed = time.time() - self._started_time_s  # type: ignore[operator]
            if _elapsed > (self._overall_max_time_ms / 1000.0):
                raise DataAPITimeoutException(
                    text="Cursor timed out.",
                    timeout_type="generic",
                    endpoint=None,
                    raw_payload=None,
                )
        try:
            next_item = self._iterator.__next__()
            self._consumed = self._consumed + 1
            return next_item
        except StopIteration:
            self.close()
            raise

    def get_sort_vector(self) -> list[float] | None:
        """
        Return the vector used in this ANN search, if applicable.
        If this is not an ANN search, or it was invoked without the
        `include_sort_vector` parameter, return None.

        Invoking this method on a pristine cursor will trigger an API call
        to get the first page of results.
        """

        if self._iterator is None:
            self._iterator = self._create_iterator()
            self._state = CursorState.STARTED
        self._iterator.preread()
        if self._api_response_status:
            return self._api_response_status.get("sortVector")
        else:
            return None

    def _item_at_index(self, index: int) -> DocumentType:
        finder_cursor = self._copy().skip(index).limit(1)
        items = list(finder_cursor)
        if items:
            return items[0]
        else:
            raise IndexError("no such item for Cursor instance")

    def _create_iterator(self) -> _LookAheadIterator:
        self._ensure_idle()
        self._ensure_alive()
        _options_0 = {
            k: v
            for k, v in {
                "limit": self._limit,
                "skip": self._skip,
                "includeSimilarity": self._include_similarity,
                "includeSortVector": self._include_sort_vector,
            }.items()
            if v is not None
        }
        _projection = normalize_optional_projection(self._projection)
        _sort = self._sort or {}
        _filter = self._filter or {}
        f0_payload = {
            "find": {
                k: v
                for k, v in {
                    "filter": _filter,
                    "projection": _projection,
                    "options": _options_0,
                    "sort": _sort,
                }.items()
                if v
            }
        }

        def _find_iterator() -> Iterator[DocumentType]:
            next_page_state: str | None = None
            #
            resp_0 = self._collection.command(
                body=f0_payload,
                max_time_ms=self._max_time_ms,
            )
            self._api_response_status = resp_0.get("status")
            if "nextPageState" not in resp_0.get("data", {}):
                raise DataAPIFaultyResponseException(
                    text="Faulty response from find API command (no 'nextPageState').",
                    raw_response=resp_0,
                )
            next_page_state = resp_0["data"]["nextPageState"]
            if "documents" not in resp_0.get("data", {}):
                raise DataAPIFaultyResponseException(
                    text="Faulty response from find API command (no 'documents').",
                    raw_response=resp_0,
                )
            for doc in resp_0["data"]["documents"]:
                yield doc
            while next_page_state is not None:
                _options_n = {**_options_0, **{"pageState": next_page_state}}
                fn_payload = {
                    "find": {
                        k: v
                        for k, v in {
                            "filter": _filter,
                            "projection": _projection,
                            "options": _options_n,
                            "sort": _sort,
                        }.items()
                        if v
                    }
                }
                resp_n = self._collection.command(
                    body=fn_payload,
                    max_time_ms=self._max_time_ms,
                )
                self._api_response_status = resp_n.get("status")
                if "nextPageState" not in resp_n.get("data", {}):
                    raise DataAPIFaultyResponseException(
                        text="Faulty response from find API command (no 'nextPageState').",
                        raw_response=resp_n,
                    )
                next_page_state = resp_n["data"]["nextPageState"]
                if "documents" not in resp_n.get("data", {}):
                    raise DataAPIFaultyResponseException(
                        text="Faulty response from find API command (no 'documents').",
                        raw_response=resp_n,
                    )
                for doc in resp_n["data"]["documents"]:
                    yield doc

        logger.info(f"creating iterator on '{self._collection.name}'")
        iterator = _find_iterator()
        logger.info(f"finished creating iterator on '{self._collection.name}'")
        self._started_time_s = time.time()
        return _LookAheadIterator(iterator)

    @property
    def collection(self) -> Collection:
        """
        The (synchronous) collection this cursor is targeting.
        """

        the_warning = deprecation.DeprecatedWarning(
            "the 'collection' property of Cursor objects",
            deprecated_in="1.5.1",
            removed_in="2.0.0",
            details="Please use the 'data_source' property",
        )
        warnings.warn(the_warning, stacklevel=2)

        return self.data_source

    @property
    def data_source(self) -> Collection:
        """
        The (synchronous) collection this cursor is targeting.
        """

        return self._collection

    def distinct(self, key: str, max_time_ms: int | None = None) -> list[Any]:
        """
        Compute a list of unique values for a specific field across all
        documents the cursor iterates through.

        Invoking this method has no effect on the cursor state, i.e.
        the position of the cursor is unchanged.

        Args:
            key: the name of the field whose value is inspected across documents.
                Keys can use dot-notation to descend to deeper document levels.
                Example of acceptable `key` values:
                    "field"
                    "field.subfield"
                    "field.3"
                    "field.3.subfield"
                if lists are encountered and no numeric index is specified,
                all items in the list are visited.
            max_time_ms: a timeout, in milliseconds, for the operation.

        Note:
            this operation works at client-side by scrolling through all
            documents matching the cursor parameters (such as `filter`).
            Please be aware of this fact, especially for a very large
            amount of documents, for this may have implications on latency,
            network traffic and possibly billing.
        """

        _item_hashes = set()
        distinct_items = []

        _extractor = _create_document_key_extractor(key)
        _key = _reduce_distinct_key_to_safe(key)

        if _key == "":
            raise ValueError(
                "The 'key' parameter for distinct cannot be empty "
                "or start with a list index."
            )

        d_cursor = self._copy(
            projection={_key: True},
            overall_max_time_ms=max_time_ms,
        )
        logger.info(f"running distinct() on '{self._collection.name}'")
        for document in d_cursor:
            for item in _extractor(document):
                _item_hash = _hash_document(item)
                if _item_hash not in _item_hashes:
                    _item_hashes.add(_item_hash)
                    distinct_items.append(item)

        logger.info(f"finished running distinct() on '{self._collection.name}'")
        return distinct_items

Ancestors

• BaseCursor

Instance variables

var collection : Collection

The (synchronous) collection this cursor is targeting.

Expand source code

@property
def collection(self) -> Collection:
    """
    The (synchronous) collection this cursor is targeting.
    """

    the_warning = deprecation.DeprecatedWarning(
        "the 'collection' property of Cursor objects",
        deprecated_in="1.5.1",
        removed_in="2.0.0",
        details="Please use the 'data_source' property",
    )
    warnings.warn(the_warning, stacklevel=2)

    return self.data_source

var data_source : Collection

The (synchronous) collection this cursor is targeting.

Expand source code

@property
def data_source(self) -> Collection:
    """
    The (synchronous) collection this cursor is targeting.
    """

    return self._collection

Methods

def distinct(self, key: str, max_time_ms: int | None = None) ‑> list[typing.Any]

Compute a list of unique values for a specific field across all documents the cursor iterates through.

Invoking this method has no effect on the cursor state, i.e. the position of the cursor is unchanged.

Args

key

the name of the field whose value is inspected across documents. Keys can use dot-notation to descend to deeper document levels. Example of acceptable key values: "field" "field.subfield" "field.3" "field.3.subfield" if lists are encountered and no numeric index is specified, all items in the list are visited.

max_time_ms

a timeout, in milliseconds, for the operation.

Note

this operation works at client-side by scrolling through all documents matching the cursor parameters (such as filter). Please be aware of this fact, especially for a very large amount of documents, for this may have implications on latency, network traffic and possibly billing.

Expand source code

def distinct(self, key: str, max_time_ms: int | None = None) -> list[Any]:
    """
    Compute a list of unique values for a specific field across all
    documents the cursor iterates through.

    Invoking this method has no effect on the cursor state, i.e.
    the position of the cursor is unchanged.

    Args:
        key: the name of the field whose value is inspected across documents.
            Keys can use dot-notation to descend to deeper document levels.
            Example of acceptable `key` values:
                "field"
                "field.subfield"
                "field.3"
                "field.3.subfield"
            if lists are encountered and no numeric index is specified,
            all items in the list are visited.
        max_time_ms: a timeout, in milliseconds, for the operation.

    Note:
        this operation works at client-side by scrolling through all
        documents matching the cursor parameters (such as `filter`).
        Please be aware of this fact, especially for a very large
        amount of documents, for this may have implications on latency,
        network traffic and possibly billing.
    """

    _item_hashes = set()
    distinct_items = []

    _extractor = _create_document_key_extractor(key)
    _key = _reduce_distinct_key_to_safe(key)

    if _key == "":
        raise ValueError(
            "The 'key' parameter for distinct cannot be empty "
            "or start with a list index."
        )

    d_cursor = self._copy(
        projection={_key: True},
        overall_max_time_ms=max_time_ms,
    )
    logger.info(f"running distinct() on '{self._collection.name}'")
    for document in d_cursor:
        for item in _extractor(document):
            _item_hash = _hash_document(item)
            if _item_hash not in _item_hashes:
                _item_hashes.add(_item_hash)
                distinct_items.append(item)

    logger.info(f"finished running distinct() on '{self._collection.name}'")
    return distinct_items

def get_sort_vector(self) ‑> list[float] | None

Return the vector used in this ANN search, if applicable. If this is not an ANN search, or it was invoked without the include_sort_vector parameter, return None.

Invoking this method on a pristine cursor will trigger an API call to get the first page of results.

Expand source code

def get_sort_vector(self) -> list[float] | None:
    """
    Return the vector used in this ANN search, if applicable.
    If this is not an ANN search, or it was invoked without the
    `include_sort_vector` parameter, return None.

    Invoking this method on a pristine cursor will trigger an API call
    to get the first page of results.
    """

    if self._iterator is None:
        self._iterator = self._create_iterator()
        self._state = CursorState.STARTED
    self._iterator.preread()
    if self._api_response_status:
        return self._api_response_status.get("sortVector")
    else:
        return None

Inherited members

• BaseCursor:
  • address
  • alive
  • clone
  • close
  • consumed
  • cursor_id
  • include_similarity
  • include_sort_vector
  • limit
  • retrieved
  • rewind
  • skip
  • sort
  • state

class CursorState (value, names=None, *, module=None, qualname=None, type=None, start=1)

An enumeration.

Expand source code

class CursorState(Enum):
    # Iteration over results has not started yet (alive=T, started=F)
    IDLE = "idle"
    # Iteration has started, *can* still yield results (alive=T, started=T)
    STARTED = "started"
    # Finished/forcibly stopped. Won't return more documents (alive=F)
    CLOSED = "closed"

Ancestors

• enum.Enum

Class variables

var CLOSED

var IDLE

var STARTED