neosqlite.collection.query_helper.aggregation module

Aggregation pipeline methods for NeoSQLite.

This module contains the AggregationMixin class, which provides Python-based aggregation pipeline processing. SQL-based aggregation lives in _sql_aggregation.py (SqlAggregationMixin).

neosqlite.collection.query_helper.aggregation._addtoset_key(value: Any) → Any

Hashable canonical form for $addToSet membership (#155).

class neosqlite.collection.query_helper.aggregation.AggregationMixin

Bases: SqlAggregationMixin

Mixin class providing aggregation pipeline methods.

This mixin assumes it will be used with a class that has the following:

self.collection

A collection instance with: - db: Database connection - name: Collection name - _load: Method to load documents - _get_val: Method to get values from documents - _set_val: Method to set values in documents

self.jsonb.jsonb_supported

Whether JSONB is supported

self.jsonb.json_function_prefix

“json” or “jsonb”

self.jsonb.json_each_function

“json_each” or “jsonb_each”

self._build_simple_where_clause

Method to build WHERE clauses

self._reorder_pipeline_for_indexes

Method to reorder pipelines

self._estimate_pipeline_cost

Method to estimate costs

self._optimize_match_pushdown

Method to optimize match pushdown

self._is_datetime_indexed_field

Method to check datetime indexes

self._build_group_query

Method to build group queries

self._apply_query

Method to apply queries to documents

collection: Collection

jsonb: JSONBContext

_build_simple_where_clause: Any

_reorder_pipeline_for_indexes: Any

_estimate_pipeline_cost: Any

_optimize_match_pushdown: Any

_is_datetime_indexed_field: Any

_apply_query: Any

_process_group_stage(group_query: dict[str, Any], docs: list[dict[str, Any]]) → list[dict[str, Any]]

Process the $group stage of an aggregation pipeline.

This method groups documents by a specified field and performs specified accumulator operations on other fields.

Parameters:

• group_query (dict[str, Any]) – A dictionary representing the $group stage of the aggregation pipeline.

• docs (list[dict[str, Any]]) – A list of documents to be grouped.

Returns:

A list of grouped documents with applied

accumulator operations.

Return type:

list[dict[str, Any]]

_run_subpipeline(sub_pipeline: list[dict[str, Any]], docs: list[dict[str, Any]], batch_size: int = 101) → str

Run a sub-pipeline (e.g., for $facet) on a list of documents.

Uses tier optimization (Tier-1/Tier-2/Tier-3) for each sub-pipeline. Results are streamed to a temporary table in batches to avoid memory issues.

Parameters:

• sub_pipeline – List of pipeline stages to execute

• docs – Input documents

• batch_size – Number of documents to process in each batch

Returns:

Name of the temporary table containing results

_apply_projection(projection: dict[str, Any], document: dict[str, Any]) → dict[str, Any]

Applies the projection to the document, selecting or excluding fields based on the projection criteria.

Parameters:

• projection (dict[str, Any]) – A dictionary specifying which fields to include or exclude.

• document (dict[str, Any]) – The document to apply the projection to.

Returns:

The document with fields applied based on the projection.

Return type:

dict[str, Any]