Module astrapy.results
Expand source code
# Copyright DataStax, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from __future__ import annotations
from abc import ABC
from dataclasses import dataclass
from typing import Any
@dataclass
class OperationResult(ABC):
"""
Class that represents the generic result of a single mutation operation.
Attributes:
raw_results: response/responses from the Data API call.
Depending on the exact delete method being used, this
list of raw responses can contain exactly one or a number of items.
"""
raw_results: list[dict[str, Any]]
def _piecewise_repr(self, pieces: list[str | None]) -> str:
return f"{self.__class__.__name__}({', '.join(pc for pc in pieces if pc)})"
@dataclass
class CollectionDeleteResult(OperationResult):
"""
Class that represents the result of delete operations on a collection.
Attributes:
deleted_count: number of deleted documents
raw_results: response/responses from the Data API call.
Depending on the exact delete method being used, this
list of raw responses can contain exactly one or a number of items.
"""
deleted_count: int
def __repr__(self) -> str:
return self._piecewise_repr(
[
f"deleted_count={self.deleted_count}",
"raw_results=..." if self.raw_results is not None else None,
]
)
@dataclass
class CollectionInsertOneResult(OperationResult):
"""
Class that represents the result of insert_one operations on a collection.
Attributes:
raw_results: one-item list with the response from the Data API call
inserted_id: the ID of the inserted document
"""
inserted_id: Any
def __repr__(self) -> str:
return self._piecewise_repr(
[
f"inserted_id={self.inserted_id}",
"raw_results=..." if self.raw_results is not None else None,
]
)
@dataclass
class CollectionInsertManyResult(OperationResult):
"""
Class that represents the result of insert_many operations on a collection.
Attributes:
raw_results: responses from the Data API calls
inserted_ids: list of the IDs of the inserted documents
"""
inserted_ids: list[Any]
def __repr__(self) -> str:
_ins_ids_str: str
if len(self.inserted_ids) > 5:
_ins_ids_str = (
f"[{', '.join(str(_iid) for _iid in self.inserted_ids[:5])} "
f"... ({len(self.inserted_ids)} total)]"
)
else:
_ins_ids_str = str(self.inserted_ids)
return self._piecewise_repr(
[
f"inserted_ids={_ins_ids_str}",
"raw_results=..." if self.raw_results is not None else None,
]
)
@dataclass
class CollectionUpdateResult(OperationResult):
"""
Class that represents the result of any update operation on a collection.
Attributes:
raw_results: responses from the Data API calls
update_info: a dictionary reporting about the update
Note:
the "update_info" field has the following fields: "n" (int),
"updatedExisting" (bool), "ok" (float), "nModified" (int)
and optionally "upserted" containing the ID of an upserted document.
"""
update_info: dict[str, Any]
def __repr__(self) -> str:
return self._piecewise_repr(
[
f"update_info={self.update_info}",
"raw_results=..." if self.raw_results is not None else None,
]
)
@dataclass
class TableInsertOneResult(OperationResult):
"""
Class that represents the result of insert_one operations on a table.
Attributes:
raw_results: one-item list with the response from the Data API call
inserted_id: the ID of the inserted document in the form of a dict
inserted_id_tuple: an ordered-tuple version of the same ID
"""
inserted_id: Any
inserted_id_tuple: tuple[Any, ...]
def __repr__(self) -> str:
return self._piecewise_repr(
[
f"inserted_id={self.inserted_id}",
f"inserted_id_tuple={self.inserted_id_tuple}",
"raw_results=..." if self.raw_results is not None else None,
]
)
@dataclass
class TableInsertManyResult(OperationResult):
"""
Class that represents the result of insert_many operations on a collection.
Attributes:
raw_results: responses from the Data API calls
inserted_ids: list of the IDs of the inserted documents in the form of dictionaries
inserted_id_tuple: an ordered-tuple version of the same inserted IDs
"""
inserted_ids: list[Any]
inserted_id_tuples: list[tuple[Any, ...]]
def __repr__(self) -> str:
_ins_ids_str: str
if len(self.inserted_ids) > 5:
_ins_ids_str = (
f"[{', '.join(str(_iid) for _iid in self.inserted_ids[:5])} "
f"... ({len(self.inserted_ids)} total)]"
)
else:
_ins_ids_str = str(self.inserted_ids)
_ins_idts_str: str
if len(self.inserted_id_tuples) > 5:
_ins_idts_str = (
f"[{', '.join(str(_iidt) for _iidt in self.inserted_id_tuples[:5])} "
f"... ({len(self.inserted_id_tuples)} total)]"
)
else:
_ins_idts_str = str(self.inserted_id_tuples)
return self._piecewise_repr(
[
f"inserted_ids={_ins_ids_str}",
f"inserted_id_tuples={_ins_idts_str}",
"raw_results=..." if self.raw_results is not None else None,
]
)
Classes
class CollectionDeleteResult (raw_results: list[dict[str, Any]], deleted_count: int)
-
Class that represents the result of delete operations on a collection.
Attributes
deleted_count
- number of deleted documents
raw_results
- response/responses from the Data API call. Depending on the exact delete method being used, this list of raw responses can contain exactly one or a number of items.
Expand source code
@dataclass class CollectionDeleteResult(OperationResult): """ Class that represents the result of delete operations on a collection. Attributes: deleted_count: number of deleted documents raw_results: response/responses from the Data API call. Depending on the exact delete method being used, this list of raw responses can contain exactly one or a number of items. """ deleted_count: int def __repr__(self) -> str: return self._piecewise_repr( [ f"deleted_count={self.deleted_count}", "raw_results=..." if self.raw_results is not None else None, ] )
Ancestors
- OperationResult
- abc.ABC
Class variables
var deleted_count : int
class CollectionInsertManyResult (raw_results: list[dict[str, Any]], inserted_ids: list[Any])
-
Class that represents the result of insert_many operations on a collection.
Attributes
raw_results
- responses from the Data API calls
inserted_ids
- list of the IDs of the inserted documents
Expand source code
@dataclass class CollectionInsertManyResult(OperationResult): """ Class that represents the result of insert_many operations on a collection. Attributes: raw_results: responses from the Data API calls inserted_ids: list of the IDs of the inserted documents """ inserted_ids: list[Any] def __repr__(self) -> str: _ins_ids_str: str if len(self.inserted_ids) > 5: _ins_ids_str = ( f"[{', '.join(str(_iid) for _iid in self.inserted_ids[:5])} " f"... ({len(self.inserted_ids)} total)]" ) else: _ins_ids_str = str(self.inserted_ids) return self._piecewise_repr( [ f"inserted_ids={_ins_ids_str}", "raw_results=..." if self.raw_results is not None else None, ] )
Ancestors
- OperationResult
- abc.ABC
Class variables
var inserted_ids : list[typing.Any]
class CollectionInsertOneResult (raw_results: list[dict[str, Any]], inserted_id: Any)
-
Class that represents the result of insert_one operations on a collection.
Attributes
raw_results
- one-item list with the response from the Data API call
inserted_id
- the ID of the inserted document
Expand source code
@dataclass class CollectionInsertOneResult(OperationResult): """ Class that represents the result of insert_one operations on a collection. Attributes: raw_results: one-item list with the response from the Data API call inserted_id: the ID of the inserted document """ inserted_id: Any def __repr__(self) -> str: return self._piecewise_repr( [ f"inserted_id={self.inserted_id}", "raw_results=..." if self.raw_results is not None else None, ] )
Ancestors
- OperationResult
- abc.ABC
Class variables
var inserted_id : Any
class CollectionUpdateResult (raw_results: list[dict[str, Any]], update_info: dict[str, Any])
-
Class that represents the result of any update operation on a collection.
Attributes
raw_results
- responses from the Data API calls
update_info
- a dictionary reporting about the update
Note
the "update_info" field has the following fields: "n" (int), "updatedExisting" (bool), "ok" (float), "nModified" (int) and optionally "upserted" containing the ID of an upserted document.
Expand source code
@dataclass class CollectionUpdateResult(OperationResult): """ Class that represents the result of any update operation on a collection. Attributes: raw_results: responses from the Data API calls update_info: a dictionary reporting about the update Note: the "update_info" field has the following fields: "n" (int), "updatedExisting" (bool), "ok" (float), "nModified" (int) and optionally "upserted" containing the ID of an upserted document. """ update_info: dict[str, Any] def __repr__(self) -> str: return self._piecewise_repr( [ f"update_info={self.update_info}", "raw_results=..." if self.raw_results is not None else None, ] )
Ancestors
- OperationResult
- abc.ABC
Class variables
var update_info : dict[str, typing.Any]
class OperationResult (raw_results: list[dict[str, Any]])
-
Class that represents the generic result of a single mutation operation.
Attributes
raw_results
- response/responses from the Data API call. Depending on the exact delete method being used, this list of raw responses can contain exactly one or a number of items.
Expand source code
@dataclass class OperationResult(ABC): """ Class that represents the generic result of a single mutation operation. Attributes: raw_results: response/responses from the Data API call. Depending on the exact delete method being used, this list of raw responses can contain exactly one or a number of items. """ raw_results: list[dict[str, Any]] def _piecewise_repr(self, pieces: list[str | None]) -> str: return f"{self.__class__.__name__}({', '.join(pc for pc in pieces if pc)})"
Ancestors
- abc.ABC
Subclasses
- CollectionDeleteResult
- CollectionInsertManyResult
- CollectionInsertOneResult
- CollectionUpdateResult
- TableInsertManyResult
- TableInsertOneResult
Class variables
var raw_results : list[dict[str, typing.Any]]
class TableInsertManyResult (raw_results: list[dict[str, Any]], inserted_ids: list[Any], inserted_id_tuples: list[tuple[Any, ...]])
-
Class that represents the result of insert_many operations on a collection.
Attributes
raw_results
- responses from the Data API calls
inserted_ids
- list of the IDs of the inserted documents in the form of dictionaries
inserted_id_tuple
- an ordered-tuple version of the same inserted IDs
Expand source code
@dataclass class TableInsertManyResult(OperationResult): """ Class that represents the result of insert_many operations on a collection. Attributes: raw_results: responses from the Data API calls inserted_ids: list of the IDs of the inserted documents in the form of dictionaries inserted_id_tuple: an ordered-tuple version of the same inserted IDs """ inserted_ids: list[Any] inserted_id_tuples: list[tuple[Any, ...]] def __repr__(self) -> str: _ins_ids_str: str if len(self.inserted_ids) > 5: _ins_ids_str = ( f"[{', '.join(str(_iid) for _iid in self.inserted_ids[:5])} " f"... ({len(self.inserted_ids)} total)]" ) else: _ins_ids_str = str(self.inserted_ids) _ins_idts_str: str if len(self.inserted_id_tuples) > 5: _ins_idts_str = ( f"[{', '.join(str(_iidt) for _iidt in self.inserted_id_tuples[:5])} " f"... ({len(self.inserted_id_tuples)} total)]" ) else: _ins_idts_str = str(self.inserted_id_tuples) return self._piecewise_repr( [ f"inserted_ids={_ins_ids_str}", f"inserted_id_tuples={_ins_idts_str}", "raw_results=..." if self.raw_results is not None else None, ] )
Ancestors
- OperationResult
- abc.ABC
Class variables
var inserted_id_tuples : list[tuple[typing.Any, ...]]
var inserted_ids : list[typing.Any]
class TableInsertOneResult (raw_results: list[dict[str, Any]], inserted_id: Any, inserted_id_tuple: tuple[Any, ...])
-
Class that represents the result of insert_one operations on a table.
Attributes
raw_results
- one-item list with the response from the Data API call
inserted_id
- the ID of the inserted document in the form of a dict
inserted_id_tuple
- an ordered-tuple version of the same ID
Expand source code
@dataclass class TableInsertOneResult(OperationResult): """ Class that represents the result of insert_one operations on a table. Attributes: raw_results: one-item list with the response from the Data API call inserted_id: the ID of the inserted document in the form of a dict inserted_id_tuple: an ordered-tuple version of the same ID """ inserted_id: Any inserted_id_tuple: tuple[Any, ...] def __repr__(self) -> str: return self._piecewise_repr( [ f"inserted_id={self.inserted_id}", f"inserted_id_tuple={self.inserted_id_tuple}", "raw_results=..." if self.raw_results is not None else None, ] )
Ancestors
- OperationResult
- abc.ABC
Class variables
var inserted_id : Any
var inserted_id_tuple : tuple[typing.Any, ...]