neosqlite.collection.query_helper.update_operations module

Update operations mixin for QueryHelper.

This module contains the UpdateOperationsMixin class which provides SQL-based and Python-based update operations for NeoSQLite collections.

class neosqlite.collection.query_helper.update_operations.UpdateOperationsMixin[source]

Bases: object

A mixin class providing update operations for QueryHelper.

This mixin assumes it will be used with a class that has: - self.collection (with db and name attributes) - self._jsonb_supported - self._json_function_prefix - self._build_simple_where_clause method

collection: Collection
_jsonb_supported: bool
_json_function_prefix: str
_get_integer_id_for_oid: Any
_internal_update(doc_id: Any, update_spec: dict[str, Any], original_doc: dict[str, Any], array_filters: list[dict[str, Any]] | None = None, query_filter: dict[str, Any] | None = None) → tuple[dict[str, Any], bool][source]

Helper method for updating documents.

Attempts to use SQL-based updates for simple operations, falling back to Python-based updates for complex operations.

Parameters:

  • doc_id (Any) – The ID of the document to update (can be ObjectId, int, etc.).

  • update_spec (dict[str, Any]) – The update specification.

  • original_doc (dict[str, Any]) – The original document before the update.

  • array_filters (list[dict[str, Any]], optional) – Filter documents for array positional operators.

  • query_filter (dict[str, Any], optional) – The query filter for $ operator.

Returns:

The updated document and whether it was modified.

Return type:

tuple[dict[str, Any], bool]

_can_use_sql_updates(update_spec: dict[str, Any], doc_id: int, original_doc: dict[str, Any] | None = None) → bool[source]

Check if all operations in the update spec can be handled with SQL.

This method determines whether the update operations can be efficiently executed using SQL directly, which allows for better performance compared to iterating over each document and applying updates in Python.

Parameters:

  • update_spec (dict[str, Any]) – The update operations to be checked.

  • doc_id (int) – The document ID, which is used to determine if the update is an upsert.

Returns:

True if all operations can be handled with SQL, False otherwise.

Return type:

bool

_perform_sql_update(doc_id: int, update_spec: dict[str, Any]) → dict[str, Any][source]

Perform update operations using SQL JSON functions.

This method builds SQL clauses for updating document fields based on the provided update specification. It supports both $set and $unset operations using SQLite’s json_set and json_remove functions, respectively. The method then executes the SQL commands to apply the updates and fetches the updated document from the database.

Parameters:

  • doc_id (int) – The ID of the document to be updated.

  • update_spec (dict[str, Any]) – A dictionary specifying the update operations to be performed.

Returns:

The updated document.

Return type:

dict[str, Any]

Raises:

RuntimeError – If no rows are updated or if an error occurs during the update process.

_perform_enhanced_sql_update(doc_id: Any, update_spec: dict[str, Any], original_doc: dict[str, Any] | None = None) → dict[str, Any][source]

Perform update operations using SQL JSON functions with field-level granularity.

This method optimizes update operations by using specialized JSON functions (json_insert, json_replace, etc.) based on whether fields already exist. It provides field-level updates rather than whole-document rewrites.

Parameters:

  • doc_id (Any) – The ID of the document to be updated.

  • update_spec (dict[str, Any]) – A dictionary specifying the update operations to be performed.

  • original_doc (dict[str, Any], optional) – The original document before the update. If provided, used to determine existing fields instead of fetching again.

Returns:

The updated document.

Return type:

dict[str, Any]

Raises:

RuntimeError – If no rows are updated or if an error occurs during the update process.

_get_document_fields(doc_id: Any) → set[str][source]

Get the set of field names in a document.

This method extracts the field names from a document to determine which fields already exist and which are new. This is used to decide between json_insert and json_replace operations.

Parameters:

doc_id (Any) – The ID of the document to analyze.

Returns:

A set of field names in the document.

Return type:

set

_build_update_clause(update: dict[str, Any]) → tuple[str, list[Any]] | None[source]

Build the SQL update clause based on the provided update operations.

Parameters:

update (dict[str, Any]) – A dictionary containing update operations.

Returns:

A tuple containing the SQL update clause

and parameters, or None if no update clauses are generated.

Return type:

tuple[str, list[Any]] | None

_build_sql_update_clause(op: str, value: Any) → tuple[list[str], list[Any]][source]

Build SQL update clause for a single operation.

Parameters:

  • op (str) – The update operation, such as “$set”, “$inc”, “$mul”, etc.

  • value (Any) – The value associated with the update operation.

Returns:

A tuple containing the SQL update clauses

and parameters.

Return type:

tuple[list[str], list[Any]]

_perform_python_update(doc_id: Any, update_spec: dict[str, Any], original_doc: dict[str, Any], array_filters: list[dict[str, Any]] | None = None, query_filter: dict[str, Any] | None = None) → tuple[dict[str, Any], bool][source]

Perform update operations using Python-based logic.

Parameters:

  • doc_id (Any) – The document ID of the document to update (can be ObjectId, int, etc.).

  • update_spec (dict[str, Any]) – A dictionary specifying the update operations to perform.

  • original_doc (dict[str, Any]) – The original document before applying the updates.

  • array_filters (list[dict[str, Any]], optional) – Filter documents for array positional operators.

  • query_filter (dict[str, Any], optional) – The query filter for $ operator.

Returns:

The updated document and whether it was modified.

Return type:

tuple[dict[str, Any], bool]

static _validate_inc_mul_types_sql(db: Any, collection_name: str, where_clause: str | None, where_params: list[Any], update: dict[str, Any], jsonb_supported: bool) → bool[source]

Validate that fields in $inc/$mul operations are numeric.

This checks the JSON type of each field being incremented/multiplied to ensure they’re numeric types.

Parameters:

  • db – Database connection

  • collection_name – Name of the collection

  • where_clause – The translated WHERE clause

  • where_params – Parameters for the WHERE clause

  • update – The update operations

  • jsonb_supported – Whether JSONB is supported

Returns:

True if all fields are numeric or don’t exist, False if any field is non-numeric