refactor(jobs): use new input/output schemas (#126)

Inputs are now objects instead of just lists

Author: nfrasser

cryosparc/api.pyi

@@ -23,7 +23,7 @@ from .models.exposure import Exposure
 from .models.external import ExternalOutputSpec
 from .models.job import Job, JobStatus
 from .models.job_register import JobRegister
-from .models.job_spec import Category, InputSpec, InputSpecs, OutputResult, OutputSpec, OutputSpecs
+from .models.job_spec import Category, InputSpec, OutputResult, OutputSpec
 from .models.license import LicenseInstance, UpdateTag
 from .models.notification import Notification
 from .models.project import GenerateIntermediateResultsSettings, Project, ProjectSymlink
@@ -602,27 +602,13 @@ class JobsNamespace(APINamespace):
         """
         ...
     def get_log_path(self, project_uid: str, job_uid: str, /) -> str: ...
-    def get_input_specs(self, project_uid: str, job_uid: str, /) -> InputSpecs: ...
-    def get_input_spec(self, project_uid: str, job_uid: str, input_name: str, /) -> InputSpec: ...
-    def add_external_input(self, project_uid: str, job_uid: str, input_name: str, /, body: InputSpec) -> Job:
-        """
-        Add or replace an external job's input.
-        """
-        ...
-    def get_output_specs(self, project_uid: str, job_uid: str, /) -> OutputSpecs: ...
     def get_output_fields(
         self, project_uid: str, job_uid: str, output_name: str, /, dtype_params: dict = {}
     ) -> List[Tuple[str, str]]:
         """
         Expected dataset column definitions for given job output, excluding passthroughs.
         """
         ...
-    def get_output_spec(self, project_uid: str, job_uid: str, output_name: str, /) -> OutputSpec: ...
-    def add_external_output(self, project_uid: str, job_uid: str, output_name: str, /, body: OutputSpec) -> Job:
-        """
-        Add or replace an external job's output.
-        """
-        ...
     def create(
         self,
         project_uid: str,
@@ -758,6 +744,16 @@ class JobsNamespace(APINamespace):
         Removes an output result connected within the given input connection.
         """
         ...
+    def add_external_input(self, project_uid: str, job_uid: str, input_name: str, /, body: InputSpec) -> Job:
+        """
+        Add or replace an external job's input.
+        """
+        ...
+    def add_external_output(self, project_uid: str, job_uid: str, output_name: str, /, body: OutputSpec) -> Job:
+        """
+        Add or replace an external job's output.
+        """
+        ...
     def enqueue(
         self,
         project_uid: str,

cryosparc/controllers/job.py

@@ -1020,19 +1020,18 @@ def print_input_spec(self):
                         |             |          |           | alignments2D    | alignments2D    | ✕
                         |             |          |           | alignments3D    | alignments3D    | ✕
         """
-        specs = self.cs.api.jobs.get_input_specs(self.project_uid, self.uid)
         headings = ["Input", "Title", "Type", "Required?", "Input Slots", "Slot Types", "Slot Required?"]
         rows = []
-        for key, spec in specs.root.items():
-            name, title, type = key, spec.title, spec.type
-            required = f"✓ ({spec.count_min}" if spec.count_min else "✕ (0"
-            if spec.count_max in (0, "inf"):
+        for key, input in self.model.spec.inputs.root.items():
+            name, title, type = key, input.title, input.type
+            required = f"✓ ({input.count_min}" if input.count_min else "✕ (0"
+            if input.count_max in (0, "inf"):
                 required += "+)"  # unlimited connections
-            elif spec.count_min == spec.count_max:
+            elif input.count_min == input.count_max:
                 required += ")"
             else:
-                required += f"-{spec.count_max})"
-            for slot in spec.slots:
+                required += f"-{input.count_max})"
+            for slot in input.slots:
                 slot = as_input_slot(slot)
                 rows.append([name, title, type, required, slot.name, slot.dtype, "✓" if slot.required else "✕"])
                 name, title, type, required = ("",) * 4  # only show group info on first iter
@@ -1061,15 +1060,10 @@ def print_output_spec(self):
             particles              | Particles   | particle | blob                   | blob            | ✕
                                    |             |          | ctf                    | ctf             | ✕
         """
-        specs = self.cs.api.jobs.get_output_specs(self.project_uid, self.uid)
         headings = ["Output", "Title", "Type", "Result Slots", "Result Types", "Passthrough?"]
         rows = []
-        for key, spec in specs.root.items():
-            output = self.model.spec.outputs.root.get(key)
-            if not output:
-                warnings.warn(f"No results for input {key}", stacklevel=2)
-                continue
-            name, title, type = key, spec.title, spec.type
+        for key, output in self.model.spec.outputs.root.items():
+            name, title, type = key, output.title, output.type
             for result in output.results:
                 rows.append([name, title, type, result.name, result.dtype, "✓" if result.passthrough else "✕"])
                 name, title, type = "", "", ""  # only these print once per group

cryosparc/models/job_spec.py

@@ -59,22 +59,6 @@ class Connection(BaseModel):
 
     job_uid: str
     output: str
-    type: Literal[
-        "exposure",
-        "particle",
-        "template",
-        "volume",
-        "volume_multi",
-        "mask",
-        "live",
-        "ml_model",
-        "symmetry_candidate",
-        "flex_mesh",
-        "flex_model",
-        "hyperparameter",
-        "denoise_model",
-        "annotation_model",
-    ]
     results: List[InputResult] = []
 
 
@@ -135,6 +119,32 @@ class InputSlot(BaseModel):
     required: bool = False
 
 
+class Input(BaseModel):
+    type: Literal[
+        "exposure",
+        "particle",
+        "template",
+        "volume",
+        "volume_multi",
+        "mask",
+        "live",
+        "ml_model",
+        "symmetry_candidate",
+        "flex_mesh",
+        "flex_model",
+        "hyperparameter",
+        "denoise_model",
+        "annotation_model",
+    ]
+    title: str
+    description: str = ""
+    slots: List[InputSlot] = []
+    count_min: int = 0
+    count_max: Union[int, str] = "inf"
+    repeat_allowed: bool = False
+    connections: List[Connection] = []
+
+
 class InputSpec(BaseModel):
     type: Literal[
         "exposure",
@@ -165,7 +175,7 @@ class InputSpecs(RootModel):
 
 
 class Inputs(RootModel):
-    root: Dict[str, List[Connection]] = {}
+    root: Dict[str, Input] = {}
 
 
 class Params(BaseModel):
@@ -211,6 +221,11 @@ class Output(BaseModel):
         "denoise_model",
         "annotation_model",
     ]
+    title: str
+    description: str = ""
+    slots: List[OutputSlot] = []
+    passthrough: Optional[str] = None
+    passthrough_exclude_slots: List[str] = []
     results: List[OutputResult] = []
     num_items: int = 0
     image: Optional[str] = None

cryosparc/models/user.py

@@ -13,7 +13,7 @@ class Bookmark(BaseModel):
     description: str
     color: str
     last_accessed: datetime.datetime
-    created_at: datetime.datetime = datetime.datetime(2025, 3, 6, 16, 44, 32, 992643, tzinfo=datetime.timezone.utc)
+    created_at: datetime.datetime
 
 
 class Email(BaseModel):

tests/conftest.py

@@ -16,6 +16,7 @@
 from cryosparc.models.job import Job
 from cryosparc.models.job_spec import (
     Connection,
+    Input,
     InputResult,
     Inputs,
     JobSpec,
@@ -292,18 +293,26 @@ def mock_new_job(mock_user, mock_project):
         spec=JobSpec(
             type="homo_abinit",
             params=Params(),
-            inputs=Inputs({"particles": []}),
+            inputs=Inputs({"particles": Input(type="particle", title="Particles")}),
             outputs=Outputs(
                 {
                     "particles_class_0": Output(
                         type="particle",
+                        title="Particles Class 1",
+                        passthrough="particles",
                         results=[
                             OutputResult(name="blob", dtype="blob"),
                             OutputResult(name="ctf", dtype="ctf"),
                             OutputResult(name="alignments3D", dtype="alignments3D"),
                         ],
                     ),
-                    "volume_class_0": Output(type="volume", results=[OutputResult(name="map", dtype="blob")]),
+                    "volume_class_0": Output(
+                        type="volume",
+                        title="Volume Class 0",
+                        results=[
+                            OutputResult(name="map", dtype="blob"),
+                        ],
+                    ),
                 }
             ),
             ui_tile_width=1,
@@ -330,7 +339,6 @@ def mock_new_job_with_params(mock_new_job: Job, mock_params: Params):
 def mock_new_job_with_connection(mock_new_job_with_params: Job):
     job = mock_new_job_with_params.model_copy(deep=True)
     input_particles = Connection(
-        type="particle",
         job_uid="J41",
         output="particles",
         results=[
@@ -341,7 +349,7 @@ def mock_new_job_with_connection(mock_new_job_with_params: Job):
         ],
     )
     passthrough_result = OutputResult(name="location", dtype="location", passthrough=True)
-    job.spec.inputs.root["particles"] = [input_particles]
+    job.spec.inputs.root["particles"] = Input(type="particle", title="Particles", connections=[input_particles])
     job.spec.outputs.root["particles_class_0"].results.append(passthrough_result)
     return job
 
@@ -352,6 +360,8 @@ def mock_job(mock_new_job_with_connection: Job):  # completed
     # fmt: off
     output_particles_class_0 = Output(
         type="particle",
+        title="Particles Class 0",
+        passthrough="particles",
         results=[
             OutputResult(
                 name="blob",
@@ -387,6 +397,7 @@ def mock_job(mock_new_job_with_connection: Job):  # completed
     )
     output_volume_class_0 = Output(
         type="volume",
+        title="Volume Class 0",
         results=[
             OutputResult(
                 name="map",

tests/controllers/test_job.py

@@ -134,6 +134,7 @@ def external_job_with_added_output(external_job: ExternalJobController, mock_ext
     mock_external_job = mock_external_job.model_copy(deep=True)
     mock_external_job.spec.outputs.root["particles"] = Output(
         type="particle",
+        title="Particles",
         results=[
             OutputResult(name="blob", dtype="blob"),
             OutputResult(name="ctf", dtype="ctf"),
@@ -150,6 +151,7 @@ def mock_external_job_with_saved_output(external_job_with_added_output: External
     mock_external_job = mock_external_job.model_copy(deep=True)
     mock_external_job.spec.outputs.root["particles"] = Output(
         type="particle",
+        title="Particles",
         results=[
             OutputResult(name="blob", dtype="blob", versions=[0], metafiles=[metafile], num_items=[10]),
             OutputResult(name="ctf", dtype="ctf", versions=[0], metafiles=[metafile], num_items=[10]),

tests/test_tools.py

@@ -41,7 +41,7 @@ def test_create_job_connect_params(
     assert isinstance(job, JobController)
     assert job.uid == mock_new_job_with_connection.uid
     assert job.model.spec.params == mock_params
-    assert len(job.model.spec.inputs.root["particles"]) == 1
+    assert len(job.model.spec.inputs.root["particles"].connections) == 1
     mock_create_endpoint.assert_called_once_with(
         project.uid, "W1", type="homo_abinit", title="", description="", params=mock_params.model_dump()
     )