MPT-17825: resolve WPS210, WPS221, WPS229, WPS231 products and audit violations

cli/core/products/models/product.py

@@ -74,22 +74,15 @@ def from_dict(cls, row_data: dict[str, Any]) -> Self:
     @classmethod
     @override
     def from_json(cls, json_data: dict[str, Any]) -> Self:
-        formatted_settings = {}
-        for setting_key, raw_setting in json_data.items():
-            if isinstance(raw_setting, dict):
-                for sub_key, sub_value in raw_setting.items():
-                    formatted_settings[f"{setting_key}.{sub_key}"] = sub_value
-            else:
-                formatted_settings[setting_key] = raw_setting
-
-        records = [
-            SettingsRecords.from_json({
-                "name": setting_name,
-                "value": formatted_settings.get(setting_path),
-            })
-            for setting_name, setting_path in constants.SETTINGS_API_MAPPING.items()
-        ]
-        return cls(records=records)
+        return cls(
+            records=[
+                SettingsRecords.from_json({
+                    "name": setting_name,
+                    "value": cls._flatten_settings(json_data).get(setting_path),
+                })
+                for setting_name, setting_path in constants.SETTINGS_API_MAPPING.items()
+            ]
+        )
 
     @override
     def to_json(self) -> dict[str, Any]:
@@ -111,6 +104,17 @@ def to_json(self) -> dict[str, Any]:
     def to_xlsx(self) -> dict[str, Any]:
         return {}
 
+    @classmethod
+    def _flatten_settings(cls, json_data: dict[str, Any]) -> dict[str, Any]:
+        formatted_settings = {}
+        for setting_key, raw_setting in json_data.items():
+            if isinstance(raw_setting, dict):
+                for sub_key, sub_value in raw_setting.items():
+                    formatted_settings[f"{setting_key}.{sub_key}"] = sub_value
+            else:
+                formatted_settings[setting_key] = raw_setting
+        return formatted_settings
+
 
 @dataclass
 class ProductData(BaseDataModel):

cli/core/products/services/product_service.py

@@ -26,16 +26,18 @@ def create(self) -> ServiceResult:
         try:
             new_product_data = self.api.post(form_payload=multipart_payload, headers=headers)
         except MPTAPIError as error:
-            self._set_error(str(error))
-            return ServiceResult(success=False, errors=[str(error)], model=None, stats=self.stats)
+            error_msg = str(error)
+            self._set_error(error_msg)
+            return ServiceResult(success=False, errors=[error_msg], model=None, stats=self.stats)
 
         product.id = new_product_data["id"]
         try:
             # TODO: Handle this gracefully using update_settings function
             self.api.update(f"{product.id}/settings", json_payload=product.settings.to_json())
         except MPTAPIError as error:
-            self._set_error(str(error))
-            return ServiceResult(success=False, errors=[str(error)], model=None, stats=self.stats)
+            error_msg = str(error)
+            self._set_error(error_msg)
+            return ServiceResult(success=False, errors=[error_msg], model=None, stats=self.stats)
 
         self._set_synced(product.id, product.coordinate)
         return ServiceResult(success=True, model=product, stats=self.stats)
@@ -71,8 +73,9 @@ def retrieve(self) -> ServiceResult:
         try:
             exists = self.api.exists({"id": product.id})
         except MPTAPIError as error:
-            self._set_error(str(error))
-            return ServiceResult(success=False, errors=[str(error)], model=None, stats=self.stats)
+            error_msg = str(error)
+            self._set_error(error_msg)
+            return ServiceResult(success=False, errors=[error_msg], model=None, stats=self.stats)
 
         return ServiceResult(success=True, model=product if exists else None, stats=self.stats)
 
@@ -81,7 +84,8 @@ def retrieve_from_mpt(self, resource_id: str) -> ServiceResult:
         try:
             product_data = self.api.get(resource_id)
         except MPTAPIError as error:
-            return ServiceResult(success=False, errors=[str(error)], model=None, stats=self.stats)
+            error_msg = str(error)
+            return ServiceResult(success=False, errors=[error_msg], model=None, stats=self.stats)
 
         product = self.data_model.from_json(product_data)
         return ServiceResult(success=True, model=product, stats=self.stats)
@@ -105,15 +109,17 @@ def validate_definition(self) -> ServiceResult:
             for section_name in error.details:
                 self.stats.errors.add_msg(section_name, "", "Required tab doesn't exist")
 
-            return ServiceResult(success=False, errors=[str(error)], model=None, stats=self.stats)
+            error_msg = str(error)
+            return ServiceResult(success=False, errors=[error_msg], model=None, stats=self.stats)
 
         try:
             self.file_manager.check_required_fields_by_section()
         except RequiredFieldsError as error:
             for field_name in error.details:
                 self.stats.errors.add_msg(field_name, "", "Required field doesn't exist")
 
-            return ServiceResult(success=False, errors=[str(error)], model=None, stats=self.stats)
+            error_msg = str(error)
+            return ServiceResult(success=False, errors=[error_msg], model=None, stats=self.stats)
 
         return ServiceResult(success=True, model=None, stats=self.stats)
 
@@ -133,7 +139,8 @@ def update(self) -> ServiceResult:
         try:
             self.api.update(product.id, SettingsData(records=setting_items).to_json())
         except MPTAPIError as error:
-            self._set_error(str(error))
-            return ServiceResult(success=False, errors=[str(error)], model=None, stats=self.stats)
+            error_msg = str(error)
+            self._set_error(error_msg)
+            return ServiceResult(success=False, errors=[error_msg], model=None, stats=self.stats)
 
         return ServiceResult(success=True, model=product, stats=self.stats)

cli/core/products/services/related_components_base_service.py

@@ -1,7 +1,7 @@
 import logging
 from abc import ABC
 from collections.abc import Callable
-from typing import override
+from typing import Any, override
 
 from cli.core.errors import MPTAPIError
 from cli.core.models import DataCollectionModel
@@ -18,23 +18,13 @@ class RelatedComponentsBaseService(RelatedBaseService, ABC):
 
     @override
     def create(self) -> ServiceResult:
-        errors = []
-        collection = {}
+        errors: list[str] = []
+        collection: dict[str, Any] = {}
         for raw_model_data in self.file_manager.read_data():
             data_model = self.prepare_data_model_to_create(raw_model_data)
-
-            try:
-                new_item = self.api.post(json=data_model.to_json())
-            except MPTAPIError as error:
-                errors.append(str(error))
-                self._set_error(str(error), data_model.id)
+            if self._add_created_item(collection, data_model, errors):
                 continue
 
-            old_id = data_model.id
-            data_model.id = new_item["id"]
-            collection[old_id] = data_model
-            self._set_synced(new_item["id"], data_model.coordinate)
-
         return ServiceResult(
             success=len(errors) == 0,
             errors=errors,
@@ -92,18 +82,10 @@ def update(self) -> ServiceResult:
                 self._set_skipped()
                 continue
 
-            try:
-                action_handler = self._get_update_action_handler(data_model.action)
-            except ValueError as error:
-                errors.append(str(error))
-                self._set_error(str(error), data_model.id)
-                continue
-
-            try:
-                action_handler(data_model)
-            except MPTAPIError as error:
-                errors.append(str(error))
-                self._set_error(str(error), data_model.id)
+            error_msg = self._handle_update_model(data_model)
+            if error_msg is not None:
+                errors.append(error_msg)
+                self._set_error(error_msg, data_model.id)
                 continue
 
             self._set_synced(data_model.id, data_model.coordinate)
@@ -124,6 +106,23 @@ def prepare_data_model_to_create(self, data_model: DataModel) -> DataModel:
         """
         return data_model
 
+    def _add_created_item(
+        self, collection: dict[str, DataModel], data_model: Any, errors: list[str]
+    ) -> bool:
+        try:
+            new_item = self.api.post(json=data_model.to_json())
+        except MPTAPIError as error:
+            error_msg = str(error)
+            errors.append(error_msg)
+            self._set_error(error_msg, data_model.id)
+            return True
+
+        old_id = data_model.id
+        data_model.id = new_item["id"]
+        collection[old_id] = data_model
+        self._set_synced(new_item["id"], data_model.coordinate)
+        return False
+
     def _action_create_item(self, data_model: DataModel):
         """Creates the item in the API.
 
@@ -188,3 +187,14 @@ def _get_update_action_handler(self, model_action: DataActionEnum) -> Callable:
             return self._action_update_item
 
         raise ValueError(f"Invalid action: {model_action}")
+
+    def _handle_update_model(self, data_model: Any) -> str | None:
+        try:
+            action_handler = self._get_update_action_handler(data_model.action)
+        except ValueError as error:
+            return str(error)
+        try:
+            action_handler(data_model)
+        except MPTAPIError as error:
+            return str(error)
+        return None

cli/plugins/audit_plugin/api.py

@@ -2,17 +2,18 @@
 
 import typer
 from cli.core.console import console
-from mpt_api_client import MPTClient, RQLQuery
+from mpt_api_client import RQLQuery
 
+SELECT_FIELDS = ("object", "actor", "details", "documents", "request.api.geolocation")
 
-def get_audit_trail(client: MPTClient, record_id: str) -> dict[str, Any]:
+
+def get_audit_trail(client: Any, record_id: str) -> dict[str, Any]:
     """Retrieve audit trail for a specific record."""
+    # TODO: Using select instead of get because render() query string isn't working with get().
+    audit_collection = client.audit.records.options(render=True)
+    audit_collection_filtered = audit_collection.filter(RQLQuery(id=record_id))
     try:
-        # TODO: Using select instead of get because render() query string isn't working with get().
-        select_fields = ["object", "actor", "details", "documents", "request.api.geolocation"]
-        audit_collection = client.audit.records.options(render=True)
-        audit_collection_filtered = audit_collection.filter(RQLQuery(id=record_id))
-        records = audit_collection_filtered.select(*select_fields).fetch_page(limit=1)
+        records = audit_collection_filtered.select(*SELECT_FIELDS).fetch_page(limit=1)
     except Exception as error:
         console.print(
             f"[red]Failed to retrieve audit trail for record {record_id}: {error!s}[/red]"
@@ -30,24 +31,29 @@ def get_audit_records_by_object(
     client: Any, object_id: str, limit: int = 10
 ) -> list[dict[str, Any]]:
     """Retrieve all audit records for a specific object."""
+    query = _records_query(client, object_id)
     try:
-        select_fields = ["object", "actor", "details", "documents", "request.api.geolocation"]
-        object_id_filter = RQLQuery(object__id=object_id)
-        order_by = "-timestamp"
-        audit_records_collection = client.audit.records.options(render=True)
-        audit_records_filtered = audit_records_collection.filter(object_id_filter)
-        raw_records = (
-            audit_records_filtered.order_by(order_by).select(*select_fields).fetch_page(limit=limit)
-        )
-        records = [raw_record.to_dict() for raw_record in raw_records]
+        raw_records = query.fetch_page(limit=limit)
     except Exception as error:
         console.print(
             f"[red]Failed to retrieve audit records for object {object_id}: {error!s}[/red]"
         )
         raise typer.Exit(1) from error
 
+    records = [raw_record.to_dict() for raw_record in raw_records]
     if not records:
         console.print(f"[red]No audit records found for object {object_id}[/red]")
         raise typer.Exit(1)
 
     return records
+
+
+def _records_query(client: Any, object_id: str) -> Any:
+    object_id_filter = RQLQuery(object__id=object_id)
+    return (
+        client.audit.records
+        .options(render=True)
+        .filter(object_id_filter)
+        .order_by("-timestamp")
+        .select(*SELECT_FIELDS)
+    )