Count

Methods for counting the number of documents.

Count

Methods for counting the number of documents.

Source code in src/scruby/mixins/count.py

class Count:
    """Methods for counting the number of documents."""

    @final
    async def estimated_document_count(self) -> int:
        """Asynchronous method.

        Get an estimate of the number of documents in this collection using collection metadata.

        Returns:
            The number of documents.
        """
        meta = await self.get_meta()
        return meta.counter_documents

    @final
    def count_documents(
        self,
        filter_fn: Callable,
    ) -> int:
        """Synchronous method.

        Count the number of documents a matching the filter in this collection.

        Attention:
            - The search is based on the effect of a quantum loop.
            - The search effectiveness depends on the number of processor threads.

        Args:
            filter_fn (Callable): A function that execute the conditions of filtering.

        Returns:
            The number of documents.
        """
        # Variable initialization
        hash_reduce_left: int = self._hash_reduce_left
        assert hash_reduce_left != 0, "Scruby.run(hash_reduce_left = 0) - Not valid for `count_documents` method."

        search_task_fn: Callable = self._task_find
        branch_numbers: range = range(self._max_number_branch)
        hash_reduce_left: int = self._hash_reduce_left
        class_model: Any = self._class_model
        stop_signal = Event()
        counter: int = 0
        # Run quantum loop
        with ThreadPoolExecutor(self._max_workers) as executor:
            futures: list[Future] = [
                executor.submit(
                    search_task_fn,
                    filter_fn,
                    hash_reduce_left,
                    branch_number,
                    class_model,
                    stop_signal,
                )
                for branch_number in branch_numbers
            ]
            for future in as_completed(futures):
                docs = future.result()
                if docs is not None:
                    counter += len(docs)
        return counter

count_documents(filter_fn)

Synchronous method.

Count the number of documents a matching the filter in this collection.

Attention

• The search is based on the effect of a quantum loop.
• The search effectiveness depends on the number of processor threads.

Parameters:

Name	Type	Description	Default
filter_fn	Callable	A function that execute the conditions of filtering.	required

Returns:

Type	Description
int	The number of documents.

Source code in src/scruby/mixins/count.py

@final
def count_documents(
    self,
    filter_fn: Callable,
) -> int:
    """Synchronous method.

    Count the number of documents a matching the filter in this collection.

    Attention:
        - The search is based on the effect of a quantum loop.
        - The search effectiveness depends on the number of processor threads.

    Args:
        filter_fn (Callable): A function that execute the conditions of filtering.

    Returns:
        The number of documents.
    """
    # Variable initialization
    hash_reduce_left: int = self._hash_reduce_left
    assert hash_reduce_left != 0, "Scruby.run(hash_reduce_left = 0) - Not valid for `count_documents` method."

    search_task_fn: Callable = self._task_find
    branch_numbers: range = range(self._max_number_branch)
    hash_reduce_left: int = self._hash_reduce_left
    class_model: Any = self._class_model
    stop_signal = Event()
    counter: int = 0
    # Run quantum loop
    with ThreadPoolExecutor(self._max_workers) as executor:
        futures: list[Future] = [
            executor.submit(
                search_task_fn,
                filter_fn,
                hash_reduce_left,
                branch_number,
                class_model,
                stop_signal,
            )
            for branch_number in branch_numbers
        ]
        for future in as_completed(futures):
            docs = future.result()
            if docs is not None:
                counter += len(docs)
    return counter

estimated_document_count() async

Asynchronous method.

Get an estimate of the number of documents in this collection using collection metadata.

Returns:

Type	Description
int	The number of documents.

Source code in src/scruby/mixins/count.py

@final
async def estimated_document_count(self) -> int:
    """Asynchronous method.

    Get an estimate of the number of documents in this collection using collection metadata.

    Returns:
        The number of documents.
    """
    meta = await self.get_meta()
    return meta.counter_documents