update coverage and lock

Author: collerek

ormar/fields/parsers.py

@@ -14,7 +14,7 @@
 
 from ormar.utils.rust_utils import HAS_RUST, ormar_rust_utils
 
-if HAS_RUST:
+if HAS_RUST:  # pragma: no cover
     _rs_encode_bytes = ormar_rust_utils.encode_bytes
     _rs_decode_bytes = ormar_rust_utils.decode_bytes
     _rs_encode_json = ormar_rust_utils.encode_json
@@ -34,7 +34,7 @@ def encode_decimal(value: decimal.Decimal, precision: Optional[int] = None) -> f
     )
 
 
-def _py_encode_bytes(
+def _py_encode_bytes(  # pragma: no cover
     value: Union[str, bytes], represent_as_string: bool = False
 ) -> str:
     if represent_as_string:
@@ -46,21 +46,23 @@ def _py_encode_bytes(
     return value
 
 
-def _py_decode_bytes(value: str, represent_as_string: bool = False) -> bytes:
+def _py_decode_bytes(
+    value: str, represent_as_string: bool = False
+) -> bytes:  # pragma: no cover
     if represent_as_string:
         return value if isinstance(value, bytes) else base64.b64decode(value)
     return value if isinstance(value, bytes) else value.encode("utf-8")
 
 
-def _py_encode_json(value: Any) -> Optional[str]:
+def _py_encode_json(value: Any) -> Optional[str]:  # pragma: no cover
     if isinstance(value, (datetime.date, datetime.datetime, datetime.time)):
         value = value.isoformat()
     value = json.dumps(value) if not isinstance(value, str) else re_dump_value(value)
     value = value.decode("utf-8") if isinstance(value, bytes) else value
     return value
 
 
-def re_dump_value(value: str) -> Union[str, bytes]:
+def re_dump_value(value: str) -> Union[str, bytes]:  # pragma: no cover
     """
     Re-dumps value due to different string representation in orjson and json
     :param value: string to re-dump

ormar/models/helpers/models.py

@@ -6,7 +6,7 @@
 from ormar.models.helpers.pydantic import populate_pydantic_default_values
 from ormar.utils.rust_utils import HAS_RUST, ormar_rust_utils
 
-if HAS_RUST:
+if HAS_RUST:  # pragma: no cover
     _rs_group_related_list = ormar_rust_utils.group_related_list
 
 if TYPE_CHECKING:  # pragma no cover
@@ -115,23 +115,24 @@ def group_related_list(list_: list) -> dict:
     :return: list converted to dictionary to avoid repetition and group nested models
     :rtype: dict[str, list]
     """
-    if HAS_RUST:
+    if HAS_RUST:  # pragma: no cover
         return _rs_group_related_list(list_)
-    groups: dict[str, list[str]] = {}
-    for item in list_:
-        head, _, tail = item.partition("__")
-        groups.setdefault(head, [])
-        if tail:
-            groups[head].append(tail)
-
-    result_dict: dict[str, Any] = {}
-    for key in sorted(groups):
-        children = sorted(groups[key])
-        if any("__" in x for x in children):
-            result_dict[key] = group_related_list(children)
-        else:
-            result_dict.setdefault(key, []).extend(children)
-    return dict(sorted(result_dict.items(), key=lambda item: len(item[1])))
+    else:  # pragma: no cover
+        groups: dict[str, list[str]] = {}
+        for item in list_:
+            head, _, tail = item.partition("__")
+            groups.setdefault(head, [])
+            if tail:
+                groups[head].append(tail)
+
+        result_dict: dict[str, Any] = {}
+        for key in sorted(groups):
+            children = sorted(groups[key])
+            if any("__" in x for x in children):
+                result_dict[key] = group_related_list(children)
+            else:
+                result_dict.setdefault(key, []).extend(children)
+        return dict(sorted(result_dict.items(), key=lambda item: len(item[1])))
 
 
 def config_field_not_set(model: type["Model"], field_name: str) -> bool:

ormar/models/mixins/merge_mixin.py

@@ -4,7 +4,7 @@
 from ormar.queryset.utils import translate_list_to_dict
 from ormar.utils.rust_utils import HAS_RUST, ormar_rust_utils
 
-if HAS_RUST:
+if HAS_RUST:  # pragma: no cover
     _rs_group_by_pk = ormar_rust_utils.group_by_pk
     _rs_plan_merge = ormar_rust_utils.plan_merge_items_lists
 
@@ -62,14 +62,14 @@ def merge_instances_list(cls, result_rows: list["Model"]) -> list["Model"]:
         """
         merged_rows: list["Model"] = []
 
-        if HAS_RUST and result_rows:
+        if HAS_RUST and result_rows:  # pragma: no cover
             pks = [model.pk for model in result_rows]
             index_groups = _rs_group_by_pk(pks)
             for group_indices in index_groups:
                 group = [result_rows[i] for i in group_indices]
                 model = cls._recursive_add(group)[0]
                 merged_rows.append(model)
-        else:
+        else:  # pragma: no cover
             grouped_instances: dict = {}
             for model in result_rows:
                 grouped_instances.setdefault(model.pk, []).append(model)
@@ -161,7 +161,7 @@ def _merge_items_lists(
         :return: merged list of models
         :rtype: list[Model]
         """
-        if HAS_RUST:
+        if HAS_RUST:  # pragma: no cover
             current_pks = [getattr(m, "pk", None) for m in current_field]
             other_pks = [getattr(m, "pk", None) for m in other_value]
             plan = _rs_plan_merge(current_pks, other_pks)
@@ -183,26 +183,26 @@ def _merge_items_lists(
                 else:
                     value_to_set.append(cur_item)
             return value_to_set
+        else:  # pragma: no cover
+            other_by_pk: dict = {}
+            for idx, item in enumerate(other_value):
+                other_by_pk.setdefault(item.pk, idx)
 
-        other_by_pk: dict = {}
-        for idx, item in enumerate(other_value):
-            other_by_pk.setdefault(item.pk, idx)
-
-        value_to_set = list(other_value)
-        for cur_field in current_field:
-            other_idx = other_by_pk.get(cur_field.pk)
-            if other_idx is not None:
-                old_value = other_value[other_idx]
-                new_val = cls.merge_two_instances(
-                    cur_field,
-                    cast("Model", old_value),
-                    relation_map=cur_field._skip_ellipsis(  # type: ignore
-                        relation_map, field_name, default_return=dict()
-                    ),
-                )
-                value_to_set = [x for x in value_to_set if x.pk != cur_field.pk] + [
-                    new_val
-                ]
-            else:
-                value_to_set.append(cur_field)
-        return value_to_set
+            value_to_set = list(other_value)
+            for cur_field in current_field:
+                other_idx = other_by_pk.get(cur_field.pk)
+                if other_idx is not None:
+                    old_value = other_value[other_idx]
+                    new_val = cls.merge_two_instances(
+                        cur_field,
+                        cast("Model", old_value),
+                        relation_map=cur_field._skip_ellipsis(  # type: ignore
+                            relation_map, field_name, default_return=dict()
+                        ),
+                    )
+                    value_to_set = [x for x in value_to_set if x.pk != cur_field.pk] + [
+                        new_val
+                    ]
+                else:
+                    value_to_set.append(cur_field)
+            return value_to_set

ormar/queryset/queries/prefetch_query.py

@@ -9,7 +9,7 @@
 from ormar.queryset.utils import translate_list_to_dict
 from ormar.utils.rust_utils import HAS_RUST, ormar_rust_utils
 
-if HAS_RUST:
+if HAS_RUST:  # pragma: no cover
     _RsUniqueList = ormar_rust_utils.UniqueList
     _rs_hash_item = ormar_rust_utils.hash_item
 
@@ -21,7 +21,7 @@
 logger = logging.getLogger(__name__)
 
 
-class _PyUniqueList(list):
+class _PyUniqueList(list):  # pragma: no cover
     """
     Simple subclass of list that prevents the duplicates.
     Uses a set for O(1) membership checks instead of O(n) list scan.
@@ -413,7 +413,7 @@ def _instantiate_models(self) -> None:
             )
             self.models.append(instance)
 
-    def _hash_item(self, item: Union[dict, list]) -> tuple:
+    def _hash_item(self, item: Union[dict, list]) -> tuple:  # pragma: no cover
         """
         Converts model dictionary or list into a hashable tuple to allow its use
         as a dictionary key - used to ensure unique instances of related models.

ormar/queryset/utils.py

@@ -4,14 +4,14 @@
 
 from ormar.utils.rust_utils import HAS_RUST, ormar_rust_utils
 
-if HAS_RUST:
+if HAS_RUST:  # pragma: no cover
     _rs_translate_list_to_dict = ormar_rust_utils.translate_list_to_dict
 
 if TYPE_CHECKING:  # pragma no cover
     from ormar import BaseField, Model
 
 
-def check_node_not_dict_or_not_last_node(
+def check_node_not_dict_or_not_last_node(  # pragma: no cover
     part: str, is_last: bool, current_level: Any
 ) -> bool:
     """
@@ -55,24 +55,25 @@ def translate_list_to_dict(  # noqa: CCR001
     :return: converted to dictionary input list
     :rtype: dict
     """
-    if HAS_RUST:
+    if HAS_RUST:  # pragma: no cover
         return _rs_translate_list_to_dict(list(list_to_trans), default)
-    is_ellipsis = default is ...
-    new_dict: dict = dict()
-    for path in list_to_trans:
-        current_level = new_dict
-        parts = path.split("__")
-        def_val: Any = default if is_ellipsis else copy.deepcopy(default)
-        for ind, part in enumerate(parts):
-            is_last = ind == len(parts) - 1
-            if check_node_not_dict_or_not_last_node(
-                part=part, is_last=is_last, current_level=current_level
-            ):
-                current_level[part] = dict()
-            elif part not in current_level:
-                current_level[part] = def_val
-            current_level = current_level[part]
-    return new_dict
+    else:  # pragma: no cover
+        is_ellipsis = default is ...
+        new_dict: dict = dict()
+        for path in list_to_trans:
+            current_level = new_dict
+            parts = path.split("__")
+            def_val: Any = default if is_ellipsis else copy.deepcopy(default)
+            for ind, part in enumerate(parts):
+                is_last = ind == len(parts) - 1
+                if check_node_not_dict_or_not_last_node(
+                    part=part, is_last=is_last, current_level=current_level
+                ):
+                    current_level[part] = dict()
+                elif part not in current_level:
+                    current_level[part] = def_val
+                current_level = current_level[part]
+        return new_dict
 
 
 def convert_set_to_required_dict(set_to_convert: set) -> dict:

ormar/utils/rust_utils.py

@@ -1,6 +1,6 @@
 """Centralized import for the optional ormar_rust_utils Rust accelerator."""
 
-try:
+try:  # pragma: no cover
     import ormar_rust_utils
 
     HAS_RUST = True