Touch parents updated_at when updating a child && update tests

Author: Marishka17

packages/examples/cvat/exchange-oracle/src/crons/cvat/state_trackers.py

@@ -77,6 +77,11 @@ def track_completed_tasks(logger: logging.Logger, session: Session) -> None:
             completed_task_ids.append(task.cvat_id)
 
     if completed_task_ids:
+        # TODO: for_update
+        cvat_service.touch_projects(
+            session, {t.project.id for t in tasks if t.cvat_id in completed_task_ids}
+        )
+
         logger.info(
             "Found new completed tasks: {}".format(", ".join(str(t) for t in completed_task_ids))
         )
@@ -118,6 +123,16 @@ def track_assignments(logger: logging.Logger) -> None:
 
             cvat_service.expire_assignment(session, assignment.id)
 
+        jobs_to_be_updated = [a.job for a in assignments]
+        tasks_to_be_updated = [j.task for j in jobs_to_be_updated]
+        projects_to_be_updated = [t.project for t in tasks_to_be_updated]
+        cvat_service.touch_jobs(session, {job.id for job in jobs_to_be_updated})
+        cvat_service.touch_tasks(session, {task.id for task in tasks_to_be_updated})
+        cvat_service.touch_projects(session, {project.id for project in projects_to_be_updated})
+        del jobs_to_be_updated
+        del tasks_to_be_updated
+        del projects_to_be_updated
+
     with SessionLocal.begin() as session:
         assignments = cvat_service.get_active_assignments(
             session,
@@ -148,6 +163,14 @@ def track_assignments(logger: logging.Logger) -> None:
 
                 cvat_service.cancel_assignment(session, assignment.id)
 
+        # touch jobs/tasks/projects updated_at
+        jobs_to_be_updated = [a.job for a in assignments]
+        tasks_to_be_updated = [j.task for j in jobs_to_be_updated]
+        projects_to_be_updated = [t.project for t in tasks_to_be_updated]
+        cvat_service.touch_jobs(session, {job.id for job in jobs_to_be_updated})
+        cvat_service.touch_tasks(session, {task.id for task in tasks_to_be_updated})
+        cvat_service.touch_projects(session, {project.id for project in projects_to_be_updated})
+
 
 @cron_job
 def track_completed_escrows(logger: logging.Logger) -> None:
@@ -213,6 +236,7 @@ def track_task_creation(logger: logging.Logger, session: Session) -> None:
                     )
 
                 completed.append(upload)
+                upload.task.touch(session, touch_parent=True)
             except cvat_api.exceptions.ApiException as e:
                 failed.append(upload)
 

packages/examples/cvat/exchange-oracle/src/endpoints/exchange.py

@@ -56,9 +56,9 @@ class JobsFilter(Filter):
 
     class SortingFields(StrEnum, metaclass=BetterEnumMeta):
         created_at = auto()
+        updated_at = auto()
         chain_id = auto()
         job_type = auto()
-        reward_amount = auto()
 
     sort: OptionalQuery[OrderingDirection] = OrderingDirection.asc
     default_sort_field: ClassVar[SortingFields] = SortingFields.created_at
@@ -227,7 +227,6 @@ class SortingFields(StrEnum, metaclass=BetterEnumMeta):
         chain_id = auto()
         job_type = auto()
         status = auto()
-        reward_amount = auto()
         created_at = auto()
         expires_at = auto()
 

packages/examples/cvat/exchange-oracle/src/endpoints/serializers.py

@@ -2,7 +2,6 @@
 from typing import Literal
 
 from human_protocol_sdk.storage import StorageFileNotFoundError
-from sqlalchemy import func
 from sqlalchemy.orm import Session
 
 import src.services.cvat as cvat_service
@@ -54,11 +53,6 @@ def serialize_job(
         else:
             raise NotImplementedError(f"Unknown status {project.status}")
 
-        updated_at = (
-            session.query(func.max(cvat_service.Job.updated_at))
-            .filter(cvat_service.Job.cvat_project_id == project.cvat_id)
-            .scalar()
-        )
         return service_api.JobResponse(
             escrow_address=project.escrow_address,
             chain_id=project.chain_id,
@@ -70,7 +64,7 @@ def serialize_job(
                 service_api.DEFAULT_TOKEN
             ),  # set a value to avoid being excluded by response_model_exclude_unset=True
             created_at=project.created_at,
-            updated_at=updated_at,
+            updated_at=project.updated_at,
             qualifications=manifest.qualifications,
         )
 

packages/examples/cvat/exchange-oracle/src/handlers/cvat_events.py

@@ -66,6 +66,7 @@ def handle_update_job_event(payload: dict) -> None:
                             "Assignment is expired, rejecting the update"
                         )
                         cvat_service.expire_assignment(session, matching_assignment.id)
+                        matching_assignment.job.touch(session, touch_parent=True)
 
                         if matching_assignment.id == latest_assignment.id:
                             cvat_api.update_job_assignee(job.cvat_id, assignee_id=None)
@@ -88,6 +89,7 @@ def handle_update_job_event(payload: dict) -> None:
                         session, matching_assignment.id, completed_at=webhook_time
                     )
                     cvat_service.update_job_status(session, job.id, new_status)
+                    job.task.touch(session, touch_parent=True)
 
                     cvat_api.update_job_assignee(job.cvat_id, assignee_id=None)
 
@@ -122,6 +124,12 @@ def handle_create_job_event(payload: dict) -> None:
                 payload.job["project_id"],
                 status=JobStatuses[payload.job["state"]],
             )
+            cvat_service.touch_tasks(
+                session, [payload.job["task_id"]], field=cvat_service.Task.cvat_id
+            )
+            cvat_service.touch_projects(
+                session, [payload.job["project_id"]], field=cvat_service.Project.cvat_id
+            )
 
         try:
             projects = cvat_service.get_projects_by_cvat_ids(

packages/examples/cvat/exchange-oracle/src/handlers/job_creation.py

@@ -358,6 +358,7 @@ def build(self):
                 )
 
                 db_service.create_data_upload(session, cvat_task.id)
+            db_service.touch_projects(session, [project_id])
 
 
 class BoxesFromPointsTaskBuilder:
@@ -1394,6 +1395,7 @@ def _create_on_cvat(self):
                 )
 
                 db_service.create_data_upload(session, cvat_task.id)
+            db_service.touch_projects(session, [project_id])
 
     @classmethod
     def _make_cloud_storage_client(cls, bucket_info: BucketAccessInfo) -> StorageClient:
@@ -2466,6 +2468,7 @@ def _job_params_label_key(ts):
                         )
 
                         db_service.create_data_upload(session, cvat_task.id)
+                    db_service.touch_projects(session, [project_id])
 
     @classmethod
     def _make_cloud_storage_client(cls, bucket_info: BucketAccessInfo) -> StorageClient:

packages/examples/cvat/exchange-oracle/src/models/cvat.py

@@ -2,7 +2,7 @@
 from __future__ import annotations
 
 from sqlalchemy import Column, DateTime, Enum, ForeignKey, Integer, String, UniqueConstraint
-from sqlalchemy.orm import Mapped, relationship
+from sqlalchemy.orm import Mapped, Session, relationship
 from sqlalchemy.sql import func
 
 from src.core.types import (
@@ -65,6 +65,9 @@ class Project(Base):
     def __repr__(self) -> str:
         return f"Project. id={self.id}"
 
+    def touch(self, session: Session) -> None:
+        session.query(Project).filter(Project.id == self.id).update({Project.updated_at: utcnow()})
+
 
 class Task(Base):
     __tablename__ = "tasks"
@@ -92,6 +95,12 @@ class Task(Base):
     def __repr__(self) -> str:
         return f"Task. id={self.id}"
 
+    def touch(self, session: Session, *, touch_parent: bool = True) -> None:
+        session.query(Task).filter(Task.id == self.id).update({Task.updated_at: utcnow()})
+
+        if touch_parent:
+            self.project.touch(session)
+
 
 class EscrowCreation(Base):
     __tablename__ = "escrow_creations"
@@ -170,6 +179,13 @@ def latest_assignment(self) -> Assignment | None:
     def __repr__(self) -> str:
         return f"Job. id={self.id}"
 
+    def touch(self, session: Session, *, touch_parent: bool = True) -> None:
+        # TODO: check .update({})
+        session.query(Job).filter(Job.id == self.id).update({Job.updated_at: utcnow()})
+
+        if touch_parent:
+            self.task.touch(session, touch_parent=touch_parent)
+
 
 class User(Base):
     __tablename__ = "users"

packages/examples/cvat/exchange-oracle/src/services/cvat.py

@@ -1,9 +1,11 @@
 import itertools
 import uuid
-from collections.abc import Sequence
+from collections.abc import Iterable, Sequence
 from datetime import datetime
+from itertools import islice
+from typing import Any
 
-from sqlalchemy import delete, insert, update
+from sqlalchemy import Column, delete, insert, update
 from sqlalchemy.orm import Session
 
 from src.core.types import AssignmentStatuses, JobStatuses, ProjectStatuses, TaskStatuses, TaskTypes
@@ -13,6 +15,13 @@
 from src.utils.time import utcnow
 
 
+def batched(iterable: Iterable, *, batch_size: int) -> Iterable[Any]:
+    assert batch_size > 0
+    iterator = iter(iterable)
+    while batch := tuple(islice(iterator, batch_size)):
+        yield batch
+
+
 # Project
 def create_project(
     session: Session,
@@ -225,6 +234,20 @@ def update_project_statuses_by_escrow_address(
     session.execute(statement)
 
 
+def touch_projects(
+    session: Session,
+    list_to_update: Iterable[str | int],
+    *,
+    field: Column = Project.id,
+    batch_size: int = 1000,
+) -> None:
+    assert field is Project.id or field is Project.cvat_id
+    updated_at = utcnow()
+    for batch in batched(list_to_update, batch_size=batch_size):
+        stmt = update(Project).where(field.in_(batch)).values({Project.updated_at: updated_at})
+        session.execute(stmt)
+
+
 def delete_project(session: Session, project_id: str) -> None:
     project = session.query(Project).filter_by(id=project_id).first()
     session.delete(project)
@@ -392,11 +415,25 @@ def get_tasks_by_status(
     return query.all()
 
 
-def update_task_status(session: Session, task_id: int, status: TaskStatuses) -> None:
+def update_task_status(session: Session, task_id: str, status: TaskStatuses) -> None:
     upd = update(Task).where(Task.id == task_id).values(status=status.value)
     session.execute(upd)
 
 
+def touch_tasks(
+    session: Session,
+    list_to_update: Iterable[str | int],
+    *,
+    field: Column = Task.id,
+    batch_size: int = 1000,
+) -> None:
+    assert field is Task.id or field is Task.cvat_id
+    updated_at = utcnow()
+    for batch in batched(list_to_update, batch_size=batch_size):
+        stmt = update(Task).where(field.in_(batch)).values({Task.updated_at: updated_at})
+        session.execute(stmt)
+
+
 def get_tasks_by_cvat_project_id(
     session: Session, cvat_project_id: int, *, for_update: bool | ForUpdateParams = False
 ) -> list[Task]:
@@ -486,11 +523,25 @@ def get_jobs_by_cvat_id(
     )
 
 
-def update_job_status(session: Session, job_id: int, status: JobStatuses) -> None:
+def update_job_status(session: Session, job_id: str, status: JobStatuses) -> None:
     upd = update(Job).where(Job.id == job_id).values(status=status.value)
     session.execute(upd)
 
 
+def touch_jobs(
+    session: Session,
+    list_to_update: Iterable[str | int],
+    *,
+    field: Column = Job.id,
+    batch_size: int = 1000,
+) -> None:
+    assert field is Job.id or field is Job.cvat_id
+    updated_at = utcnow()
+    for batch in batched(list_to_update, batch_size=batch_size):
+        stmt = update(Job).where(field.in_(batch)).values({Job.updated_at: updated_at})
+        session.execute(stmt)
+
+
 def get_jobs_by_cvat_task_id(
     session: Session, cvat_task_id: int, *, for_update: bool | ForUpdateParams = False
 ) -> list[Job]:

packages/examples/cvat/exchange-oracle/src/services/exchange.py

@@ -62,6 +62,10 @@ def create_assignment(escrow_address: str, chain_id: Networks, wallet_address: s
         if not unassigned_job:
             return None
 
+        cvat_service.get_task_by_id(
+            session, unassigned_job.task.id, for_update=True
+        )  # lock the row
+
         assignment_id = cvat_service.create_assignment(
             session,
             wallet_address=user.wallet_address,
@@ -70,6 +74,8 @@ def create_assignment(escrow_address: str, chain_id: Networks, wallet_address: s
             + timedelta(seconds=get_default_assignment_timeout(TaskTypes(project.job_type))),
         )
 
+        unassigned_job.touch(session)  # project|task|job rows are locked for update
+
         with cvat_api.api_client_context(cvat_api.get_api_client()):
             cvat_api.clear_job_annotations(unassigned_job.cvat_id)
             cvat_api.restart_job(unassigned_job.cvat_id, assignee_id=user.cvat_id)
@@ -104,3 +110,12 @@ async def resign_assignment(assignment_id: str, wallet_address: str) -> None:
             raise NoAccessError
 
         cvat_service.cancel_assignment(session, assignment_id)
+
+        job = assignment.job
+        task = job.task
+        project = task.project
+        cvat_service.get_job_by_id(session, job.id, for_update=True)  # lock the row
+        cvat_service.get_task_by_id(session, task.id, for_update=True)  # lock the row
+        cvat_service.get_project_by_id(session, project.id, for_update=True)  # lock the row
+
+        assignment.job.touch(session)  # project|task rows are locked for update