feat: Generate JobId as UUIDv7

backend/mypy.ini

@@ -0,0 +1,2 @@
+[mypy]
+mypy_path = stubs

backend/oqtopus_cloud/user/routers/jobs.py

@@ -1,5 +1,4 @@
 import json
-import uuid
 from datetime import datetime
 from typing import Any
 
@@ -12,6 +11,7 @@
 from sqlalchemy.orm import (
     Session,
 )
+from uuid_extensions import uuid7
 from zoneinfo import ZoneInfo
 
 from oqtopus_cloud.common.models.device import Device
@@ -120,8 +120,7 @@ def submit_jobs(
         description = validate_description(request)
 
         job = Job(
-            # TODO: UUIDv7
-            id=str(uuid.uuid4()),
+            id=uuid7(as_type="str"),
             owner=owner,
             name=name,
             description=description,

backend/stubs/uuid_extensions/__init__.pyi

@@ -0,0 +1,3 @@
+__version__ = "0.1.0"
+
+from .uuid7 import *

backend/stubs/uuid_extensions/uuid7.pyi

@@ -0,0 +1,30 @@
+__all__ = (
+    "uuid7",
+    "uuid7str",
+    "time_ns",
+    "check_timing_precision",
+    "uuid_to_datetime",
+)
+
+import datetime
+import time
+import uuid
+from typing import Callable, Optional, Union
+
+time_ns = time.time_ns
+
+def uuid7(
+    ns: Optional[int] = None,
+    as_type: Optional[str] = None,
+    time_func: Callable[[], int] = time_ns,
+    _last=[0, 0, 0, 0],
+    _last_as_of=[0, 0, 0, 0],
+) -> Union[uuid.UUID, str, int, bytes]: ...
+def uuid7str(ns: Optional[int] = None) -> str: ...
+def check_timing_precision(
+    timing_func: Optional[Callable[[], int]] = None,
+) -> str: ...
+def uuid_to_datetime(
+    s: Union[str, uuid.UUID, int],
+    suppress_error=True,
+) -> Optional[datetime.datetime]: ...

backend/tests/oqtopus_cloud/user/routers/test_jobs.py

@@ -69,6 +69,33 @@ def test_get_job_404(
     assert response.json() == {"detail": "job not found with the given id"}
 
 
+def test_job_sortedness(test_db):
+    def mk_job(n: int) -> SubmitJobRequest:
+        return SubmitJobRequest(
+            name=f"test-job-{n}",
+            device_id="Kawasaki",
+            status=JobStatus.submitted,
+            job_info=JobInfoSampling(job_type="sampling", code="code"),
+            simulator_info="{}",
+            transpiler_info="{}",
+            mitigation_info="{}",
+            shots=1000,
+        )
+
+    def is_sorted(xs: list[str]) -> bool:
+        return xs == sorted(xs)
+
+    test_db.flush()
+    job_ids: list[str] = []
+    for n in range(1, 10):
+        submit_resp = client.post("/jobs", content=mk_job(n).model_dump_json())
+        print(f"submit_resp={submit_resp.json()}")
+        job_id = SubmitJobResponse.model_validate(submit_resp.json()).job_id
+        job_ids.append(job_id)
+
+    assert is_sorted(job_ids)
+
+
 def test_get_jobs_handler(
     test_db,
 ):

pyproject.toml

@@ -22,6 +22,8 @@ fastapi = "^0.110.0"
 aws-xray-sdk = "^2.13.0"
 setuptools = "^75.1.0"
 types-setuptools = "^75.1.0.20240917"
+uuid7 = "^0.1.0"
+uuid-v7 = "^1.0.0"
 
 
 [tool.poetry.group.dev.dependencies]