Replaced Any usage for better type checking

server/planning/events/events_history_async_service.py

@@ -24,7 +24,7 @@
 
 
 class EventsHistoryAsyncService(HistoryAsyncService[EventsHistoryResourceModel]):
-    async def on_item_created(self, items: list[EventResourceModel | Any], operation: str | None = None):
+    async def on_item_created(self, items: list[dict[str, Any]], operation: str | None = None):
         created_from_planning = []
         regular_events = []
         for item in items:
@@ -59,7 +59,7 @@ async def on_item_updated(self, updates: dict[str, Any], original: dict[str, Any
 
         await self._save_history(item, diff, operation)
 
-    async def _save_history(self, item, update: dict[str, Any], operation: str | None = None):
+    async def _save_history(self, item: dict[str, Any], update: dict[str, Any], operation: str | None = None):
         history = {
             "event_id": item[ID_FIELD],
             "user_id": self.get_user_id(),

server/planning/history_async_service.py

@@ -47,7 +47,7 @@
 class HistoryAsyncService(AsyncResourceService[Generic[HistoryResourceModelType]]):
     """Provide common async methods for tracking history of Creation, Updates and Spiking to collections"""
 
-    async def on_item_created(self, items: list[Any], operation: str | None = None):
+    async def on_item_created(self, items: list[dict[str, Any]], operation: str | None = None):
         for item in items:
             if not item.get("duplicate_from"):
                 await self._save_history(
@@ -56,7 +56,7 @@ async def on_item_created(self, items: list[Any], operation: str | None = None):
                     operation or "create",
                 )
 
-    async def on_item_updated(self, updates: dict[str, Any], original: Any, operation: str | None = None):
+    async def on_item_updated(self, updates: dict[str, Any], original: dict[str, Any], operation: str | None = None):
         item = deepcopy(original)
         if list(item.keys()) == ["_id"]:
             diff = updates
@@ -67,24 +67,24 @@ async def on_item_updated(self, updates: dict[str, Any], original: Any, operatio
 
         await self._save_history(item, diff, operation or "edited")
 
-    async def on_spike(self, updates: dict[str, Any], original: Any):
+    async def on_spike(self, updates: dict[str, Any], original: dict[str, Any]):
         await self.on_item_updated(updates, original, "spiked")
 
-    async def on_unspike(self, updates: dict[str, Any], original: Any):
+    async def on_unspike(self, updates: dict[str, Any], original: dict[str, Any]):
         await self.on_item_updated(updates, original, "unspiked")
 
-    async def on_cancel(self, updates: dict[str, Any], original: Any):
+    async def on_cancel(self, updates: dict[str, Any], original: dict[str, Any]):
         operation = "events_cancel" if original.get(ITEM_TYPE) == "event" else "planning_cancel"
         await self.on_item_updated(updates, original, operation)
 
-    async def on_reschedule(self, updates: dict[str, Any], original: Any):
+    async def on_reschedule(self, updates: dict[str, Any], original: dict[str, Any]):
         await self.on_item_updated(updates, original, "reschedule")
 
-    async def on_reschedule_from(self, item: Any):
+    async def on_reschedule_from(self, item: dict[str, Any]):
         new_item = deepcopy(item)
         await self._save_history({ID_FIELD: str(item[ID_FIELD])}, new_item, "reschedule_from")
 
-    async def on_postpone(self, updates: dict[str, Any], original: Any):
+    async def on_postpone(self, updates: dict[str, Any], original: dict[str, Any]):
         await self.on_item_updated(updates, original, "postpone")
 
     async def get_user_id(self):

server/planning/planning/planning_history_async_service.py

@@ -13,6 +13,7 @@
 from typing import Any
 
 
+from planning.core.service import PlanningResourceModelType
 from planning.types import PlanningHistoryResourceModel
 from superdesk.flask import request
 from superdesk.resource_fields import ID_FIELD
@@ -35,7 +36,7 @@
 
 
 class PlanningHistoryAsyncService(HistoryAsyncService[PlanningHistoryResourceModel]):
-    async def on_item_created(self, items: list[PlanningResourceModel], operation=None):
+    async def on_item_created(self, items: list[dict[str, Any]], operation=None):
         add_to_planning = False
         if request and hasattr(request, "args"):
             add_to_planning = strtobool(request.args.get("add_to_planning", "false"))
@@ -63,14 +64,12 @@ async def _save_history(self, item, update: dict[str, Any], operation: str | Non
 
         await self.create([history])
 
-    async def on_item_updated(
-        self, updates: dict[str, Any], original: PlanningResourceModel, operation: str | None = None
-    ):
-        item = deepcopy(original.to_dict())
+    async def on_item_updated(self, updates: dict[str, Any], original: dict[str, Any], operation: str | None = None):
+        item = deepcopy(original)
         if list(item.keys()) == ["_id"]:
             diff = self._remove_unwanted_fields(updates)
         else:
-            diff = await self._changes(original.to_dict(), updates)
+            diff = await self._changes(original, updates)
             diff.pop("coverages", None)
             if updates:
                 item.update(updates)
@@ -87,9 +86,9 @@ async def on_item_updated(
 
             await self._save_history(item, diff, operation)
 
-        await self._save_coverage_history(updates, original.to_dict())
+        await self._save_coverage_history(updates, original)
 
-    async def on_cancel(self, updates: dict[str, Any], original):
+    async def on_cancel(self, updates: dict[str, Any], original: dict[str, Any]):
         await self.on_item_updated(
             updates,
             original,
@@ -181,7 +180,7 @@ async def _save_coverage_history(self, updates: dict[str, Any], original: dict[s
         for cov in deleted:
             await self._save_history(item, {"coverage_id": cov.get("coverage_id")}, "coverage_deleted")
 
-    async def on_spike(self, updates: dict[str, Any], original: PlanningResourceModel):
+    async def on_spike(self, updates: dict[str, Any], original: dict[str, Any]):
         """Spike event
 
         On spike of a planning item the history of any agendas that the item belongs to will have an entry added to
@@ -192,7 +191,7 @@ async def on_spike(self, updates: dict[str, Any], original: PlanningResourceMode
         """
         await super().on_spike(updates, original)
 
-    async def on_unspike(self, updates: dict[str, Any], original: PlanningResourceModel):
+    async def on_unspike(self, updates: dict[str, Any], original: dict[str, Any]):
         await super().on_unspike(updates, original)
 
     async def on_duplicate(self, parent: dict[str, Any], duplicate: dict[str, Any]):