[CVAT] Oracle API updates (#2812)

zhiltsov-max · web-flow · commit 878d76f430dd · 2024-11-18T18:22:30.000+03:00
* Hide escrows in internal states from /job output

* Update final assignments on escrow finish
diff --git a/packages/examples/cvat/exchange-oracle/src/crons/webhooks/recording_oracle.py b/packages/examples/cvat/exchange-oracle/src/crons/webhooks/recording_oracle.py
@@ -82,6 +82,12 @@ def handle_recording_oracle_event(webhook: Webhook, *, db_session: Session, logg
 
                     cvat_db_service.update_project_status(db_session, project.id, new_status)
 
+                cvat_db_service.touch_final_assignments(
+                    db_session,
+                    cvat_project_ids=[p.cvat_id for p in projects_chunk],
+                    touch_parents=True,
+                )
+
             cvat_db_service.update_escrow_validation(
                 db_session,
                 webhook.escrow_address,
diff --git a/packages/examples/cvat/exchange-oracle/src/endpoints/exchange.py b/packages/examples/cvat/exchange-oracle/src/endpoints/exchange.py
@@ -27,7 +27,6 @@
 from src.endpoints.pagination import Page, paginate
 from src.endpoints.serializers import (
     ASSIGNMENT_PROJECT_VALIDATION_STATUSES,
-    PROJECT_ACTIVE_STATUSES,
     PROJECT_COMPLETED_STATUSES,
     serialize_assignment,
     serialize_job,
@@ -117,8 +116,16 @@ async def list_jobs(
 
     query = select(cvat_service.Project)
 
+    # These states are internal, they should not be visible through the API
     query = query.filter(
-        cvat_service.Project.status.not_in([ProjectStatuses.creation, ProjectStatuses.deleted])
+        cvat_service.Project.status.not_in(
+            [
+                ProjectStatuses.creation,
+                ProjectStatuses.deleted,
+                ProjectStatuses.completed,
+                ProjectStatuses.validation,
+            ]
+        )
     )
 
     # We need only high-level jobs (i.e. escrows) without project details
@@ -145,7 +152,7 @@ async def list_jobs(
     if status:
         match status:
             case JobStatuses.active:
-                query = query.filter(cvat_service.Project.status.in_(PROJECT_ACTIVE_STATUSES))
+                query = query.filter(cvat_service.Project.status == ProjectStatuses.annotation)
             case JobStatuses.canceled:
                 query = query.filter(
                     cvat_service.Project.status == cvat_service.ProjectStatuses.canceled
diff --git a/packages/examples/cvat/exchange-oracle/src/endpoints/serializers.py b/packages/examples/cvat/exchange-oracle/src/endpoints/serializers.py
@@ -12,11 +12,6 @@
 from src.schemas import exchange as service_api
 from src.utils.assignments import compose_assignment_url, parse_manifest
 
-PROJECT_ACTIVE_STATUSES = {
-    ProjectStatuses.annotation,
-    ProjectStatuses.completed,
-    ProjectStatuses.validation,
-}
 PROJECT_COMPLETED_STATUSES = {
     ProjectStatuses.recorded,
     ProjectStatuses.deleted,
@@ -50,12 +45,12 @@ def serialize_job(
 
         if project.status == ProjectStatuses.canceled:
             api_status = service_api.JobStatuses.canceled
-        elif project.status in PROJECT_ACTIVE_STATUSES:
+        elif project.status == ProjectStatuses.annotation:
             api_status = service_api.JobStatuses.active
         elif project.status in PROJECT_COMPLETED_STATUSES:
             api_status = service_api.JobStatuses.completed
         else:
-            raise NotImplementedError(f"Unknown status {project.status}")
+            raise AssertionError(f"Unexpected project status '{project.status}'")
 
         return service_api.JobResponse(
             escrow_address=project.escrow_address,
diff --git a/packages/examples/cvat/exchange-oracle/src/models/cvat.py b/packages/examples/cvat/exchange-oracle/src/models/cvat.py
@@ -222,7 +222,7 @@ def __repr__(self) -> str:
         return f"User. wallet_address={self.wallet_address} cvat_id={self.cvat_id}"
 
 
-class Assignment(BaseUUID):
+class Assignment(ChildOf[Job]):
     __tablename__ = "assignments"
     created_at = Column(DateTime(timezone=True), server_default=func.now(), nullable=False)
     updated_at = Column(
diff --git a/packages/examples/cvat/exchange-oracle/src/services/cvat.py b/packages/examples/cvat/exchange-oracle/src/services/cvat.py
@@ -979,7 +979,19 @@ def touch(
         time = utcnow()
 
     session.execute(update(cls).where(cls.id.in_(ids)).values({cls.updated_at: time}))
-    while touch_parents and issubclass(cls, ChildOf):
+
+    if touch_parents:
+        touch_parent_objects(session, cls, ids, time=time)
+
+
+def touch_parent_objects(
+    session: Session,
+    cls: type["Base"],
+    ids: list[str],
+    *,
+    time: datetime | None = None,
+):
+    while issubclass(cls, ChildOf):
         parent_cls = cls.parent_cls
         foreign_key_column = next(iter(cls.parent.property.local_columns))
         parent_id_column = next(iter(foreign_key_column.foreign_keys)).column
@@ -997,3 +1009,49 @@ def touch(
         )
         ids = session.execute(parent_update_stmt).scalars().all()
         cls = parent_cls
+
+
+def touch_final_assignments(
+    session: Session,
+    cvat_project_ids: list[int],
+    *,
+    touch_parents: bool = True,
+    time: datetime | None = None,
+) -> None:
+    if time is None:
+        time = utcnow()
+
+    last_assignment_time_per_job_id_subquery = (
+        select(
+            Assignment.cvat_job_id.label("cvat_job_id"),
+            func.max(Assignment.created_at).label("max_created_at"),
+        )
+        .join(Job)
+        .where(
+            Assignment.status == AssignmentStatuses.completed,
+            Job.cvat_project_id.in_(cvat_project_ids),
+        )
+        .order_by(Assignment.cvat_job_id)
+        .group_by(Assignment.cvat_job_id)
+        .subquery()
+    )
+
+    last_assignment_ids_query = (
+        select(Assignment.id)
+        .join(
+            last_assignment_time_per_job_id_subquery,
+            Assignment.cvat_job_id == last_assignment_time_per_job_id_subquery.c.cvat_job_id,
+            isouter=True,
+        )
+        .where(Assignment.created_at == last_assignment_time_per_job_id_subquery.c.max_created_at)
+    )
+
+    ids = session.execute(
+        update(Assignment)
+        .where(Assignment.id.in_(last_assignment_ids_query))
+        .values({Assignment.updated_at: time})
+        .returning(Assignment.id)
+    )
+
+    if touch_parents:
+        touch_parent_objects(session, Assignment, ids.scalars().all(), time=time)
diff --git a/packages/examples/cvat/exchange-oracle/tests/api/test_exchange_api.py b/packages/examples/cvat/exchange-oracle/tests/api/test_exchange_api.py
@@ -406,9 +406,7 @@ def test_can_list_jobs_200_with_filters(client: TestClient, session: Session):
             ProjectStatuses.validation,
             ProjectStatuses.canceled,
             ProjectStatuses.recorded,
-            # TODO: ProjectStatuses.deleted,
-            # raise NotImplementedError(f"Unknown status {project.status}")
-            # NotImplementedError: Unknown status deleted
+            ProjectStatuses.deleted,
         ]
     ):
         cvat_project = create_project(
@@ -437,7 +435,20 @@ def test_can_list_jobs_200_with_filters(client: TestClient, session: Session):
         session.add(assignment)
         assignments.append(assignment)
         cvat.touch(session, Job, [cvat_job.id])
-        session.commit()  # imitate different created_dates
+        session.commit()  # TODO: imitate different created_dates
+
+    visible_projects_ids = set(
+        p.cvat_id
+        for p in cvat_projects
+        if p.status
+        not in [
+            ProjectStatuses.creation,
+            ProjectStatuses.completed,
+            ProjectStatuses.validation,
+            ProjectStatuses.deleted,
+        ]
+    )
+    visible_projects_count = len(visible_projects_ids)
 
     middle_init_time = utcnow()
 
@@ -460,23 +471,28 @@ def test_can_list_jobs_200_with_filters(client: TestClient, session: Session):
             "status": (
                 (
                     APIJobStatuses.active.value,
-                    3,
-                ),  # ProjectStatuses::annotation,completed,validation
-                (APIJobStatuses.completed.value, 1),
-                (APIJobStatuses.canceled.value, 1),
+                    1,
+                    # ProjectStatuses.annotation
+                    # completed, validation are internal, so hidden
+                ),
+                (APIJobStatuses.completed.value, 1),  # ProjectStatuses.recorded
+                (APIJobStatuses.canceled.value, 1),  # ProjectStatuses.canceled
             ),
-            "chain_id": ((cvat_projects[0].chain_id, len(cvat_projects)),),
+            "chain_id": ((cvat_projects[0].chain_id, visible_projects_count),),
             "job_type": (
-                (cvat_projects[0].job_type, len(cvat_projects)),
+                (cvat_projects[0].job_type, visible_projects_count),
                 (TaskTypes.image_boxes_from_points.value, 0),
             ),
             "created_after": (
-                (str(pre_init_time - timedelta(minutes=1)), len(cvat_projects)),
+                (str(pre_init_time - timedelta(minutes=1)), visible_projects_count),
                 (str(post_init_time), 0),
             ),
             "updated_after": (
-                (str(pre_init_time - timedelta(minutes=1)), len(cvat_projects)),
-                (str(middle_init_time), len(updated_cvat_project_ids)),
+                (str(pre_init_time - timedelta(minutes=1)), visible_projects_count),
+                (
+                    str(middle_init_time),
+                    len(visible_projects_ids.intersection(updated_cvat_project_ids)),
+                ),
                 (str(post_init_time + timedelta(minutes=1)), 0),
             ),
         }.items():
diff --git a/packages/examples/cvat/exchange-oracle/tests/integration/cron/test_process_recording_oracle_webhooks.py b/packages/examples/cvat/exchange-oracle/tests/integration/cron/test_process_recording_oracle_webhooks.py
@@ -29,6 +29,7 @@
 from src.utils.time import utcnow
 
 from tests.utils.constants import DEFAULT_MANIFEST_URL, RECORDING_ORACLE_ADDRESS
+from tests.utils.db_helper import create_project_task_and_job
 
 escrow_address = "0x86e83d346041E8806e352681f3F14549C0d2BC67"
 chain_id = Networks.localhost.value
@@ -42,22 +43,59 @@ def tearDown(self):
         self.session.close()
 
     def test_process_incoming_recording_oracle_webhooks_job_completed_type(self):
-        project_id = str(uuid.uuid4())
-        cvat_project = Project(
-            id=project_id,
+        user = User(
+            wallet_address="sample_wallet",
             cvat_id=1,
-            cvat_cloudstorage_id=1,
-            status=ProjectStatuses.validation.value,
-            job_type=TaskTypes.image_label_binary.value,
-            escrow_address=escrow_address,
-            chain_id=Networks.localhost.value,
-            bucket_url="https://test.storage.googleapis.com/",
+            cvat_email="email@sample.com",
+        )
+        self.session.add(user)
+
+        project_creation_time = utcnow() - timedelta(days=1)
+        project_completion_time = project_creation_time + timedelta(hours=10)
+
+        cvat_project, cvat_task, cvat_job = create_project_task_and_job(
+            self.session, escrow_address, 1
         )
+        cvat_project.status = ProjectStatuses.validation
+        cvat_project.created_at = project_creation_time
+        cvat_project.updated_at = project_completion_time
         self.session.add(cvat_project)
 
-        webhok_id = str(uuid.uuid4())
+        cvat_task.status = TaskStatuses.completed
+        cvat_task.created_at = project_creation_time
+        cvat_task.updated_at = project_completion_time
+        self.session.add(cvat_task)
+
+        cvat_job.status = JobStatuses.completed
+        cvat_job.created_at = project_creation_time
+        cvat_job.updated_at = project_completion_time
+        self.session.add(cvat_job)
+
+        cvat_assignment1 = Assignment(
+            id=str(uuid.uuid4()),
+            created_at=project_creation_time,
+            updated_at=project_creation_time + timedelta(hours=1),
+            expires_at=project_creation_time + timedelta(hours=1),
+            user_wallet_address=user.wallet_address,
+            cvat_job_id=cvat_job.cvat_id,
+            status=AssignmentStatuses.expired,
+        )
+        self.session.add(cvat_assignment1)
+
+        cvat_assignment2 = Assignment(
+            id=str(uuid.uuid4()),
+            created_at=project_creation_time + timedelta(minutes=5),
+            updated_at=project_completion_time,
+            expires_at=project_creation_time + timedelta(minutes=5) + timedelta(hours=1),
+            user_wallet_address=user.wallet_address,
+            cvat_job_id=cvat_job.cvat_id,
+            status=AssignmentStatuses.completed,
+        )
+        self.session.add(cvat_assignment2)
+
+        webhook_id = str(uuid.uuid4())
         webhook = Webhook(
-            id=webhok_id,
+            id=webhook_id,
             signature="signature",
             escrow_address=escrow_address,
             chain_id=chain_id,
@@ -72,15 +110,25 @@ def test_process_incoming_recording_oracle_webhooks_job_completed_type(self):
 
         process_incoming_recording_oracle_webhooks()
 
-        updated_webhook = (
-            self.session.execute(select(Webhook).where(Webhook.id == webhok_id)).scalars().first()
-        )
-
+        updated_webhook = self.session.query(Webhook).get(webhook_id)
         assert updated_webhook.status == OracleWebhookStatuses.completed.value
         assert updated_webhook.attempts == 1
-        db_project = self.session.query(Project).filter_by(id=project_id).first()
 
+        db_project = self.session.query(Project).get(cvat_project.id)
         assert db_project.status == ProjectStatuses.recorded.value
+        assert db_project.updated_at > project_completion_time
+
+        db_task = self.session.query(Task).get(cvat_task.id)
+        assert db_task.updated_at > project_completion_time
+
+        db_job = self.session.query(Job).get(cvat_job.id)
+        assert db_job.updated_at > project_completion_time
+
+        db_assignment1 = self.session.query(Assignment).get(cvat_assignment1.id)
+        assert db_assignment1.updated_at < project_completion_time
+
+        db_assignment2 = self.session.query(Assignment).get(cvat_assignment2.id)
+        assert db_assignment2.updated_at > project_completion_time
 
     def test_process_incoming_recording_oracle_webhooks_job_completed_type_invalid_project_status(
         self,
