fix: add user details to task_history endpoint

src/backend/app/tasks/tasks_crud.py

@@ -341,7 +341,7 @@ def process_history_entry(history_entry):
     return tasks
 
 
-def get_project_task_history(
+async def get_project_task_history(
     project_id: int,
     end_date: Optional[datetime],
     db: Session,
@@ -413,18 +413,3 @@ async def count_validated_and_mapped_tasks(
         entry.update({"validated": total_validated, "mapped": total_mapped})
 
     return results
-
-
-async def append(tasks:List, db: Session):
-    """Get task history of project."""
-    response = []
-
-    for task in tasks if isinstance(tasks, list) else [tasks]:
-        task_id = task.id
-        task_history = task.task_history
-        if isinstance(task_history, list):
-            for history_entry in task_history:
-                user = db.query(db_models.DbUser).filter_by(id=history_entry.user_id).first()
-                response.append(tasks_schemas.TaskHistory.map_entry_to_model(task_id, history_entry, user))
-
-    return response

src/backend/app/tasks/tasks_routes.py

@@ -206,37 +206,38 @@ async def task_activity(
     """Retrieves the validate and mapped task count for a specific project.
 
     Args:
-        project_id: The ID of the project.
-        days: The number of days to consider for the
-        task activity (default: 10).
-        db: The database session.
+        project_id (int): The ID of the project.
+        days (int): The number of days to consider for the
+            task activity (default: 10).
+        db (Session): The database session.
 
     Returns:
         list[TaskHistoryCount]: A list of task history counts.
 
     """
     end_date = datetime.now() - timedelta(days=days)
-    task_history = tasks_crud.get_project_task_history(project_id, end_date, db)
+    task_history = await tasks_crud.get_project_task_history(project_id, end_date, db)
 
     return await tasks_crud.count_validated_and_mapped_tasks(
         task_history,
         end_date,
     )
 
 
-@router.get("/task_history/", response_model = List[tasks_schemas.TaskHistory])
-async def task_history(project_id: int, days: int = 10, db: Session = Depends(database.get_db)):
-    """
-    Get the detailed task history for a project.
+@router.get("/task_history/", response_model=List[tasks_schemas.TaskHistory])
+async def task_history(
+    project_id: int, days: int = 10, db: Session = Depends(database.get_db)
+):
+    """Get the detailed task history for a project.
 
     Args:
         project_id (int): The ID of the project.
-        db (Session): The database session. (default: Depends(database.get_db))
+        days (int): The number of days to consider for the
+            task activity (default: 10).
+        db (Session): The database session.
 
     Returns:
         List[TaskHistory]: A list of task history.
     """
     end_date = datetime.now() - timedelta(days=days)
-    task_history = tasks_crud.get_project_task_history(project_id, end_date, db)
-
-    return await tasks_crud.get_task_history(tasks, db)
+    return await tasks_crud.get_project_task_history(project_id, end_date, db)

src/backend/app/tasks/tasks_schemas.py

@@ -22,7 +22,7 @@
 
 from geojson_pydantic import Feature as GeojsonFeature
 from loguru import logger as log
-from pydantic import BaseModel, ConfigDict, Field, ValidationInfo
+from pydantic import BaseModel, ConfigDict, Field, ValidationInfo, computed_field
 from pydantic.functional_serializers import field_serializer
 from pydantic.functional_validators import field_validator
 
@@ -96,7 +96,9 @@ def get_geojson_from_outline(cls, value: Any, info: ValidationInfo) -> str:
 
     @field_validator("outline_centroid", mode="before")
     @classmethod
-    def get_centroid_from_outline(cls, value: Any, info: ValidationInfo) -> str:
+    def get_centroid_from_outline(
+        cls, value: Any, info: ValidationInfo
+    ) -> Optional[str]:
         """Get outline_centroid from Shapely geom."""
         if outline := info.data.get("outline"):
             properties = {
@@ -109,21 +111,21 @@ def get_centroid_from_outline(cls, value: Any, info: ValidationInfo) -> str:
         return None
 
     @field_serializer("locked_by_uid")
-    def get_locked_by_uid(self, value: str) -> str:
+    def get_locked_by_uid(self, value: str) -> Optional[str]:
         """Get lock uid from lock_holder details."""
         if self.lock_holder:
             return self.lock_holder.id
         return None
 
     @field_serializer("locked_by_username")
-    def get_locked_by_username(self, value: str) -> str:
+    def get_locked_by_username(self, value: str) -> Optional[str]:
         """Get lock username from lock_holder details."""
         if self.lock_holder:
             return self.lock_holder.username
         return None
 
     @field_serializer("odk_token")
-    def decrypt_password(self, value: str) -> str:
+    def decrypt_password(self, value: str) -> Optional[str]:
         """Decrypt the ODK Token extracted from the db."""
         if not value:
             return ""
@@ -139,20 +141,42 @@ class ReadTask(Task):
 
 class TaskHistory(BaseModel):
     """Task history details."""
+
+    model_config = ConfigDict(
+        from_attributes=True,
+    )
+
+    # Excluded
+    user: Any = Field(exclude=True)
+
     task_id: int
     action_text: str
     action_date: datetime
-    status: str
-    username: str
-    profile_img: Optional[str]
 
-    @classmethod
-    def map_entry_to_model(cls, task_id, history_entry, user):
-        return cls(
-            task_id=task_id,
-            action_text=history_entry.action_text,
-            action_date=history_entry.action_date,
-            status=history_entry.action_text.split()[5],
-            username=user.username if user else None,
-            profile_img=user.profile_img if user else None
-        )
+    @computed_field
+    @property
+    def username(self) -> Optional[str]:
+        """Get username from user db obj."""
+        if self.user:
+            return self.user.username
+        return None
+
+    @computed_field
+    @property
+    def profile_img(self) -> Optional[str]:
+        """Get profile_img from user db obj."""
+        if self.user:
+            return self.user.profile_img
+        return None
+
+    @computed_field
+    @property
+    def status(self) -> Optional[str]:
+        """Extract status from standard format action_text."""
+        if self.action_text:
+            split_text = self.action_text.split()
+            if len(split_text) > 5:
+                return split_text[5]
+            else:
+                return self.action_text
+        return None