From 400e8ecee242a5ec5f6fbf7a4f1e8b7cfbaa4afa Mon Sep 17 00:00:00 2001 From: Andrew Lamb Date: Wed, 29 Jul 2026 11:07:47 -0700 Subject: [PATCH 1/7] add ability to start compute tasks --- .../extensions/curator/compute_tasks.md | 353 ++++++ docs/reference/experimental/async/curator.md | 56 + docs/reference/experimental/sync/curator.md | 56 + mkdocs.yml | 1 + .../core/constants/concrete_types.py | 19 + synapseclient/models/__init__.py | 16 + synapseclient/models/curation.py | 1030 +++++++++++++++- synapseclient/models/mixins/CLAUDE.md | 4 +- .../models/mixins/asynchronous_job.py | 92 +- .../models/async/test_curation_async.py | 318 ++++- .../async/unit_test_asynchronous_job.py | 178 ++- .../models/async/unit_test_curation_async.py | 1095 ++++++++++++++++- 12 files changed, 3119 insertions(+), 99 deletions(-) create mode 100644 docs/guides/extensions/curator/compute_tasks.md diff --git a/docs/guides/extensions/curator/compute_tasks.md b/docs/guides/extensions/curator/compute_tasks.md new file mode 100644 index 000000000..49072d151 --- /dev/null +++ b/docs/guides/extensions/curator/compute_tasks.md @@ -0,0 +1,353 @@ +# How to Run Compute Tasks + +This guide is for **curation administrators** who want Synapse to produce metadata automatically instead of asking a contributor to type it into a Grid. A *compute task* is a `CurationTask` whose properties describe a computation, and which you run with [CurationTask.execute][synapseclient.models.CurationTask.execute]. + +If you're setting up a task for a person to fill in by hand, see [How to Set Up Metadata Curation Workflows](metadata_curation.md) instead. The two kinds of task live side by side in the same project — a compute task usually *writes into* the RecordSet of a record-based task created by that guide. + +## What you'll accomplish + +By following this guide, you will: + +- Choose between the two kinds of compute task Synapse supports +- Create a compute task and point it at a destination RecordSet +- Run the computation and wait for it to finish +- Read the execution details, including the error message when a run fails + +## Prerequisites + +- A Synapse account, and either assignment to the task or `UPDATE` access on the task's project +- Python environment with synapseclient and the `curator` extension installed (`pip install --upgrade "synapseclient[curator]"`) +- A **destination task**: an existing record-based `CurationTask` whose RecordSet will receive the output. Create one with [create_record_based_metadata_task][synapseclient.extensions.curator.create_record_based_metadata_task] as described in [How to Set Up Metadata Curation Workflows](metadata_curation.md). The JSON Schema bound to that RecordSet defines the shape of the output +- Depending on the kind of compute task, either a file-based `CurationTask` to read annotations from, or a folder of source files to transform + +## Step 1: Authenticate and import + +```python +from synapseclient import Synapse +from synapseclient.models import ( + CurationTask, + RecordSetGenerationExecutionProperties, + SampleSheetGenerationExecutionProperties, +) + +syn = Synapse() +syn.login() +``` + +## Step 2: Choose the kind of compute task + +### Option A: Sample sheet generation + +Use this when the metadata you need already exists as annotations on files, and you want it reshaped into a sample sheet. Synapse reads the FileView of an existing **file-based** curation task and writes a sample sheet into the destination RecordSet. + +```python +task = CurationTask( + project_id="syn9876543", + data_type="animal_sample_sheet", # Must be unique within the project + instructions="Generate the sample sheet from the annotated sequencing files.", + task_properties=SampleSheetGenerationExecutionProperties( + input_task_id=123, # A file-based CurationTask; its FileView is the source + destination_task_id=456, # A record-based CurationTask; its RecordSet is the target + ), +).store() + +print(f"Created compute task: {task.task_id}") +``` + +`input_task_id` must refer to a task with `FileBasedMetadataTaskProperties` — the annotations shown in its FileView are the source data. The JSON Schema bound to the destination task's RecordSet establishes the target sample sheet format. + +### Option B: Record set generation + +Use this when the metadata is locked up in documents rather than annotations. Synapse reads the files in a folder, follows your free-text instructions to transform them into a CSV, and writes that CSV to the destination RecordSet. + +```python +task = CurationTask( + project_id="syn9876543", + data_type="clinical_records_from_pdfs", # Must be unique within the project + instructions="Extract the clinical measurements from the uploaded study reports.", + task_properties=RecordSetGenerationExecutionProperties( + folder_id="syn987654321", # Folder holding the source files + instructions=( + "Each PDF is one subject visit. Produce one row per visit with the " + "subject identifier, visit date, and every measurement recorded in the " + "summary table." + ), + destination_task_id=456, # A record-based CurationTask; its RecordSet is the target + ), +).store() + +print(f"Created compute task: {task.task_id}") +``` + +!!! note "Limits on the source folder" + The transformation reads the **direct children** of `folder_id` only — files in subfolders are ignored. There is a maximum of 20 files, each under 100 MB, and each must be a PDF, CSV, TXT, or JSON file. + +The `instructions` on the properties describe the transformation to perform and are what the computation acts on. The `instructions` on the task itself are the human-facing description shown in the Synapse web UI, the same as for any other curation task. + +## Step 3: Mark the task as executable + +A newly created task has no execution details, and Synapse refuses to dispatch a task that has none: + +``` +400 Client Error: Task does not have ExecutableTaskExecutionDetails. Cannot dispatch for execution. +``` + +Attach the execution details that match the task's properties with [CurationTask.set_execution_details][synapseclient.models.CurationTask.set_execution_details] — `SampleSheetGenerationExecutionDetails` for Option A, `RecordSetGenerationExecutionDetails` for Option B. Create them empty; Synapse fills in `async_job_id`, `started_by`, and `started_on` as the job runs, and the error fields if it fails. + +```python +from synapseclient.models import RecordSetGenerationExecutionDetails + +task.set_execution_details( + execution_details=RecordSetGenerationExecutionDetails() +) +``` + +This is a one-time step per task. It does not change the task's state, which must still be `NOT_STARTED` when you run it. + +## Step 4: Run the computation + +[CurationTask.execute][synapseclient.models.CurationTask.execute] starts the job and blocks until it finishes, then returns the task's execution details. + +```python +details = task.execute() + +print(f"Started by: {details.started_by}") +print(f"Started on: {details.started_on}") +``` + +The task must be in the `NOT_STARTED` state. A task that is already running, or one whose properties describe manual metadata entry, cannot be executed. + +Generation can take a while. `execute` waits up to `timeout` seconds for the job to complete *or report progress*, so a long-running job that keeps reporting progress will not trip the timeout. Raise it for large inputs: + +```python +details = task.execute(timeout=600) +``` + +### Running asynchronously + +Every method has an `_async` counterpart, which is the better choice when you're running several computations at once: + +```python +import asyncio +from synapseclient import Synapse +from synapseclient.models import CurationTask + +syn = Synapse() +syn.login() + +async def main(): + results = await asyncio.gather( + CurationTask(task_id=111).execute_async(timeout=600), + CurationTask(task_id=222).execute_async(timeout=600), + ) + for details in results: + print(details.started_on) + +asyncio.run(main()) +``` + +## Step 5: Check the outcome + +The returned execution details record how the run went. The same details are attached to the task's status, so a run that finished earlier — or one whose client was interrupted mid-wait — can be inspected at any time. + +!!! warning "A failed run may not raise" + When the computation itself fails, the error is recorded on the task's execution details and the task stays in `NOT_STARTED`, but the failure is not always reported back through the async job. `execute` can keep polling well past its `timeout`, since the timeout only fires when a job stops reporting progress. If a run appears to hang, interrupt it and read `execution_details.error_message` — the job runs server-side and is unaffected by the client stopping. + +```python +from synapseclient import Synapse +from synapseclient.core.exceptions import SynapseError, SynapseTimeoutError +from synapseclient.models import CurationTask + +syn = Synapse() +syn.login() + +task = CurationTask(task_id=789) + +try: + details = task.execute(timeout=600) + print("Execution finished.") +except SynapseError as error: + print(f"Execution failed: {error}") +except SynapseTimeoutError: + print("Still running; check the task status later.") + +# The status carries the same execution details, so you can check a run at any time +status = task.get_status() +print(f"State: {status.state}") + +if status.execution_details and status.execution_details.error_message: + print(f"Error: {status.execution_details.error_message}") + print(f"Details: {status.execution_details.error_details}") +``` + +A successful run writes its output as a **new version** of the destination task's RecordSet, leaving the previous version intact, and the compute task moves to `IN_REVIEW` — the results are waiting for a human to look at them. + +The output lands in the *destination* task's RecordSet, not in anything owned by the compute task, so read the RecordSet's synId off that task rather than hardcoding it: + +```python +from synapseclient.models import RecordSet + +destination_task = CurationTask(task_id=456).get() +record_set = RecordSet(id=destination_task.task_properties.record_set_id).get() +print(f"Now at version {record_set.version_number}") +``` + +Generating a version does not validate it: `validation_file_handle_id` stays empty until a Grid session is exported back to the RecordSet. To check the generated records against the bound schema, open a Grid on the RecordSet and export it, then read the results as described in [Getting detailed validation results](metadata_curation.md#getting-detailed-validation-results). + +Once you're satisfied with the output, close the task out: + +```python +from synapseclient.models import TaskState + +CurationTask(task_id=789).set_task_state(state=TaskState.COMPLETED) +``` + +## Complete example scripts + +### Record set generation + +This script creates a record set generation task, runs it, and reports the outcome: + +```python +from synapseclient import Synapse +from synapseclient.core.exceptions import SynapseError +from synapseclient.models import ( + CurationTask, + RecordSetGenerationExecutionDetails, + RecordSetGenerationExecutionProperties, + TaskState, +) + +PROJECT_ID = "syn9876543" # The project both tasks belong to +SOURCE_FOLDER_ID = "syn987654321" # Folder of PDFs to transform +DESTINATION_TASK_ID = 456 # A record-based CurationTask + +syn = Synapse() +syn.login() + +# Step 1: Create the compute task +task = CurationTask( + project_id=PROJECT_ID, + data_type="clinical_records_from_pdfs", + instructions="Extract the clinical measurements from the uploaded study reports.", + task_properties=RecordSetGenerationExecutionProperties( + folder_id=SOURCE_FOLDER_ID, + instructions=( + "Each PDF is one subject visit. Produce one row per visit with the " + "subject identifier, visit date, and every measurement recorded in the " + "summary table." + ), + destination_task_id=DESTINATION_TASK_ID, + ), +).store() + +print(f"Created compute task {task.task_id}") + +# Step 2: Mark the task as executable. Synapse refuses to dispatch a task that +# has no ExecutableTaskExecutionDetails, and a new task has none. +task.set_execution_details(execution_details=RecordSetGenerationExecutionDetails()) + +# Step 3: Run it, allowing 10 minutes +try: + details = task.execute(timeout=600) + print(f"Execution started on {details.started_on} by {details.started_by}") +except SynapseError as error: + print(f"Execution failed: {error}") + +# Step 4: Report the outcome +status = task.get_status() +print(f"Task state: {status.state}") + +if status.state == TaskState.IN_REVIEW: + print("Results are ready for review in the destination RecordSet.") +elif status.execution_details and status.execution_details.error_message: + print(f"Error: {status.execution_details.error_message}") +``` + +### Sample sheet generation + +The same four steps, reading from a file-based task instead of a folder. Note that there are no transformation instructions to write: the schema bound to the destination RecordSet is what determines the sample sheet format. + +```python +from synapseclient import Synapse +from synapseclient.core.exceptions import SynapseError +from synapseclient.models import ( + CurationTask, + SampleSheetGenerationExecutionDetails, + SampleSheetGenerationExecutionProperties, + TaskState, +) + +PROJECT_ID = "syn9876543" # The project the tasks belong to +INPUT_TASK_ID = 123 # A file-based CurationTask; its FileView is the source +DESTINATION_TASK_ID = 456 # A record-based CurationTask; its RecordSet is the target + +syn = Synapse() +syn.login() + +# Step 1: Create the compute task +task = CurationTask( + project_id=PROJECT_ID, + data_type="patient_sample_sheet", + instructions="Generate the sample sheet from the annotated sequencing files.", + task_properties=SampleSheetGenerationExecutionProperties( + input_task_id=INPUT_TASK_ID, + destination_task_id=DESTINATION_TASK_ID, + ), +).store() + +print(f"Created compute task {task.task_id}") + +# Step 2: Mark the task as executable. Synapse refuses to dispatch a task that +# has no ExecutableTaskExecutionDetails, and a new task has none. +task.set_execution_details(execution_details=SampleSheetGenerationExecutionDetails()) + +# Step 3: Run it, allowing 10 minutes +try: + details = task.execute(timeout=600) + print(f"Execution started on {details.started_on} by {details.started_by}") +except SynapseError as error: + print(f"Execution failed: {error}") + +# Step 4: Report the outcome +status = task.get_status() +print(f"Task state: {status.state}") + +if status.state == TaskState.IN_REVIEW: + print("Results are ready for review in the destination RecordSet.") +elif status.execution_details and status.execution_details.error_message: + print(f"Error: {status.execution_details.error_message}") +``` + +The files behind `INPUT_TASK_ID` must already carry the annotations the sample sheet is built from — the computation reads them through that task's FileView, not from the file contents. + +## Task states for compute tasks + +Compute tasks move through the same lifecycle as any other curation task, using two states that manual tasks rarely see: + +| State | Meaning | +|---|---| +| `NOT_STARTED` | The task has been created but not run. This is the only state `execute` accepts. A run that fails leaves the task here, with the reason on its execution details, so it can simply be run again. | +| `EXECUTING` | An automated execution is currently running. | +| `IN_REVIEW` | The execution completed successfully and the results are pending human review. | +| `COMPLETED` | The task has been completed and verified. | +| `CANCELED` | The task has been canceled and is no longer needed. | + +## References + +### API Documentation + +- [CurationTask.execute][synapseclient.models.CurationTask.execute] - Run a compute task and wait for the result +- [CurationTask.store][synapseclient.models.CurationTask.store] - Create or update a curation task +- [CurationTask.get_status][synapseclient.models.CurationTask.get_status] - Read a task's state and execution details +- [CurationTask.set_task_state][synapseclient.models.CurationTask.set_task_state] - Update the lifecycle state of a curation task +- [SampleSheetGenerationExecutionProperties][SampleSheetGenerationExecutionProperties-reference] - Properties for a sample sheet generation task +- [RecordSetGenerationExecutionProperties][RecordSetGenerationExecutionProperties-reference] - Properties for a record set generation task +- [SampleSheetGenerationExecutionDetails][SampleSheetGenerationExecutionDetails-reference] - Execution details for a sample sheet generation run +- [RecordSetGenerationExecutionDetails][RecordSetGenerationExecutionDetails-reference] - Execution details for a record set generation run + +### Related Documentation + +- [How to Set Up Metadata Curation Workflows](metadata_curation.md) - Create the record-based task that receives the output +- [How to Enter and Update Metadata for a Curation Task](metadata_contribution.md) - The contributor-facing guide for manual tasks +- [JSON Schema Tutorial](../../../tutorials/python/json_schema.md) - Learn how to register the schemas that define the output format diff --git a/docs/reference/experimental/async/curator.md b/docs/reference/experimental/async/curator.md index 39e62f705..5a79bf921 100644 --- a/docs/reference/experimental/async/curator.md +++ b/docs/reference/experimental/async/curator.md @@ -16,6 +16,8 @@ - list_async - create_grid_session_async - set_task_state_async + - set_execution_details_async + - execute_async --- [](){ #RecordSet-reference-async } @@ -38,6 +40,12 @@ - validate_schema_async - get_schema_derived_keys_async --- +[](){ #CurationTaskProperties-reference-async } +::: synapseclient.models.CurationTaskProperties + options: + inherited_members: true + members: +--- [](){ #RecordBasedMetadataTaskProperties-reference-async } ::: synapseclient.models.RecordBasedMetadataTaskProperties options: @@ -50,6 +58,54 @@ inherited_members: true members: --- +[](){ #SampleSheetGenerationExecutionProperties-reference-async } +::: synapseclient.models.SampleSheetGenerationExecutionProperties + options: + inherited_members: true + members: +--- +[](){ #RecordSetGenerationExecutionProperties-reference-async } +::: synapseclient.models.RecordSetGenerationExecutionProperties + options: + inherited_members: true + members: +--- +[](){ #UnknownCurationTaskProperties-reference-async } +::: synapseclient.models.UnknownCurationTaskProperties + options: + inherited_members: true + members: +--- +[](){ #TaskExecutionDetails-reference-async } +::: synapseclient.models.TaskExecutionDetails + options: + inherited_members: true + members: +--- +[](){ #ExecutableTaskExecutionDetails-reference-async } +::: synapseclient.models.ExecutableTaskExecutionDetails + options: + inherited_members: true + members: +--- +[](){ #SampleSheetGenerationExecutionDetails-reference-async } +::: synapseclient.models.SampleSheetGenerationExecutionDetails + options: + inherited_members: true + members: +--- +[](){ #RecordSetGenerationExecutionDetails-reference-async } +::: synapseclient.models.RecordSetGenerationExecutionDetails + options: + inherited_members: true + members: +--- +[](){ #UnknownTaskExecutionDetails-reference-async } +::: synapseclient.models.UnknownTaskExecutionDetails + options: + inherited_members: true + members: +--- [](){ #AuthorizationMode-reference-async } ::: synapseclient.models.AuthorizationMode options: diff --git a/docs/reference/experimental/sync/curator.md b/docs/reference/experimental/sync/curator.md index b330fc3ec..3608ae475 100644 --- a/docs/reference/experimental/sync/curator.md +++ b/docs/reference/experimental/sync/curator.md @@ -16,6 +16,8 @@ - list - create_grid_session - set_task_state + - set_execution_details + - execute --- [](){ #RecordSet-reference } @@ -38,6 +40,12 @@ - validate_schema - get_schema_derived_keys --- +[](){ #CurationTaskProperties-reference } +::: synapseclient.models.CurationTaskProperties + options: + inherited_members: true + members: +--- [](){ #RecordBasedMetadataTaskProperties-reference } ::: synapseclient.models.RecordBasedMetadataTaskProperties options: @@ -50,6 +58,54 @@ inherited_members: true members: --- +[](){ #SampleSheetGenerationExecutionProperties-reference } +::: synapseclient.models.SampleSheetGenerationExecutionProperties + options: + inherited_members: true + members: +--- +[](){ #RecordSetGenerationExecutionProperties-reference } +::: synapseclient.models.RecordSetGenerationExecutionProperties + options: + inherited_members: true + members: +--- +[](){ #UnknownCurationTaskProperties-reference } +::: synapseclient.models.UnknownCurationTaskProperties + options: + inherited_members: true + members: +--- +[](){ #TaskExecutionDetails-reference } +::: synapseclient.models.TaskExecutionDetails + options: + inherited_members: true + members: +--- +[](){ #ExecutableTaskExecutionDetails-reference } +::: synapseclient.models.ExecutableTaskExecutionDetails + options: + inherited_members: true + members: +--- +[](){ #SampleSheetGenerationExecutionDetails-reference } +::: synapseclient.models.SampleSheetGenerationExecutionDetails + options: + inherited_members: true + members: +--- +[](){ #RecordSetGenerationExecutionDetails-reference } +::: synapseclient.models.RecordSetGenerationExecutionDetails + options: + inherited_members: true + members: +--- +[](){ #UnknownTaskExecutionDetails-reference } +::: synapseclient.models.UnknownTaskExecutionDetails + options: + inherited_members: true + members: +--- [](){ #AuthorizationMode-reference } ::: synapseclient.models.AuthorizationMode options: diff --git a/mkdocs.yml b/mkdocs.yml index daa3cecf6..f8efe9d13 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -66,6 +66,7 @@ nav: - Curator JSONschemas: guides/extensions/curator/schema_operations.md - Curator (administrators): guides/extensions/curator/metadata_curation.md - Curator (contributors): guides/extensions/curator/metadata_contribution.md + - Curator compute tasks: guides/extensions/curator/compute_tasks.md # - Expermental Features: # - Validating Annotations: guides/validate_annotations.md - API Reference: diff --git a/synapseclient/core/constants/concrete_types.py b/synapseclient/core/constants/concrete_types.py index d8cbdbd59..3f492d869 100644 --- a/synapseclient/core/constants/concrete_types.py +++ b/synapseclient/core/constants/concrete_types.py @@ -132,9 +132,28 @@ RECORD_BASED_METADATA_TASK_PROPERTIES = ( "org.sagebionetworks.repo.model.curation.metadata.RecordBasedMetadataTaskProperties" ) +SAMPLE_SHEET_GENERATION_EXECUTION_PROPERTIES = ( + "org.sagebionetworks.repo.model.curation.execution." + "SampleSheetGenerationExecutionProperties" +) +RECORD_SET_GENERATION_EXECUTION_PROPERTIES = ( + "org.sagebionetworks.repo.model.curation.execution." + "RecordSetGenerationExecutionProperties" +) GRID_EXECUTION_DETAILS = ( "org.sagebionetworks.repo.model.curation.execution.GridExecutionDetails" ) +SAMPLE_SHEET_GENERATION_EXECUTION_DETAILS = ( + "org.sagebionetworks.repo.model.curation.execution." + "SampleSheetGenerationExecutionDetails" +) +RECORD_SET_GENERATION_EXECUTION_DETAILS = ( + "org.sagebionetworks.repo.model.curation.execution." + "RecordSetGenerationExecutionDetails" +) +COMPUTE_TASK_EXECUTION_REQUEST = ( + "org.sagebionetworks.repo.model.curation.ComputeTaskExecutionRequest" +) # Download List DOWNLOAD_LIST_MANIFEST_REQUEST = ( diff --git a/synapseclient/models/__init__.py b/synapseclient/models/__init__.py index ce1a3bf9e..90d5724e6 100644 --- a/synapseclient/models/__init__.py +++ b/synapseclient/models/__init__.py @@ -10,13 +10,21 @@ from synapseclient.models.curation import ( AuthorizationMode, CurationTask, + CurationTaskProperties, CurationTaskStatus, + ExecutableTaskExecutionDetails, FileBasedMetadataTaskProperties, Grid, GridExecutionDetails, RecordBasedMetadataTaskProperties, + RecordSetGenerationExecutionDetails, + RecordSetGenerationExecutionProperties, + SampleSheetGenerationExecutionDetails, + SampleSheetGenerationExecutionProperties, TaskExecutionDetails, TaskState, + UnknownCurationTaskProperties, + UnknownTaskExecutionDetails, ) from synapseclient.models.dataset import Dataset, DatasetCollection, EntityRef from synapseclient.models.docker import DockerRepository @@ -101,12 +109,20 @@ "AuthorizationMode", "CurationTask", "CurationTaskStatus", + "CurationTaskProperties", "FileBasedMetadataTaskProperties", "RecordBasedMetadataTaskProperties", + "SampleSheetGenerationExecutionProperties", + "RecordSetGenerationExecutionProperties", + "UnknownCurationTaskProperties", "TaskState", "Grid", "GridExecutionDetails", "TaskExecutionDetails", + "ExecutableTaskExecutionDetails", + "SampleSheetGenerationExecutionDetails", + "RecordSetGenerationExecutionDetails", + "UnknownTaskExecutionDetails", "UserProfile", "UserPreference", "UserGroupHeader", diff --git a/synapseclient/models/curation.py b/synapseclient/models/curation.py index bcb507abb..58acc40c4 100644 --- a/synapseclient/models/curation.py +++ b/synapseclient/models/curation.py @@ -46,6 +46,7 @@ wrap_async_generator_to_sync_generator, ) from synapseclient.core.constants.concrete_types import ( + COMPUTE_TASK_EXECUTION_REQUEST, CREATE_GRID_REQUEST, DOWNLOAD_FROM_GRID_REQUEST, FILE_BASED_METADATA_TASK_PROPERTIES, @@ -55,10 +56,15 @@ LIST_GRID_SESSIONS_REQUEST, LIST_GRID_SESSIONS_RESPONSE, RECORD_BASED_METADATA_TASK_PROPERTIES, + RECORD_SET_GENERATION_EXECUTION_DETAILS, + RECORD_SET_GENERATION_EXECUTION_PROPERTIES, + SAMPLE_SHEET_GENERATION_EXECUTION_DETAILS, + SAMPLE_SHEET_GENERATION_EXECUTION_PROPERTIES, SYNCHRONIZE_GRID_REQUEST, UPLOAD_TO_TABLE_PREVIEW_REQUEST, ) from synapseclient.core.download.download_functions import download_from_url +from synapseclient.core.exceptions import SynapseError from synapseclient.core.upload.upload_functions_async import upload_synapse_s3 from synapseclient.core.utils import ( coerce_enum_list, @@ -119,7 +125,57 @@ class AuthorizationMode(str, Enum): @dataclass -class FileBasedMetadataTaskProperties(EnumCoercionMixin): +class CurationTaskProperties(ABC): + """ + Base class for the properties of a CurationTask, describing what is being curated + and where the curated data lives. + + + + The concrete subclass is determined by the concreteType field in the REST response. + + This class is abstract and cannot be instantiated: it does not define a + concreteType. Construct one of its subclasses instead, matching the kind of work + the task describes: FileBasedMetadataTaskProperties, + RecordBasedMetadataTaskProperties, SampleSheetGenerationExecutionProperties or + RecordSetGenerationExecutionProperties. Use this class for isinstance checks or + type hints when the kind of properties does not matter. + """ + + @property + @abstractmethod + def concrete_type(self) -> str: + """The concreteType of this implementation of CurationTaskProperties.""" + ... + + @abstractmethod + def fill_from_dict( + self, synapse_response: dict[str, Any] + ) -> "CurationTaskProperties": + """ + Converts a response from the REST API into this dataclass. + + Arguments: + synapse_response: The response from the REST API. + + Returns: + The CurationTaskProperties object. + """ + ... + + @abstractmethod + def to_synapse_request(self) -> dict[str, Any]: + """ + Converts this dataclass to a dictionary suitable for a Synapse REST API request. + + Returns: + A dictionary representation of this object for API requests. + """ + ... + + +@dataclass +class FileBasedMetadataTaskProperties(CurationTaskProperties, EnumCoercionMixin): """ A CurationTaskProperties for file-based data, describing where data is uploaded and a view which contains the annotations. @@ -145,13 +201,13 @@ class FileBasedMetadataTaskProperties(EnumCoercionMixin): "suggested_authorization_mode": AuthorizationMode } - upload_folder_id: Optional[str] = None + upload_folder_id: str | None = None """The synId of the folder where data files of this type are to be uploaded""" - file_view_id: Optional[str] = None + file_view_id: str | None = None """The synId of the FileView that shows all data of this type""" - suggested_authorization_mode: Optional[Union[AuthorizationMode, str]] = None + suggested_authorization_mode: AuthorizationMode | str | None = None """Recommends who is allowed to access the curation grid session that a client opens for this task. The value is stored on the task as a suggestion; the client applies it when it creates a new session. @@ -166,13 +222,18 @@ class FileBasedMetadataTaskProperties(EnumCoercionMixin): resets the task's active session, so a new grid session must be opened before curation can continue.""" - collaborator_principal_ids: Optional[list[str]] = None + collaborator_principal_ids: list[str] | None = None """Not actively used at this time. The set of principal IDs that should collaborate on the grid session. Used to set the owner(s) of a linked GridSession when suggested_authorization_mode is SESSION_OWNER""" + @property + def concrete_type(self) -> str: + """The concreteType of a FileBasedMetadataTaskProperties.""" + return FILE_BASED_METADATA_TASK_PROPERTIES + def fill_from_dict( - self, synapse_response: Union[Dict[str, Any], Any] + self, synapse_response: dict[str, Any] ) -> "FileBasedMetadataTaskProperties": """ Converts a response from the REST API into this dataclass. @@ -193,7 +254,7 @@ def fill_from_dict( ) return self - def to_synapse_request(self) -> Dict[str, Any]: + def to_synapse_request(self) -> dict[str, Any]: """ Converts this dataclass to a dictionary suitable for a Synapse REST API request. @@ -201,7 +262,7 @@ def to_synapse_request(self) -> Dict[str, Any]: A dictionary representation of this object for API requests. """ request_dict = { - "concreteType": FILE_BASED_METADATA_TASK_PROPERTIES, + "concreteType": self.concrete_type, "uploadFolderId": self.upload_folder_id, "fileViewId": self.file_view_id, "suggestedAuthorizationMode": ( @@ -216,7 +277,7 @@ def to_synapse_request(self) -> Dict[str, Any]: @dataclass -class RecordBasedMetadataTaskProperties(EnumCoercionMixin): +class RecordBasedMetadataTaskProperties(CurationTaskProperties, EnumCoercionMixin): """ A CurationTaskProperties for record-based metadata. @@ -240,10 +301,10 @@ class RecordBasedMetadataTaskProperties(EnumCoercionMixin): "suggested_authorization_mode": AuthorizationMode } - record_set_id: Optional[str] = None + record_set_id: str | None = None """The synId of the RecordSet that will contain all record-based metadata""" - suggested_authorization_mode: Optional[Union[AuthorizationMode, str]] = None + suggested_authorization_mode: AuthorizationMode | str | None = None """Recommends who is allowed to access the curation grid session that a client opens for this task. The value is stored on the task as a suggestion; the client applies it when it creates a new session. @@ -258,13 +319,18 @@ class RecordBasedMetadataTaskProperties(EnumCoercionMixin): resets the task's active session, so a new grid session must be opened before curation can continue.""" - collaborator_principal_ids: Optional[list[str]] = None + collaborator_principal_ids: list[str] | None = None """Not actively used at this time. The set of principal IDs that should collaborate on the grid session. Used to set the owner(s) of a linked GridSession when suggested_authorization_mode is SESSION_OWNER""" + @property + def concrete_type(self) -> str: + """The concreteType of a RecordBasedMetadataTaskProperties.""" + return RECORD_BASED_METADATA_TASK_PROPERTIES + def fill_from_dict( - self, synapse_response: Union[Dict[str, Any], Any] + self, synapse_response: dict[str, Any] ) -> "RecordBasedMetadataTaskProperties": """ Converts a response from the REST API into this dataclass. @@ -284,7 +350,7 @@ def fill_from_dict( ) return self - def to_synapse_request(self) -> Dict[str, Any]: + def to_synapse_request(self) -> dict[str, Any]: """ Converts this dataclass to a dictionary suitable for a Synapse REST API request. @@ -292,7 +358,7 @@ def to_synapse_request(self) -> Dict[str, Any]: A dictionary representation of this object for API requests. """ request_dict = { - "concreteType": RECORD_BASED_METADATA_TASK_PROPERTIES, + "concreteType": self.concrete_type, "recordSetId": self.record_set_id, "suggestedAuthorizationMode": ( self.suggested_authorization_mode.value @@ -305,29 +371,253 @@ def to_synapse_request(self) -> Dict[str, Any]: return request_dict +@dataclass +class SampleSheetGenerationExecutionProperties(CurationTaskProperties): + """ + A CurationTaskProperties for a task that generates a sample sheet from the + annotations of an existing file-based curation task. + + Represents a [Synapse SampleSheetGenerationExecutionProperties](https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/curation/execution/SampleSheetGenerationExecutionProperties.html). + + Attributes: + input_task_id: The ID of the file-based CurationTask (with + FileBasedMetadataTaskProperties) whose FileView provides the source + annotations. + destination_task_id: The ID of the record-based CurationTask whose RecordSet + will receive the generated sample sheet as a new version. The JSON Schema + bound to that RecordSet establishes the target sample sheet format. + """ + + input_task_id: int | None = None + """The ID of the file-based CurationTask (with FileBasedMetadataTaskProperties) + whose FileView provides the source annotations.""" + + destination_task_id: int | None = None + """The ID of the record-based CurationTask whose RecordSet will receive the + generated sample sheet as a new version. The JSON Schema bound to that RecordSet + establishes the target sample sheet format.""" + + @property + def concrete_type(self) -> str: + """The concreteType of a SampleSheetGenerationExecutionProperties.""" + return SAMPLE_SHEET_GENERATION_EXECUTION_PROPERTIES + + def fill_from_dict( + self, synapse_response: dict[str, Any] + ) -> "SampleSheetGenerationExecutionProperties": + """ + Converts a response from the REST API into this dataclass. + + Arguments: + synapse_response: The response from the REST API. + + Returns: + The SampleSheetGenerationExecutionProperties object. + """ + input_task_id = synapse_response.get("inputTaskId", None) + self.input_task_id = int(input_task_id) if input_task_id is not None else None + destination_task_id = synapse_response.get("destinationTaskId", None) + self.destination_task_id = ( + int(destination_task_id) if destination_task_id is not None else None + ) + return self + + def to_synapse_request(self) -> dict[str, Any]: + """ + Converts this dataclass to a dictionary suitable for a Synapse REST API request. + + Returns: + A dictionary representation of this object for API requests. + """ + request_dict = { + "concreteType": self.concrete_type, + "inputTaskId": self.input_task_id, + "destinationTaskId": self.destination_task_id, + } + delete_none_keys(request_dict) + return request_dict + + +@dataclass +class RecordSetGenerationExecutionProperties(CurationTaskProperties): + """ + A CurationTaskProperties for a task that transforms source files in a folder into + a CSV that is written to the RecordSet of another curation task. + + Represents a [Synapse RecordSetGenerationExecutionProperties](https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/curation/execution/RecordSetGenerationExecutionProperties.html). + + Attributes: + folder_id: The synId of a Folder providing the source FileEntities for the + transformation. Limited to 20 direct-child files, each under 100 MB and in + PDF, CSV, TXT, or JSON format. + instructions: Free-text instructions from the data manager describing how the + input files should be transformed to produce the output CSV. + destination_task_id: The ID of the record-based CurationTask whose RecordSet + will receive the generated CSV output as a new version. + """ + + folder_id: str | None = None + """The synId of a Folder providing the source FileEntities for the transformation. + Limited to 20 direct-child files, each under 100 MB and in PDF, CSV, TXT, or JSON + format.""" + + instructions: str | None = None + """Free-text instructions from the data manager describing how the input files + should be transformed to produce the output CSV.""" + + destination_task_id: int | None = None + """The ID of the record-based CurationTask whose RecordSet will receive the + generated CSV output as a new version.""" + + @property + def concrete_type(self) -> str: + """The concreteType of a RecordSetGenerationExecutionProperties.""" + return RECORD_SET_GENERATION_EXECUTION_PROPERTIES + + def fill_from_dict( + self, synapse_response: dict[str, Any] + ) -> "RecordSetGenerationExecutionProperties": + """ + Converts a response from the REST API into this dataclass. + + Arguments: + synapse_response: The response from the REST API. + + Returns: + The RecordSetGenerationExecutionProperties object. + """ + self.folder_id = synapse_response.get("folderId", None) + self.instructions = synapse_response.get("instructions", None) + destination_task_id = synapse_response.get("destinationTaskId", None) + self.destination_task_id = ( + int(destination_task_id) if destination_task_id is not None else None + ) + return self + + def to_synapse_request(self) -> dict[str, Any]: + """ + Converts this dataclass to a dictionary suitable for a Synapse REST API request. + + Returns: + A dictionary representation of this object for API requests. + """ + request_dict = { + "concreteType": self.concrete_type, + "folderId": self.folder_id, + "instructions": self.instructions, + "destinationTaskId": self.destination_task_id, + } + delete_none_keys(request_dict) + return request_dict + + +@dataclass +class UnknownCurationTaskProperties(CurationTaskProperties): + """ + Curation task properties whose concreteType this version of the client does not + recognize. Produced when Synapse returns a CurationTaskProperties subtype that was + added after this client was released. + + The whole response is kept in raw_properties and the concreteType Synapse sent is + exposed as a read-only property over it, so a task carrying properties this client + cannot model can still be read, listed and stored without losing them. The + properties are read-only because there is nothing useful to write: properties this + client does not model cannot be constructed correctly, only reported back as they + arrived. To edit them, upgrade synapseclient to a version that models this + concreteType. + + A task carrying these properties cannot be used with + [CurationTask.create_grid_session][synapseclient.models.CurationTask.create_grid_session] + or with delete_source on + [CurationTask.delete][synapseclient.models.CurationTask.delete], since this client + cannot tell where the curated data lives. + + Attributes: + raw_properties: The unmodified taskProperties from the response. + """ + + raw_properties: dict[str, Any] = field(default_factory=dict) + """The unmodified taskProperties from the response.""" + + @property + def concrete_type(self) -> str: + """The concreteType that Synapse reported for these properties.""" + return self.raw_properties.get("concreteType", "") + + def fill_from_dict( + self, synapse_response: dict[str, Any] + ) -> "UnknownCurationTaskProperties": + """ + Converts a response from the REST API into this dataclass. + + Arguments: + synapse_response: The response from the REST API. + + Returns: + The UnknownCurationTaskProperties object. + """ + self.raw_properties = deepcopy(synapse_response) + return self + + def to_synapse_request(self) -> dict[str, Any]: + """ + Converts this dataclass to a dictionary suitable for a Synapse REST API request. + + Returns the properties exactly as Synapse sent them, so that storing a task + this client cannot fully model does not drop the fields it does not know about. + + Returns: + A dictionary representation of this object for API requests. + + Raises: + ValueError: If these properties carry no concreteType for the server to + dispatch on, either because they were never populated from a Synapse + response or because the response did not name one. + """ + if not self.concrete_type: + raise ValueError( + "UnknownCurationTaskProperties can only be serialized after being " + "populated from a Synapse response carrying a concreteType. To " + "attach properties to a task, construct the type matching the work " + "it describes, such as FileBasedMetadataTaskProperties or " + "RecordBasedMetadataTaskProperties." + ) + return deepcopy(self.raw_properties) + + +TASK_PROPERTIES_DICT: dict[str, type[CurationTaskProperties]] = { + FILE_BASED_METADATA_TASK_PROPERTIES: FileBasedMetadataTaskProperties, + RECORD_BASED_METADATA_TASK_PROPERTIES: RecordBasedMetadataTaskProperties, + SAMPLE_SHEET_GENERATION_EXECUTION_PROPERTIES: SampleSheetGenerationExecutionProperties, + RECORD_SET_GENERATION_EXECUTION_PROPERTIES: RecordSetGenerationExecutionProperties, +} + + def _create_task_properties_from_dict( - properties_dict: Dict[str, Any], -) -> Union[FileBasedMetadataTaskProperties, RecordBasedMetadataTaskProperties]: + properties_dict: dict[str, Any], +) -> CurationTaskProperties: """ - Factory method to create the appropriate FileBasedMetadataTaskProperties/RecordBasedMetadataTaskProperties + Factory method to create the appropriate CurationTaskProperties implementation based on the concreteType. + An unrecognized concreteType is not an error. Refusing to parse it would make every + other field of the task unreadable, including for callers that only want to list + tasks or read their state. The properties are returned as + UnknownCurationTaskProperties, which reports the concreteType Synapse sent and + round-trips the response unchanged. + Arguments: properties_dict: Dictionary containing task properties data Returns: - The appropriate FileBasedMetadataTaskProperties/RecordBasedMetadataTaskProperties instance + The appropriate CurationTaskProperties instance """ concrete_type = properties_dict.get("concreteType", "") - if concrete_type == FILE_BASED_METADATA_TASK_PROPERTIES: - return FileBasedMetadataTaskProperties().fill_from_dict(properties_dict) - elif concrete_type == RECORD_BASED_METADATA_TASK_PROPERTIES: - return RecordBasedMetadataTaskProperties().fill_from_dict(properties_dict) - else: - raise ValueError( - f"Unknown concreteType for CurationTaskProperties: {concrete_type}" - ) + properties_class = TASK_PROPERTIES_DICT.get( + concrete_type, UnknownCurationTaskProperties + ) + return properties_class().fill_from_dict(properties_dict) @dataclass @@ -340,6 +630,12 @@ class TaskExecutionDetails(ABC): The concrete subclass is determined by the concreteType field in the REST response. """ + @property + @abstractmethod + def concrete_type(self) -> str: + """The concreteType of this implementation of TaskExecutionDetails.""" + ... + @abstractmethod def fill_from_dict( self, synapse_response: dict[str, Any] @@ -371,18 +667,246 @@ class GridExecutionDetails(TaskExecutionDetails): """ Execution details for a metadata curation task involving a collaborative grid session. - + + + Attributes: + active_session_id: The unique identifier of the active CRDT grid session linked to this task. + """ + + active_session_id: str | None = None + """The unique identifier of the active CRDT grid session linked to this task.""" + + @property + def concrete_type(self) -> str: + """The concreteType of a GridExecutionDetails.""" + return GRID_EXECUTION_DETAILS + + def fill_from_dict( + self, synapse_response: dict[str, Any] + ) -> "GridExecutionDetails": + """ + Converts a response from the REST API into this dataclass. + + Arguments: + synapse_response: The response from the REST API. + + Returns: + The GridExecutionDetails object. + """ + self.active_session_id = synapse_response.get("activeSessionId") + return self + + def to_synapse_request(self) -> dict[str, Any]: + """ + Converts this dataclass to a dictionary suitable for a Synapse REST API request. + + Returns: + A dictionary representation of this object for API requests. + """ + request_dict: dict[str, Any] = {"concreteType": self.concrete_type} + if self.active_session_id is not None: + request_dict["activeSessionId"] = self.active_session_id + return request_dict + + +@dataclass +class ExecutableTaskExecutionDetails(TaskExecutionDetails): + """ + Base class for the execution details of a CurationTask that supports automated + execution by a sub-worker. The concrete type determines which sub-worker handles + the execution. + + + + A CurationTask must have execution details of this kind for + [CurationTask.execute][synapseclient.models.CurationTask.execute] to run. + + This class is abstract and cannot be instantiated: it does not define a + concreteType. Construct one of its subclasses instead, matching the kind of + computation the task's properties describe: + SampleSheetGenerationExecutionDetails or RecordSetGenerationExecutionDetails. + Use this class for isinstance checks when you need to know whether a task's + execution details support automated execution, regardless of their kind. Details + whose concreteType this client does not recognize are deliberately excluded from + this class, since an unrecognized type may not be executable. + + Attributes: + async_job_id: The ID of the async job currently executing this task. Set when + execution starts; cleared on completion or failure. + started_by: The principal ID of the user who started the execution. + started_on: When the execution was started. + error_message: If execution failed, the error description. + error_details: If execution failed, additional error details. + """ + + async_job_id: str | None = None + """The ID of the async job currently executing this task. Set when execution + starts; cleared on completion or failure.""" + + started_by: str | None = None + """The principal ID of the user who started the execution.""" + + started_on: str | None = None + """When the execution was started.""" + + error_message: str | None = None + """If execution failed, the error description.""" + + error_details: str | None = None + """If execution failed, additional error details.""" + + def fill_from_dict( + self, synapse_response: dict[str, Any] + ) -> "ExecutableTaskExecutionDetails": + """ + Converts a response from the REST API into this dataclass. + + Arguments: + synapse_response: The response from the REST API. + + Returns: + The ExecutableTaskExecutionDetails object. + """ + self.async_job_id = synapse_response.get("asyncJobId", None) + self.started_by = synapse_response.get("startedBy", None) + self.started_on = synapse_response.get("startedOn", None) + self.error_message = synapse_response.get("errorMessage", None) + self.error_details = synapse_response.get("errorDetails", None) + return self + + def to_synapse_request(self) -> dict[str, Any]: + """ + Converts this dataclass to a dictionary suitable for a Synapse REST API request. + + Every field that is set is sent. The Synapse status endpoint replaces + executionDetails wholesale rather than merging, so omitting a field that the + server has populated deletes it. Details constructed empty, as they are when + making a task executable, still serialize to just the concreteType because + every other field is None. + + Returns: + A dictionary representation of this object for API requests. + """ + request_dict: dict[str, Any] = { + "concreteType": self.concrete_type, + "asyncJobId": self.async_job_id, + "startedBy": self.started_by, + "startedOn": self.started_on, + "errorMessage": self.error_message, + "errorDetails": self.error_details, + } + delete_none_keys(request_dict) + return request_dict + + +@dataclass +class SampleSheetGenerationExecutionDetails(ExecutableTaskExecutionDetails): + """ + Execution details for a curation task that generates a sample sheet. Used with + the task's SampleSheetGenerationExecutionProperties. + + + + Attributes: + async_job_id: The ID of the async job currently executing this task. Set when + execution starts; cleared on completion or failure. + started_by: The principal ID of the user who started the execution. + started_on: When the execution was started. + error_message: If execution failed, the error description. + error_details: If execution failed, additional error details. + """ + + @property + def concrete_type(self) -> str: + """The concreteType of a SampleSheetGenerationExecutionDetails.""" + return SAMPLE_SHEET_GENERATION_EXECUTION_DETAILS + + +@dataclass +class RecordSetGenerationExecutionDetails(ExecutableTaskExecutionDetails): + """ + Execution details for a curation task that generates a RecordSet from source + files. Used with the task's RecordSetGenerationExecutionProperties. + + + + Attributes: + async_job_id: The ID of the async job currently executing this task. Set when + execution starts; cleared on completion or failure. + started_by: The principal ID of the user who started the execution. + started_on: When the execution was started. + error_message: If execution failed, the error description. + error_details: If execution failed, additional error details. + """ + + @property + def concrete_type(self) -> str: + """The concreteType of a RecordSetGenerationExecutionDetails.""" + return RECORD_SET_GENERATION_EXECUTION_DETAILS + + +@dataclass +class UnknownTaskExecutionDetails(TaskExecutionDetails): + """ + Execution details whose concreteType this version of the client does not + recognize. Produced when Synapse returns a TaskExecutionDetails subtype that was + added after this client was released. + + The whole response is kept in raw_details, and the fields common to every + execution details type are exposed as read-only properties over it, so the outcome + of a run, including its error message, can still be read. Serializing these + details back to Synapse therefore does not lose the fields this client cannot + model. The properties are read-only because there is nothing useful to write: an + execution detail this client does not model cannot be constructed correctly, only + reported back as it arrived. + + These details are not an ExecutableTaskExecutionDetails. Not every + TaskExecutionDetails subtype supports automated execution (GridExecutionDetails + does not), so an unrecognized concreteType cannot be assumed to be executable. + A task carrying these details may or may not be accepted by + [CurationTask.execute][synapseclient.models.CurationTask.execute]; upgrade + synapseclient to find out locally rather than from the server. + + Attributes: + raw_details: The unmodified executionDetails from the response. + """ + + raw_details: dict[str, Any] = field(default_factory=dict) + """The unmodified executionDetails from the response.""" + + @property + def concrete_type(self) -> str: + """The concreteType that Synapse reported for these details.""" + return self.raw_details.get("concreteType", "") + + @property + def async_job_id(self) -> str | None: + """The ID of the async job that was executing this task.""" + return self.raw_details.get("asyncJobId", None) + + @property + def started_by(self) -> str | None: + """The principal ID of the user who started the execution.""" + return self.raw_details.get("startedBy", None) - Attributes: - active_session_id: The unique identifier of the active CRDT grid session linked to this task. - """ + @property + def started_on(self) -> str | None: + """When the execution was started.""" + return self.raw_details.get("startedOn", None) - active_session_id: str | None = None - """The unique identifier of the active CRDT grid session linked to this task.""" + @property + def error_message(self) -> str | None: + """If execution failed, the error description.""" + return self.raw_details.get("errorMessage", None) + + @property + def error_details(self) -> str | None: + """If execution failed, additional error details.""" + return self.raw_details.get("errorDetails", None) def fill_from_dict( self, synapse_response: dict[str, Any] - ) -> "GridExecutionDetails": + ) -> "UnknownTaskExecutionDetails": """ Converts a response from the REST API into this dataclass. @@ -390,29 +914,70 @@ def fill_from_dict( synapse_response: The response from the REST API. Returns: - The GridExecutionDetails object. + The UnknownTaskExecutionDetails object. """ - self.active_session_id = synapse_response.get("activeSessionId") + self.raw_details = deepcopy(synapse_response) return self def to_synapse_request(self) -> dict[str, Any]: """ Converts this dataclass to a dictionary suitable for a Synapse REST API request. + Returns the details exactly as Synapse sent them. The status endpoint replaces + executionDetails rather than merging, so a read-modify-write that dropped the + fields this client does not model would delete them server-side. + Returns: A dictionary representation of this object for API requests. + + Raises: + ValueError: If these details carry no concreteType for the server to + dispatch on, either because they were never populated from a Synapse + response or because the response did not name one. """ - request_dict: dict[str, Any] = {"concreteType": GRID_EXECUTION_DETAILS} - if self.active_session_id is not None: - request_dict["activeSessionId"] = self.active_session_id - return request_dict + if not self.concrete_type: + raise ValueError( + "UnknownTaskExecutionDetails can only be serialized after being " + "populated from a Synapse response carrying a concreteType. To attach " + "execution details to a task, construct the type matching its " + "properties, such as SampleSheetGenerationExecutionDetails or " + "RecordSetGenerationExecutionDetails." + ) + return deepcopy(self.raw_details) TASK_EXECUTION_DETAILS_DICT: dict[str, type[TaskExecutionDetails]] = { GRID_EXECUTION_DETAILS: GridExecutionDetails, + SAMPLE_SHEET_GENERATION_EXECUTION_DETAILS: SampleSheetGenerationExecutionDetails, + RECORD_SET_GENERATION_EXECUTION_DETAILS: RecordSetGenerationExecutionDetails, } +def _create_task_execution_details_from_dict( + details_dict: dict[str, Any], +) -> TaskExecutionDetails: + """ + Factory method to create the appropriate TaskExecutionDetails implementation + based on the concreteType. + + An unrecognized concreteType is not an error. Execution details are a read-only + report about work that Synapse has already done, so refusing to parse them would + discard the result of a completed job. The details are returned as + UnknownTaskExecutionDetails, which reports the concreteType Synapse sent. + + Arguments: + details_dict: Dictionary containing task execution details data + + Returns: + The appropriate TaskExecutionDetails instance + """ + concrete_type = details_dict.get("concreteType", "") + task_execution_details = TASK_EXECUTION_DETAILS_DICT.get( + concrete_type, UnknownTaskExecutionDetails + ) + return task_execution_details().fill_from_dict(details_dict) + + @dataclass class CurationTaskStatus(EnumCoercionMixin): """ @@ -469,16 +1034,11 @@ def fill_from_dict(self, synapse_response: dict[str, Any]) -> "CurationTaskStatu self.etag = synapse_response.get("etag") details_dict: dict[str, Any] | None = synapse_response.get("executionDetails") - if details_dict is None: - self.execution_details = None - else: - concrete_type = details_dict.get("concreteType", "") - cls = TASK_EXECUTION_DETAILS_DICT.get(concrete_type) - if cls is None: - raise ValueError( - f"Unknown concreteType for TaskExecutionDetails: {concrete_type}" - ) - self.execution_details = cls().fill_from_dict(details_dict) + self.execution_details = ( + None + if details_dict is None + else _create_task_execution_details_from_dict(details_dict) + ) return self def to_synapse_request(self) -> dict[str, Any]: @@ -676,6 +1236,60 @@ def set_active_grid_session( """ return CurationTaskStatus() + def set_execution_details( + self, + *, + execution_details: "TaskExecutionDetails", + synapse_client: Synapse | None = None, + ) -> "CurationTaskStatus": + """ + Replace the execution details on this CurationTask's status. + + Fetches the current CurationTaskStatus first so the update carries a fresh + etag, then writes back the given execution details. Does not transition the + task state. + + A compute task needs this before it can run: a newly created task has no + execution details, and Synapse will not dispatch one without details that + support automated execution. Pass empty details of the type matching the + task's properties, such as SampleSheetGenerationExecutionDetails or + RecordSetGenerationExecutionDetails, and Synapse populates their fields as + the job runs. + + Arguments: + execution_details: The execution details to attach to this task's status. + synapse_client: If not passed in and caching was not disabled by + Synapse.allow_client_caching(False) this will use the last created + instance from the Synapse class constructor. + + Returns: + The updated CurationTaskStatus object. + + Raises: + ValueError: If the CurationTask object does not have a task_id. + + Example: Make a compute task executable +   + + ```python + from synapseclient import Synapse + from synapseclient.models import ( + CurationTask, + RecordSetGenerationExecutionDetails, + ) + + syn = Synapse() + syn.login() + + task = CurationTask(task_id=123) + task.set_execution_details( + execution_details=RecordSetGenerationExecutionDetails() + ) + task.execute() + ``` + """ + return CurationTaskStatus() + def set_task_state( self, state: "TaskState | str", @@ -799,6 +1413,66 @@ def create_grid_session( """ return Grid() + def execute( + self, + *, + timeout: int = 120, + synapse_client: Synapse | None = None, + ) -> "TaskExecutionDetails": + """ + Run the automated computation for this CurationTask and wait for it to finish. + + The task must be in the NOT_STARTED state and its status must carry execution + details that support automated execution, such as + SampleSheetGenerationExecutionDetails or RecordSetGenerationExecutionDetails. + The computation itself is described by the task's task_properties, either + SampleSheetGenerationExecutionProperties or + RecordSetGenerationExecutionProperties. + + A newly created task has no execution details, and Synapse will not dispatch + it until they are set. Attach empty details of the matching type with + set_execution_details once, before the first run: + + task.set_execution_details( + execution_details=RecordSetGenerationExecutionDetails() + ) + + The caller must be the assignee of the task or have UPDATE access on the + task's project. + + Arguments: + timeout: Seconds to wait for the execution job to complete or progress + before raising a SynapseTimeoutError. Defaults to 120. + synapse_client: If not passed in and caching was not disabled by + Synapse.allow_client_caching(False) this will use the last created + instance from the Synapse class constructor. + + Returns: + The execution details of the task after the job completed. + + Raises: + ValueError: If the CurationTask object does not have a task_id. + SynapseError: If the execution job fails, or if it completes without + returning execution details. + SynapseTimeoutError: If the execution job does not complete within the + timeout. + + Example: Execute a curation task +   + + ```python + from synapseclient import Synapse + from synapseclient.models import CurationTask + + syn = Synapse() + syn.login() + + details = CurationTask(task_id=123).execute() + print(details.started_on) + ``` + """ + return RecordSetGenerationExecutionDetails() + def delete( self, delete_source: bool = False, @@ -811,12 +1485,16 @@ def delete( Arguments: delete_source: If True, the associated source data (EntityView or RecordSet) will also be deleted if the task is a FileBasedMetadataTask or RecordBasedMetadataTask respectively. Defaults to False. + A compute task has no source of its own, so passing True for one raises + a ValueError. synapse_client: If not passed in and caching was not disabled by `Synapse.allow_client_caching(False)` this will use the last created instance from the Synapse class constructor. Raises: ValueError: If the CurationTask object does not have a task_id. + ValueError: If delete_source is True and the task properties do not + identify a source to delete. Example: Delete a curation task   @@ -1078,8 +1756,10 @@ class CurationTask(CurationTaskSynchronousProtocol): data_type: Will match the data type that a contributor plans to contribute project_id: The synId of the project instructions: Instructions to the data contributor - task_properties: The properties of a CurationTask. This can be either - FileBasedMetadataTaskProperties or RecordBasedMetadataTaskProperties. + task_properties: The properties of a CurationTask. This can be + FileBasedMetadataTaskProperties, RecordBasedMetadataTaskProperties, + SampleSheetGenerationExecutionProperties, or + RecordSetGenerationExecutionProperties. etag: Synapse employs an Optimistic Concurrency Control (OCC) scheme to handle concurrent updates. Since the E-Tag changes every time an entity is updated it is used to detect when a client's current representation of an entity is @@ -1137,9 +1817,7 @@ class CurationTask(CurationTaskSynchronousProtocol): instructions: Optional[str] = None """Instructions to the data contributor""" - task_properties: Optional[ - Union[FileBasedMetadataTaskProperties, RecordBasedMetadataTaskProperties] - ] = None + task_properties: Optional[CurationTaskProperties] = None """The properties of a CurationTask""" etag: Optional[str] = None @@ -1361,6 +2039,8 @@ async def delete_async( Arguments: delete_source: If True, the associated source data (EntityView or RecordSet) will also be deleted if the task is a FileBasedMetadataTask or RecordBasedMetadataTask respectively. Defaults to False. + A compute task has no source of its own, so passing True for one raises + a ValueError. synapse_client: If not passed in and caching was not disabled by `Synapse.allow_client_caching(False)` this will use the last created instance from the Synapse class constructor. @@ -1445,13 +2125,29 @@ async def main(): synapse_client=synapse_client ) - else: + elif self.task_properties is None: raise ValueError( - "'task_property' attribute is None. " + "'task_properties' attribute is None. " "Deletion only supports FileBasedMetadataTaskProperties or " "RecordBasedMetadataTaskProperties." ) + elif isinstance(self.task_properties, UnknownCurationTaskProperties): + raise ValueError( + "delete_source is not supported for task properties of type " + f"{self.task_properties.concrete_type}, which this version of " + "synapseclient does not recognize, so the source of the task " + "cannot be identified. Upgrade synapseclient, or delete this task " + "without delete_source." + ) + + else: + raise ValueError( + "delete_source is not supported for " + f"{type(self.task_properties).__name__}. A compute task has no " + "source of its own. Delete this task without delete_source." + ) + await delete_curation_task(task_id=self.task_id, synapse_client=synapse_client) async def store_async( @@ -1669,16 +2365,80 @@ async def main(): asyncio.run(main()) ``` """ - status = await self.get_status_async(synapse_client=synapse_client) - status.execution_details = GridExecutionDetails( - active_session_id=active_session_id + return await self.set_execution_details_async( + execution_details=GridExecutionDetails(active_session_id=active_session_id), + synapse_client=synapse_client, + ) + + @otel_trace_method( + method_to_trace_name=lambda self, *args, **kwargs: ( + f"CurationTask_SetExecutionDetails: ID: {self.task_id}" ) + ) + async def set_execution_details_async( + self, + *, + execution_details: "TaskExecutionDetails", + synapse_client: Synapse | None = None, + ) -> "CurationTaskStatus": + """ + Replace the execution details on this CurationTask's status. + + Fetches the current CurationTaskStatus first so the update carries a fresh + etag, then writes back the given execution details. Does not transition the + task state. + + A compute task needs this before it can run: a newly created task has no + execution details, and Synapse will not dispatch one without details that + support automated execution. Pass empty details of the type matching the + task's properties, such as SampleSheetGenerationExecutionDetails or + RecordSetGenerationExecutionDetails, and Synapse populates their fields as + the job runs. + + Arguments: + execution_details: The execution details to attach to this task's status. + synapse_client: If not passed in and caching was not disabled by + Synapse.allow_client_caching(False) this will use the last created + instance from the Synapse class constructor. + + Returns: + The updated CurationTaskStatus object. + + Raises: + ValueError: If the CurationTask object does not have a task_id. + + Example: Make a compute task executable asynchronously +   + + ```python + import asyncio + from synapseclient import Synapse + from synapseclient.models import ( + CurationTask, + RecordSetGenerationExecutionDetails, + ) + + syn = Synapse() + syn.login() + + async def main(): + task = CurationTask(task_id=123) + await task.set_execution_details_async( + execution_details=RecordSetGenerationExecutionDetails() + ) + await task.execute_async() + + asyncio.run(main()) + ``` + """ + status = await self.get_status_async(synapse_client=synapse_client) + status.execution_details = execution_details return await self.update_status_async( curation_task_status=status, synapse_client=synapse_client ) @otel_trace_method( - method_to_trace_name=lambda self, **kwargs: ( + method_to_trace_name=lambda self, *args, **kwargs: ( f"CurationTask_SetTaskState: ID: {self.task_id}" ) ) @@ -1915,6 +2675,88 @@ async def main(): return grid + @otel_trace_method( + method_to_trace_name=lambda self, **kwargs: ( + f"CurationTask_Execute: ID: {self.task_id}" + ) + ) + async def execute_async( + self, + *, + timeout: int = 120, + synapse_client: Synapse | None = None, + ) -> "TaskExecutionDetails": + """ + Run the automated computation for this CurationTask and wait for it to finish. + + The task must be in the NOT_STARTED state and its status must carry execution + details that support automated execution, such as + SampleSheetGenerationExecutionDetails or RecordSetGenerationExecutionDetails. + The computation itself is described by the task's task_properties, either + SampleSheetGenerationExecutionProperties or + RecordSetGenerationExecutionProperties. + + A newly created task has no execution details, and Synapse will not dispatch + it until they are set. Attach empty details of the matching type with + set_execution_details once, before the first run: + + task.set_execution_details( + execution_details=RecordSetGenerationExecutionDetails() + ) + + The caller must be the assignee of the task or have UPDATE access on the + task's project. + + Arguments: + timeout: Seconds to wait for the execution job to complete or progress + before raising a SynapseTimeoutError. Defaults to 120. + synapse_client: If not passed in and caching was not disabled by + Synapse.allow_client_caching(False) this will use the last created + instance from the Synapse class constructor. + + Returns: + The execution details of the task after the job completed. + + Raises: + ValueError: If the CurationTask object does not have a task_id. + SynapseError: If the execution job fails, or if it completes without + returning execution details. + SynapseTimeoutError: If the execution job does not complete within the + timeout. + + Example: Execute a curation task asynchronously +   + + ```python + import asyncio + from synapseclient import Synapse + from synapseclient.models import CurationTask + + syn = Synapse() + syn.login() + + async def main(): + details = await CurationTask(task_id=123).execute_async() + print(details.started_on) + + asyncio.run(main()) + ``` + """ + if not self.task_id: + raise ValueError("task_id is required to execute a CurationTask") + + request = ComputeTaskExecutionRequest(task_id=self.task_id) + result = await request.send_job_and_wait_async( + timeout=timeout, + synapse_client=synapse_client, + ) + if result.execution_details is None: + raise SynapseError( + f"The execution job for CurationTask {self.task_id} completed without " + "returning execution details." + ) + return result.execution_details + @skip_async_to_sync @classmethod async def list_async( @@ -2085,6 +2927,72 @@ async def main(): yield task +@dataclass +class ComputeTaskExecutionRequest(AsynchronousCommunicator): + """ + Start a job to execute the automated computation for a CurationTask. + + The task must be in the NOT_STARTED state and have execution details that support + automated execution (an ExecutableTaskExecutionDetails). + + Represents a [Synapse ComputeTaskExecutionRequest](https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/curation/ComputeTaskExecutionRequest.html) + and the [ComputeTaskExecutionResponse](https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/curation/ComputeTaskExecutionResponse.html) + it produces. + + Attributes: + concrete_type: The concrete type for the request + task_id: The ID of the CurationTask to execute + execution_details: The execution details of the task after the job completed + (populated from response) + """ + + concrete_type: str = COMPUTE_TASK_EXECUTION_REQUEST + """The concrete type for the request""" + + task_id: int | None = None + """The ID of the CurationTask to execute""" + + execution_details: TaskExecutionDetails | None = None + """The execution details of the task after the job completed. The concrete type + determines which task-type-specific properties are available.""" + + def fill_from_dict(self, synapse_response: Any) -> "ComputeTaskExecutionRequest": + """ + Converts a response from the REST API into this dataclass. + + Arguments: + synapse_response: The response from the REST API. + + Returns: + The ComputeTaskExecutionRequest object. + """ + task_id = synapse_response.get("taskId", None) + if task_id is not None: + self.task_id = int(task_id) + + details_dict = synapse_response.get("executionDetails", None) + self.execution_details = ( + None + if details_dict is None + else _create_task_execution_details_from_dict(details_dict) + ) + return self + + def to_synapse_request(self) -> dict[str, Any]: + """ + Converts this dataclass to a dictionary suitable for a Synapse REST API request. + + Returns: + A dictionary representation of this object for API requests. + """ + request_dict = { + "concreteType": self.concrete_type, + "taskId": self.task_id, + } + delete_none_keys(request_dict) + return request_dict + + @dataclass class CreateGridRequest(EnumCoercionMixin, AsynchronousCommunicator): """ diff --git a/synapseclient/models/mixins/CLAUDE.md b/synapseclient/models/mixins/CLAUDE.md index 90ec7a098..dede83932 100644 --- a/synapseclient/models/mixins/CLAUDE.md +++ b/synapseclient/models/mixins/CLAUDE.md @@ -1,4 +1,4 @@ - + ## Project @@ -15,6 +15,8 @@ Queue-based concurrent download/upload via `_worker()` coroutine processing `asy ### asynchronous_job.py `ASYNC_JOB_URIS` dict maps concrete types to REST endpoints — when adding a new async job type, register here AND in `core/constants/concrete_types.py`. Subclasses must implement `to_synapse_request()` and `fill_from_dict()`. +Resource-scoped endpoints use `{placeholder}` segments (e.g. `/curation/task/{taskId}/execute/async`). `_resolve_async_job_uri()` fills them from the request body, matching each placeholder to the identically-named camelCase key emitted by `to_synapse_request()` — so a `{taskId}` URI requires the request to carry `taskId`. + ### table_components.py Column type mapping between Python types and Synapse column types. Multiple TODOs for incomplete features (SYNPY-1651). diff --git a/synapseclient/models/mixins/asynchronous_job.py b/synapseclient/models/mixins/asynchronous_job.py index 88cce73be..bf27870d7 100644 --- a/synapseclient/models/mixins/asynchronous_job.py +++ b/synapseclient/models/mixins/asynchronous_job.py @@ -3,7 +3,9 @@ import time from dataclasses import dataclass from enum import Enum +from string import Formatter from typing import Any, Dict, Optional +from urllib.parse import quote from tqdm.contrib.logging import logging_redirect_tqdm from typing_extensions import Self @@ -11,6 +13,7 @@ from synapseclient import Synapse from synapseclient.core.constants.concrete_types import ( AGENT_CHAT_REQUEST, + COMPUTE_TASK_EXECUTION_REQUEST, CREATE_GRID_REQUEST, CREATE_SCHEMA_REQUEST, DOWNLOAD_FROM_GRID_REQUEST, @@ -33,6 +36,7 @@ ASYNC_JOB_URIS = { AGENT_CHAT_REQUEST: "/agent/chat/async", + COMPUTE_TASK_EXECUTION_REQUEST: "/curation/task/{taskId}/execute/async", CREATE_GRID_REQUEST: "/grid/session/async", DOWNLOAD_FROM_GRID_REQUEST: "/grid/download/csv/async", DOWNLOAD_LIST_MANIFEST_REQUEST: "/download/list/manifest/async", @@ -48,6 +52,72 @@ } +def _resolve_async_job_uri( + request_type: str, request: dict[str, Any] | None = None +) -> str: + """ + Looks up the asynchronous job URI for a request type and fills in any path + placeholders (for example {entityId} or {taskId}) from the request body. + + A placeholder matches the camelCase key that the request's to_synapse_request() + emits, so a URI of /curation/task/{taskId}/execute/async is resolved with the + taskId of the request. + + Placeholders are read with the same parser that fills them in, so a URI whose + placeholder cannot name a request key is rejected here rather than being sent to + Synapse with its braces intact. + + Each value is percent-encoded as a single path segment. Request values come from + plain model attributes that are not type-checked at runtime, so a value carrying + a slash or a dot-dot segment must not be able to change which endpoint is called. + + Arguments: + request_type: The concreteType of the asynchronous job request. + request: The request that was, or will be, sent to the server. Only required + when the URI for the request type contains placeholders. + + Returns: + The asynchronous job URI with all placeholders resolved. + + Raises: + ValueError: If the request type is not supported, if the URI has a placeholder + that is not a plain name, or if it has a placeholder that is not present + in the request. + """ + if not request_type or request_type not in ASYNC_JOB_URIS: + raise ValueError(f"Unsupported request type: {request_type}") + + uri = ASYNC_JOB_URIS[request_type] + placeholders = [ + field_name + for _, field_name, _, _ in Formatter().parse(uri) + if field_name is not None + ] + if not placeholders: + return uri + + unusable = [name for name in placeholders if not name.isidentifier()] + if unusable: + raise ValueError( + f"Cannot resolve async job uri {uri}: {unusable} cannot name a request " + "key. Each placeholder must be a plain camelCase name matching a key " + "that to_synapse_request() emits." + ) + + if not request: + raise ValueError(f"Cannot resolve async job uri {uri}: no request provided.") + + for placeholder in placeholders: + if request.get(placeholder) is None: + raise ValueError( + f"Cannot resolve async job uri {uri}: missing {placeholder} in request." + ) + + return uri.format( + **{key: quote(str(request[key]), safe="") for key in placeholders} + ) + + class AsynchronousCommunicator: """Mixin to handle communication with the Synapse Asynchronous Job service.""" @@ -395,15 +465,8 @@ async def send_job_async( request_type = request.get("concreteType") - if not request_type or request_type not in ASYNC_JOB_URIS: - raise ValueError(f"Unsupported request type: {request_type}") - client = Synapse.get_client(synapse_client=synapse_client) - uri = ASYNC_JOB_URIS[request_type] - if "{entityId}" in uri: - if "entityId" not in request: - raise ValueError(f"Attempting to send job with missing id in uri: {uri}") - uri = uri.format(entityId=request["entityId"]) + uri = _resolve_async_job_uri(request_type=request_type, request=request) response = await client.rest_post_async( uri=f"{uri}/start", body=json.dumps(request) @@ -452,23 +515,14 @@ async def get_job_async( last_progress = 0 last_total = 1 progressed = False + uri = _resolve_async_job_uri(request_type=request_type, request=request) progress_bar = create_progress_bar( total=last_total, - desc="", + desc=uri, synapse_client=client, ) with logging_redirect_tqdm(loggers=[client.logger]): while time.time() - start_time < timeout: - uri = ASYNC_JOB_URIS[request_type] - if "{entityId}" in uri: - if not request: - raise ValueError("Attempting to get job with missing request.") - if "entityId" not in request: - raise ValueError( - f"Attempting to get job with missing id in uri: {uri}" - ) - uri = uri.format(entityId=request["entityId"]) - progress_bar.desc = uri result = await client.rest_get_async( uri=f"{uri}/get/{job_id}", endpoint=endpoint, diff --git a/tests/integration/synapseclient/models/async/test_curation_async.py b/tests/integration/synapseclient/models/async/test_curation_async.py index 77bbc7934..dfd668e01 100644 --- a/tests/integration/synapseclient/models/async/test_curation_async.py +++ b/tests/integration/synapseclient/models/async/test_curation_async.py @@ -16,19 +16,28 @@ CurationTask, CurationTaskStatus, EntityView, + File, FileBasedMetadataTaskProperties, Folder, Grid, GridExecutionDetails, + JSONSchema, Project, RecordBasedMetadataTaskProperties, RecordSet, + RecordSetGenerationExecutionDetails, + RecordSetGenerationExecutionProperties, + SampleSheetGenerationExecutionDetails, + SampleSheetGenerationExecutionProperties, + SchemaOrganization, TaskState, UserProfile, ViewTypeMask, + query_async, ) from synapseclient.models.table_components import Query -from tests.integration import ASYNC_JOB_TIMEOUT_SEC +from tests.integration import ASYNC_JOB_TIMEOUT_SEC, QUERY_TIMEOUT_SEC +from tests.integration.helpers import wait_for_condition @pytest.fixture(scope="function") @@ -955,3 +964,310 @@ async def test_set_task_state_async( # AND the change persists on the server refetched_status = await stored_task.get_status_async(synapse_client=syn) assert refetched_status.state == TaskState.IN_PROGRESS + + +class TestCurationTaskExecuteAsync: + """Tests for the CurationTask.execute_async method.""" + + # The generated sample sheet format is established by the JSON Schema bound to the + # destination RecordSet, so both compute task kinds need a destination whose + # RecordSet carries a schema. + SAMPLE_SHEET_COLUMNS = ["specimen_id", "assay", "read_length"] + + @pytest.fixture(scope="function") + async def sample_sheet_schema_uri( + self, syn: Synapse, request: pytest.FixtureRequest + ) -> str: + """Create a JSON schema describing the target sample sheet format.""" + random_name = "".join(i for i in str(uuid.uuid4()) if i.isalpha()) + organization = SchemaOrganization(name=f"SYNPY.TEST.{random_name}") + await organization.store_async(synapse_client=syn) + + json_schema = JSONSchema( + name="curation.compute.samplesheet", organization_name=organization.name + ) + await json_schema.store_async( + schema_body={ + "$schema": "http://json-schema.org/draft-07/schema#", + "title": "Curation Compute Sample Sheet", + "type": "object", + "properties": { + "specimen_id": { + "description": "The identifier of the specimen", + "type": "string", + }, + "assay": { + "description": "The assay the specimen was run through", + "type": "string", + }, + "read_length": { + "description": "The sequencing read length", + "type": "integer", + }, + }, + }, + version="0.0.1", + synapse_client=syn, + ) + + def delete_organization() -> None: + for schema in organization.get_json_schemas(synapse_client=syn): + schema.delete(synapse_client=syn) + organization.delete(synapse_client=syn) + + request.addfinalizer(delete_organization) + + return json_schema.uri + + @pytest.fixture(scope="function") + async def destination_task( + self, + syn: Synapse, + project_model: Project, + schedule_for_cleanup: Callable[..., None], + sample_sheet_schema_uri: str, + ) -> CurationTask: + """ + Create the record-based CurationTask that a compute task writes its output to. + + Its RecordSet is seeded with the target columns and has the sample sheet + schema bound to it. + """ + folder = await Folder( + name=str(uuid.uuid4()), + parent_id=project_model.id, + ).store_async(synapse_client=syn) + schedule_for_cleanup(folder.id) + + temp_fd, filename = tempfile.mkstemp(suffix=".csv") + os.close(temp_fd) + schedule_for_cleanup(filename) + pd.DataFrame(columns=self.SAMPLE_SHEET_COLUMNS).to_csv(filename, index=False) + + record_set = await RecordSet( + name=str(uuid.uuid4()), + parent_id=folder.id, + path=filename, + upsert_keys=["specimen_id"], + ).store_async(synapse_client=syn) + schedule_for_cleanup(record_set.id) + + await record_set.bind_schema_async( + json_schema_uri=sample_sheet_schema_uri, synapse_client=syn + ) + + return await CurationTask( + data_type=f"destination_{str(uuid.uuid4()).replace('-', '_')}", + project_id=project_model.id, + instructions="Receives the output of a compute task.", + task_properties=RecordBasedMetadataTaskProperties( + record_set_id=record_set.id + ), + ).store_async(synapse_client=syn) + + async def test_execute_record_set_generation_async( + self, + syn: Synapse, + project_model: Project, + destination_task: CurationTask, + schedule_for_cleanup: Callable[..., None], + ) -> None: + # GIVEN a folder of small source files to transform + source_folder = await Folder( + name=str(uuid.uuid4()), + parent_id=project_model.id, + ).store_async(synapse_client=syn) + schedule_for_cleanup(source_folder.id) + + for index, specimen in enumerate(["SPEC_001", "SPEC_002"]): + temp_fd, source_path = tempfile.mkstemp(suffix=".csv") + os.close(temp_fd) + schedule_for_cleanup(source_path) + pd.DataFrame( + { + "specimen_id": [specimen], + "assay": ["rna_seq"], + "read_length": [100 + index], + } + ).to_csv(source_path, index=False) + + source_file = await File( + parent_id=source_folder.id, + name=f"source_{index}.csv", + path=source_path, + ).store_async(synapse_client=syn) + schedule_for_cleanup(source_file.id) + + # AND a compute task that transforms that folder into the destination RecordSet + compute_task = await CurationTask( + data_type=f"record_set_generation_{str(uuid.uuid4()).replace('-', '_')}", + project_id=project_model.id, + instructions="Transform the source files into the destination RecordSet.", + task_properties=RecordSetGenerationExecutionProperties( + folder_id=source_folder.id, + instructions=( + "Each file describes one specimen. Produce one row per specimen " + "with its specimen_id, assay, and read_length." + ), + destination_task_id=destination_task.task_id, + ), + ).store_async(synapse_client=syn) + + # AND the round trip preserved the compute properties + assert isinstance( + compute_task.task_properties, RecordSetGenerationExecutionProperties + ) + assert compute_task.task_properties.folder_id == source_folder.id + assert ( + compute_task.task_properties.destination_task_id == destination_task.task_id + ) + + # AND the task is made executable, since a new task has no execution details + # and Synapse will not dispatch one without them + status_with_details = await compute_task.set_execution_details_async( + execution_details=RecordSetGenerationExecutionDetails(), + synapse_client=syn, + ) + assert isinstance( + status_with_details.execution_details, RecordSetGenerationExecutionDetails + ) + # AND attaching the details did not transition the task out of NOT_STARTED + assert status_with_details.state == TaskState.NOT_STARTED + + # WHEN I execute the task + details = await compute_task.execute_async( + timeout=ASYNC_JOB_TIMEOUT_SEC, synapse_client=syn + ) + + # THEN the job reports back details of the matching kind + assert isinstance(details, RecordSetGenerationExecutionDetails) + assert details.started_by is not None + assert details.started_on is not None + assert details.error_message is None, ( + "The record set generation job failed: " + f"{details.error_message} {details.error_details}" + ) + + # AND the task's status carries the same kind of details, with the results + # left pending human review + final_status = await compute_task.get_status_async(synapse_client=syn) + assert isinstance( + final_status.execution_details, RecordSetGenerationExecutionDetails + ) + assert final_status.execution_details.started_on == details.started_on + assert final_status.state == TaskState.IN_REVIEW + + async def test_execute_sample_sheet_generation_async( + self, + syn: Synapse, + project_model: Project, + destination_task: CurationTask, + folder_with_view: tuple[Folder, EntityView], + schedule_for_cleanup: Callable[..., None], + ) -> None: + # GIVEN a folder whose files carry the annotations the sample sheet is built + # from, and an EntityView over that folder + folder, entity_view = folder_with_view + + for index, specimen in enumerate(["SPEC_001", "SPEC_002"]): + annotated_file = await File( + parent_id=folder.id, + name=f"annotated_{index}.txt", + path=make_bogus_uuid_file(), + annotations={ + "specimen_id": [specimen], + "assay": ["rna_seq"], + "read_length": [100 + index], + }, + ).store_async(synapse_client=syn) + schedule_for_cleanup(annotated_file.id) + + # AND the view has indexed those files, since the computation reads the + # annotations through the view rather than from the file contents + async def view_has_indexed_the_files() -> bool: + rows = await query_async( + query=f"SELECT * FROM {entity_view.id}", synapse_client=syn + ) + return len(rows) >= 2 + + await wait_for_condition( + view_has_indexed_the_files, + timeout_seconds=QUERY_TIMEOUT_SEC, + description="entity view to index the annotated files", + ) + + # AND a file-based curation task over that folder to act as the input + input_task = await CurationTask( + data_type=f"sample_sheet_input_{str(uuid.uuid4()).replace('-', '_')}", + project_id=project_model.id, + instructions="Provides the annotated files for the sample sheet.", + task_properties=FileBasedMetadataTaskProperties( + upload_folder_id=folder.id, + file_view_id=entity_view.id, + ), + ).store_async(synapse_client=syn) + + # AND a compute task that builds a sample sheet from the input task + compute_task = await CurationTask( + data_type=f"sample_sheet_generation_{str(uuid.uuid4()).replace('-', '_')}", + project_id=project_model.id, + instructions="Generate the sample sheet from the annotated files.", + task_properties=SampleSheetGenerationExecutionProperties( + input_task_id=input_task.task_id, + destination_task_id=destination_task.task_id, + ), + ).store_async(synapse_client=syn) + + # AND the round trip preserved the compute properties + assert isinstance( + compute_task.task_properties, SampleSheetGenerationExecutionProperties + ) + assert compute_task.task_properties.input_task_id == input_task.task_id + assert ( + compute_task.task_properties.destination_task_id == destination_task.task_id + ) + + # AND the task is made executable, since a new task has no execution details + # and Synapse will not dispatch one without them + status_with_details = await compute_task.set_execution_details_async( + execution_details=SampleSheetGenerationExecutionDetails(), + synapse_client=syn, + ) + assert isinstance( + status_with_details.execution_details, + SampleSheetGenerationExecutionDetails, + ) + # AND attaching the details did not transition the task out of NOT_STARTED + assert status_with_details.state == TaskState.NOT_STARTED + + # WHEN I execute the task + details = await compute_task.execute_async( + timeout=ASYNC_JOB_TIMEOUT_SEC, synapse_client=syn + ) + + # THEN the job reports back details of the matching kind + assert isinstance(details, SampleSheetGenerationExecutionDetails) + assert details.started_by is not None + assert details.started_on is not None + assert details.error_message is None, ( + "The sample sheet generation job failed: " + f"{details.error_message} {details.error_details}" + ) + + # AND the task's status carries the same kind of details, with the results + # left pending human review + final_status = await compute_task.get_status_async(synapse_client=syn) + assert isinstance( + final_status.execution_details, SampleSheetGenerationExecutionDetails + ) + assert final_status.execution_details.started_on == details.started_on + assert final_status.state == TaskState.IN_REVIEW + + async def test_execute_validation_error_async(self, syn: Synapse) -> None: + # GIVEN a CurationTask without a task_id + # WHEN I try to execute it + # THEN it should raise a ValueError before any request is sent + with pytest.raises( + ValueError, match="task_id is required to execute a CurationTask" + ): + await CurationTask().execute_async(synapse_client=syn) diff --git a/tests/unit/synapseclient/mixins/async/unit_test_asynchronous_job.py b/tests/unit/synapseclient/mixins/async/unit_test_asynchronous_job.py index b771119ab..6bb00f309 100644 --- a/tests/unit/synapseclient/mixins/async/unit_test_asynchronous_job.py +++ b/tests/unit/synapseclient/mixins/async/unit_test_asynchronous_job.py @@ -6,18 +6,194 @@ import pytest from synapseclient import Synapse -from synapseclient.core.constants.concrete_types import AGENT_CHAT_REQUEST +from synapseclient.core.constants.concrete_types import ( + AGENT_CHAT_REQUEST, + COMPUTE_TASK_EXECUTION_REQUEST, + QUERY_BUNDLE_REQUEST, + TABLE_UPDATE_TRANSACTION_REQUEST, +) from synapseclient.core.exceptions import SynapseError, SynapseTimeoutError from synapseclient.models.mixins.asynchronous_job import ( ASYNC_JOB_URIS, AsynchronousJobState, AsynchronousJobStatus, + _resolve_async_job_uri, get_job_async, send_job_and_wait_async, send_job_async, ) +class TestResolveAsyncJobUri: + """Unit tests for _resolve_async_job_uri.""" + + @pytest.mark.parametrize( + "request_type,job_request,expected_uri", + [ + (AGENT_CHAT_REQUEST, None, "/agent/chat/async"), + ( + AGENT_CHAT_REQUEST, + {"concreteType": AGENT_CHAT_REQUEST, "sessionId": "123"}, + "/agent/chat/async", + ), + ( + COMPUTE_TASK_EXECUTION_REQUEST, + {"concreteType": COMPUTE_TASK_EXECUTION_REQUEST, "taskId": 42}, + "/curation/task/42/execute/async", + ), + ( + COMPUTE_TASK_EXECUTION_REQUEST, + {"concreteType": COMPUTE_TASK_EXECUTION_REQUEST, "taskId": "42"}, + "/curation/task/42/execute/async", + ), + ( + COMPUTE_TASK_EXECUTION_REQUEST, + {"concreteType": COMPUTE_TASK_EXECUTION_REQUEST, "taskId": 0}, + "/curation/task/0/execute/async", + ), + ( + COMPUTE_TASK_EXECUTION_REQUEST, + { + "concreteType": COMPUTE_TASK_EXECUTION_REQUEST, + "taskId": 42, + "unrelatedKey": "ignored", + }, + "/curation/task/42/execute/async", + ), + ( + TABLE_UPDATE_TRANSACTION_REQUEST, + { + "concreteType": TABLE_UPDATE_TRANSACTION_REQUEST, + "entityId": "syn123", + "changes": [], + }, + "/entity/syn123/table/transaction/async", + ), + ], + ids=[ + "static_uri_needs_no_request", + "static_uri_ignores_request", + "placeholder_filled_from_request", + "placeholder_value_is_a_string", + "placeholder_value_is_zero", + "unrelated_keys_are_ignored", + "entity_id_placeholder", + ], + ) + def test_uri_is_resolved( + self, request_type: str, job_request: dict | None, expected_uri: str + ) -> None: + """A registered request type maps to a uri with no placeholders left in it.""" + uri = _resolve_async_job_uri(request_type=request_type, request=job_request) + assert uri == expected_uri + + @pytest.mark.parametrize( + "request_type,job_request,expected_error", + [ + ( + "InvalidConcreteType", + {}, + "Unsupported request type: InvalidConcreteType", + ), + (None, {}, "Unsupported request type: None"), + (COMPUTE_TASK_EXECUTION_REQUEST, None, "no request provided"), + ( + COMPUTE_TASK_EXECUTION_REQUEST, + {"concreteType": COMPUTE_TASK_EXECUTION_REQUEST}, + "missing taskId in request", + ), + ( + COMPUTE_TASK_EXECUTION_REQUEST, + {"concreteType": COMPUTE_TASK_EXECUTION_REQUEST, "taskId": None}, + "missing taskId in request", + ), + ], + ids=[ + "unregistered_request_type", + "no_request_type", + "placeholder_with_no_request", + "placeholder_absent_from_request", + "placeholder_none_in_request", + ], + ) + def test_unresolvable_uri_raises( + self, request_type: str | None, job_request: dict | None, expected_error: str + ) -> None: + """A uri that cannot be completed fails loudly instead of being requested.""" + with pytest.raises(ValueError, match=expected_error): + _resolve_async_job_uri(request_type=request_type, request=job_request) + + @pytest.mark.parametrize( + "uri", + [ + "/foo/{entity-id}/async", + "/foo/{taskId}/{entity-id}/async", + ], + ids=["only_placeholder_is_unparseable", "one_placeholder_is_unparseable"], + ) + def test_placeholder_that_is_not_an_identifier_raises( + self, monkeypatch: pytest.MonkeyPatch, uri: str + ) -> None: + """A placeholder the resolver cannot fill must raise, not reach the server.""" + # GIVEN a registered uri with a placeholder that is not a plain identifier. + fake_request_type = "org.sagebionetworks.repo.model.FakeRequest" + monkeypatch.setitem(ASYNC_JOB_URIS, fake_request_type, uri) + + # WHEN I resolve it with a request that carries every value the uri names + # THEN I should get a ValueError naming what could not be resolved, rather + # than a uri that still contains braces and would be sent to Synapse as-is + with pytest.raises(ValueError, match="cannot name a request key"): + _resolve_async_job_uri( + request_type=fake_request_type, + request={ + "concreteType": fake_request_type, + "taskId": 42, + "entity-id": "syn123", + }, + ) + + def test_synapse_ids_are_not_altered_by_encoding(self) -> None: + """Percent-encoding leaves the ids callers actually pass untouched.""" + # GIVEN a request carrying an ordinary synId + uri = _resolve_async_job_uri( + request_type=QUERY_BUNDLE_REQUEST, + request={"concreteType": QUERY_BUNDLE_REQUEST, "entityId": "syn123"}, + ) + + # THEN the synId should appear verbatim: percent-encoding must not disturb + # the values that are actually used in practice + assert uri == "/entity/syn123/table/query/async" + + @pytest.mark.parametrize( + "task_id,expected", + [ + ("123/../../../admin", "123%2F..%2F..%2F..%2Fadmin"), + ("123?foo=bar", "123%3Ffoo%3Dbar"), + ("123 456", "123%20456"), + ], + ids=["path_traversal", "query_injection", "space"], + ) + def test_placeholder_is_encoded_as_a_single_path_segment( + self, task_id: str, expected: str + ) -> None: + """A placeholder value cannot change which endpoint is called.""" + # GIVEN a request whose id carries characters that are significant in a uri. + # task_id is annotated int but dataclasses do not enforce annotations, so a + # string can reach this point. + uri = _resolve_async_job_uri( + request_type=COMPUTE_TASK_EXECUTION_REQUEST, + request={ + "concreteType": COMPUTE_TASK_EXECUTION_REQUEST, + "taskId": task_id, + }, + ) + + # THEN the value should be percent-encoded, leaving the surrounding path + # structure intact: the id still occupies exactly one segment + assert uri == f"/curation/task/{expected}/execute/async" + assert uri.split("/") == ["", "curation", "task", expected, "execute", "async"] + + class TestSendJobAsync: """Unit tests for send_job_async.""" diff --git a/tests/unit/synapseclient/models/async/unit_test_curation_async.py b/tests/unit/synapseclient/models/async/unit_test_curation_async.py index 1e5859964..cd308bd97 100644 --- a/tests/unit/synapseclient/models/async/unit_test_curation_async.py +++ b/tests/unit/synapseclient/models/async/unit_test_curation_async.py @@ -7,28 +7,45 @@ from synapseclient import Synapse from synapseclient.core.constants.concrete_types import ( + COMPUTE_TASK_EXECUTION_REQUEST, FILE_BASED_METADATA_TASK_PROPERTIES, GRID_CSV_IMPORT_REQUEST, GRID_EXECUTION_DETAILS, RECORD_BASED_METADATA_TASK_PROPERTIES, + RECORD_SET_GENERATION_EXECUTION_DETAILS, + RECORD_SET_GENERATION_EXECUTION_PROPERTIES, + SAMPLE_SHEET_GENERATION_EXECUTION_DETAILS, + SAMPLE_SHEET_GENERATION_EXECUTION_PROPERTIES, UPLOAD_TO_TABLE_PREVIEW_REQUEST, ) +from synapseclient.core.exceptions import SynapseError from synapseclient.models import EntityView, RecordSet from synapseclient.models.curation import ( AuthorizationMode, + ComputeTaskExecutionRequest, CreateGridRequest, CurationTask, + CurationTaskProperties, CurationTaskStatus, DownloadFromGridRequest, + ExecutableTaskExecutionDetails, FileBasedMetadataTaskProperties, Grid, GridCsvImportRequest, GridExecutionDetails, GridRecordSetExportRequest, RecordBasedMetadataTaskProperties, + RecordSetGenerationExecutionDetails, + RecordSetGenerationExecutionProperties, + SampleSheetGenerationExecutionDetails, + SampleSheetGenerationExecutionProperties, SynchronizeGridRequest, + TaskExecutionDetails, TaskState, + UnknownCurationTaskProperties, + UnknownTaskExecutionDetails, UploadToTablePreviewRequest, + _create_task_execution_details_from_dict, _create_task_properties_from_dict, ) from synapseclient.models.recordset import ValidationSummary @@ -55,6 +72,15 @@ STARTED_ON = "2024-03-01T00:00:00.000Z" FILE_HANDLE_ID = "1234567" OWNER_PRINCIPAL_ID = 987654 +ASYNC_JOB_ID = "async-job-abc-123" +ERROR_MESSAGE = "Execution failed" +ERROR_DETAILS = "A longer explanation of the failure" +UNKNOWN_EXECUTION_DETAILS_CONCRETE_TYPE = ( + "org.sagebionetworks.repo.model.curation.execution.FutureExecutionDetails" +) +UNKNOWN_TASK_PROPERTIES_CONCRETE_TYPE = ( + "org.sagebionetworks.repo.model.curation.metadata.FutureTaskProperties" +) def _get_file_based_task_api_response(): @@ -141,6 +167,47 @@ def _get_curation_task_status_response( return response +class TestCurationTaskProperties: + """Tests for the CurationTaskProperties abstract base class.""" + + def test_cannot_be_instantiated(self) -> None: + # GIVEN the abstract base class + # WHEN I try to instantiate it + # THEN a TypeError is raised because it defines no concreteType + with pytest.raises(TypeError, match="abstract"): + CurationTaskProperties() + + @pytest.mark.parametrize( + "properties_class,expected_concrete_type", + [ + (FileBasedMetadataTaskProperties, FILE_BASED_METADATA_TASK_PROPERTIES), + (RecordBasedMetadataTaskProperties, RECORD_BASED_METADATA_TASK_PROPERTIES), + ( + SampleSheetGenerationExecutionProperties, + SAMPLE_SHEET_GENERATION_EXECUTION_PROPERTIES, + ), + ( + RecordSetGenerationExecutionProperties, + RECORD_SET_GENERATION_EXECUTION_PROPERTIES, + ), + ], + ) + def test_implementations_share_the_base_contract( + self, properties_class: type, expected_concrete_type: str + ) -> None: + # GIVEN an implementation of CurationTaskProperties + properties = properties_class() + + # THEN it is an instance of the base class and reports its concreteType + assert isinstance(properties, CurationTaskProperties) + assert properties.concrete_type == expected_concrete_type + + # AND an empty instance serializes to just the concreteType + assert properties.to_synapse_request() == { + "concreteType": expected_concrete_type + } + + class TestFileBasedMetadataTaskProperties: """Tests for the FileBasedMetadataTaskProperties dataclass.""" @@ -310,14 +377,513 @@ def test_record_based_properties(self) -> None: assert isinstance(result, RecordBasedMetadataTaskProperties) assert result.record_set_id == RECORD_SET_ID - def test_unknown_concrete_type_raises_error(self) -> None: - # GIVEN a dict with an unknown concrete type - data = {"concreteType": "org.sagebionetworks.Unknown"} + def test_sample_sheet_generation_properties(self) -> None: + # GIVEN a dict with the sample sheet generation concrete type + data = { + "concreteType": SAMPLE_SHEET_GENERATION_EXECUTION_PROPERTIES, + "inputTaskId": TASK_ID, + "destinationTaskId": TASK_ID_2, + } + + # WHEN I create task properties from the dict + result = _create_task_properties_from_dict(data) + + # THEN it should be a SampleSheetGenerationExecutionProperties + assert isinstance(result, SampleSheetGenerationExecutionProperties) + assert result.input_task_id == TASK_ID + assert result.destination_task_id == TASK_ID_2 + + def test_record_set_generation_properties(self) -> None: + # GIVEN a dict with the record set generation concrete type + data = { + "concreteType": RECORD_SET_GENERATION_EXECUTION_PROPERTIES, + "folderId": UPLOAD_FOLDER_ID, + "instructions": INSTRUCTIONS, + "destinationTaskId": TASK_ID_2, + } + + # WHEN I create task properties from the dict + result = _create_task_properties_from_dict(data) + + # THEN it should be a RecordSetGenerationExecutionProperties + assert isinstance(result, RecordSetGenerationExecutionProperties) + assert result.folder_id == UPLOAD_FOLDER_ID + assert result.instructions == INSTRUCTIONS + assert result.destination_task_id == TASK_ID_2 + + def test_unknown_concrete_type_falls_back(self) -> None: + # GIVEN a dict with a concrete type this client does not recognize + data = { + "concreteType": UNKNOWN_TASK_PROPERTIES_CONCRETE_TYPE, + "someFutureField": "a value this client knows nothing about", + } + + # WHEN I create task properties from it + result = _create_task_properties_from_dict(data) + + # THEN the properties are returned as the fallback rather than raising, so the + # rest of the task remains readable + assert isinstance(result, UnknownCurationTaskProperties) + assert result.concrete_type == UNKNOWN_TASK_PROPERTIES_CONCRETE_TYPE + assert result.raw_properties == data + + +class TestUnknownCurationTaskProperties: + """Tests for the fallback used when a properties concreteType is not recognized.""" + + def test_to_synapse_request_round_trips_unmodelled_fields(self) -> None: + """Serializing back must not drop the fields this client cannot model.""" + # GIVEN properties of an unknown type carrying a field no known subtype has + response = { + "concreteType": UNKNOWN_TASK_PROPERTIES_CONCRETE_TYPE, + "uploadFolderId": UPLOAD_FOLDER_ID, + "someFutureField": "a value this client knows nothing about", + } + + # WHEN I read the properties and serialize them back + properties = UnknownCurationTaskProperties().fill_from_dict(response) + request = properties.to_synapse_request() + + # THEN the response is reproduced exactly, so storing a task this client cannot + # fully model does not erase the parts it does not understand + assert request == response + + def test_nested_values_are_not_shared_with_the_response(self) -> None: + """The copy must be deep: the point of raw_properties is a verbatim record.""" + # GIVEN a response whose unmodelled portion is a nested structure + response = { + "concreteType": UNKNOWN_TASK_PROPERTIES_CONCRETE_TYPE, + "someFutureField": {"nested": "original"}, + } + properties = UnknownCurationTaskProperties().fill_from_dict(response) + + # WHEN the caller mutates the nested value in the original response + response["someFutureField"]["nested"] = "mutated" + + # THEN the properties still hold what Synapse actually sent + assert properties.raw_properties["someFutureField"]["nested"] == "original" + + def test_to_synapse_request_is_a_copy(self) -> None: + # GIVEN properties read from a response + response = { + "concreteType": UNKNOWN_TASK_PROPERTIES_CONCRETE_TYPE, + "uploadFolderId": UPLOAD_FOLDER_ID, + } + properties = UnknownCurationTaskProperties().fill_from_dict(response) + + # WHEN a caller mutates the request dict + request = properties.to_synapse_request() + request["uploadFolderId"] = "mutated" + + # THEN neither the properties nor the original response are affected + assert properties.to_synapse_request()["uploadFolderId"] == UPLOAD_FOLDER_ID + assert response["uploadFolderId"] == UPLOAD_FOLDER_ID + + def test_concrete_type_is_empty_when_absent(self) -> None: + # GIVEN properties that were never populated from a Synapse response + properties = UnknownCurationTaskProperties() + + # THEN the concrete type reads as empty rather than raising + assert properties.concrete_type == "" + + def test_to_synapse_request_without_a_response_raises(self) -> None: + """An empty fallback must not be sent as taskProperties. + + The class is exported, so a user can construct one; serializing it would send + a taskProperties with no concreteType for the server to dispatch on. + """ + # GIVEN properties that were never populated from a Synapse response + properties = UnknownCurationTaskProperties() + + # WHEN I convert them to a request dict + # THEN it should raise rather than produce a payload with no concreteType + with pytest.raises( + ValueError, match="can only be serialized after being populated" + ): + properties.to_synapse_request() + + def test_to_synapse_request_without_a_concrete_type_raises(self) -> None: + """Populated but unusable properties must not be sent either. + + The fallback catches a missing concreteType as well as an unrecognized one, + so raw_properties can be non-empty while still carrying nothing for the + server to dispatch on. Guarding on emptiness alone would let that through. + """ + # GIVEN properties read from a response that named no concreteType + properties = _create_task_properties_from_dict( + {"someFutureField": "a value with nothing to dispatch on"} + ) + + # THEN the fallback should have accepted it, non-empty but with no type + assert isinstance(properties, UnknownCurationTaskProperties) + assert properties.raw_properties + assert properties.concrete_type == "" + + # WHEN I convert them to a request dict + # THEN it should raise here rather than sending a payload the server will + # reject for having no concreteType + with pytest.raises( + ValueError, match="can only be serialized after being populated" + ): + properties.to_synapse_request() + + async def test_delete_source_is_refused(self, syn: Synapse) -> None: + """delete_source cannot work when the source cannot be identified.""" + # GIVEN a task whose properties this client does not recognize + task = CurationTask( + task_id=TASK_ID, + task_properties=UnknownCurationTaskProperties().fill_from_dict( + {"concreteType": UNKNOWN_TASK_PROPERTIES_CONCRETE_TYPE} + ), + ) + + # WHEN I delete it with delete_source + # THEN it should raise, naming the type Synapse reported + with pytest.raises( + ValueError, + match=f"delete_source is not supported for task properties of type " + f"{UNKNOWN_TASK_PROPERTIES_CONCRETE_TYPE}", + ): + await task.delete_async(delete_source=True, synapse_client=syn) + + +class TestSampleSheetGenerationExecutionProperties: + """Tests for the SampleSheetGenerationExecutionProperties dataclass.""" + + def test_fill_from_dict(self) -> None: + # GIVEN a response dict where the task ids are returned as strings + response = { + "concreteType": SAMPLE_SHEET_GENERATION_EXECUTION_PROPERTIES, + "inputTaskId": str(TASK_ID), + "destinationTaskId": str(TASK_ID_2), + } + + # WHEN I fill the properties from the dict + props = SampleSheetGenerationExecutionProperties().fill_from_dict(response) + + # THEN the task ids should be coerced to ints + assert props.input_task_id == TASK_ID + assert props.destination_task_id == TASK_ID_2 + + def test_fill_from_dict_empty(self) -> None: + # GIVEN a response dict with no task ids + # WHEN I fill the properties from the dict + props = SampleSheetGenerationExecutionProperties().fill_from_dict( + {"concreteType": SAMPLE_SHEET_GENERATION_EXECUTION_PROPERTIES} + ) + + # THEN both task ids should be None rather than coerced from None + assert props.input_task_id is None + assert props.destination_task_id is None + + def test_to_synapse_request(self) -> None: + # GIVEN properties with only an input task id + props = SampleSheetGenerationExecutionProperties(input_task_id=TASK_ID) + + # WHEN I convert them to a request dict + request = props.to_synapse_request() + + # THEN the concreteType is included and the absent id is dropped + assert request == { + "concreteType": SAMPLE_SHEET_GENERATION_EXECUTION_PROPERTIES, + "inputTaskId": TASK_ID, + } + + +class TestRecordSetGenerationExecutionProperties: + """Tests for the RecordSetGenerationExecutionProperties dataclass.""" + + def test_fill_from_dict(self) -> None: + # GIVEN a response dict where the destination task id is returned as a string + response = { + "concreteType": RECORD_SET_GENERATION_EXECUTION_PROPERTIES, + "folderId": UPLOAD_FOLDER_ID, + "instructions": INSTRUCTIONS, + "destinationTaskId": str(TASK_ID_2), + } + + # WHEN I fill the properties from the dict + props = RecordSetGenerationExecutionProperties().fill_from_dict(response) + + # THEN all fields should be populated and the id coerced to an int + assert props.folder_id == UPLOAD_FOLDER_ID + assert props.instructions == INSTRUCTIONS + assert props.destination_task_id == TASK_ID_2 + + def test_to_synapse_request(self) -> None: + # GIVEN properties with a folder and instructions but no destination + props = RecordSetGenerationExecutionProperties( + folder_id=UPLOAD_FOLDER_ID, instructions=INSTRUCTIONS + ) + + # WHEN I convert them to a request dict + request = props.to_synapse_request() + + # THEN the concreteType is included and the absent id is dropped + assert request == { + "concreteType": RECORD_SET_GENERATION_EXECUTION_PROPERTIES, + "folderId": UPLOAD_FOLDER_ID, + "instructions": INSTRUCTIONS, + } + + +class TestExecutableTaskExecutionDetails: + """Tests for the ExecutableTaskExecutionDetails implementations.""" + + @pytest.mark.parametrize( + "details_class,concrete_type", + [ + ( + SampleSheetGenerationExecutionDetails, + SAMPLE_SHEET_GENERATION_EXECUTION_DETAILS, + ), + ( + RecordSetGenerationExecutionDetails, + RECORD_SET_GENERATION_EXECUTION_DETAILS, + ), + ], + ) + def test_fill_from_dict(self, details_class, concrete_type: str) -> None: + # GIVEN a response dict for a failed execution + response = { + "concreteType": concrete_type, + "asyncJobId": ASYNC_JOB_ID, + "startedBy": STARTED_BY, + "startedOn": STARTED_ON, + "errorMessage": ERROR_MESSAGE, + "errorDetails": ERROR_DETAILS, + } + + # WHEN I fill the details from the dict + details = details_class().fill_from_dict(response) + + # THEN every field should be populated + assert details.async_job_id == ASYNC_JOB_ID + assert details.started_by == STARTED_BY + assert details.started_on == STARTED_ON + assert details.error_message == ERROR_MESSAGE + assert details.error_details == ERROR_DETAILS + assert details.concrete_type == concrete_type + + @pytest.mark.parametrize( + "details_class,concrete_type", + [ + ( + SampleSheetGenerationExecutionDetails, + SAMPLE_SHEET_GENERATION_EXECUTION_DETAILS, + ), + ( + RecordSetGenerationExecutionDetails, + RECORD_SET_GENERATION_EXECUTION_DETAILS, + ), + ], + ) + def test_to_synapse_request_empty_sends_only_concrete_type( + self, details_class, concrete_type: str + ) -> None: + # GIVEN empty details, as constructed to make a task executable + details = details_class() + + # WHEN I convert them to a request dict + request = details.to_synapse_request() + + # THEN only the concreteType is sent: every other field is None and dropped, + # so set_execution_details does not claim ownership of server-managed fields + assert request == {"concreteType": concrete_type} + + @pytest.mark.parametrize( + "details_class,concrete_type", + [ + ( + SampleSheetGenerationExecutionDetails, + SAMPLE_SHEET_GENERATION_EXECUTION_DETAILS, + ), + ( + RecordSetGenerationExecutionDetails, + RECORD_SET_GENERATION_EXECUTION_DETAILS, + ), + ], + ) + def test_to_synapse_request_round_trips_populated_fields( + self, details_class, concrete_type: str + ) -> None: + # GIVEN details read back from an earlier failed run + details = details_class( + async_job_id=ASYNC_JOB_ID, + started_by=STARTED_BY, + started_on=STARTED_ON, + error_message=ERROR_MESSAGE, + error_details=ERROR_DETAILS, + ) + + # WHEN I convert them to a request dict + request = details.to_synapse_request() + + # THEN every populated field is sent. The status endpoint replaces + # executionDetails rather than merging, so a read-modify-write that omitted + # these would delete the recorded failure reason server-side. + assert request == { + "concreteType": concrete_type, + "asyncJobId": ASYNC_JOB_ID, + "startedBy": STARTED_BY, + "startedOn": STARTED_ON, + "errorMessage": ERROR_MESSAGE, + "errorDetails": ERROR_DETAILS, + } + + +class TestUnknownTaskExecutionDetails: + """Tests for the fallback used when a concreteType is not recognized.""" + + def test_to_synapse_request_round_trips_unmodelled_fields(self) -> None: + """Serializing back must not drop the fields this client cannot model.""" + # GIVEN details of an unknown type carrying a field no known subtype has + response = { + "concreteType": UNKNOWN_EXECUTION_DETAILS_CONCRETE_TYPE, + "startedOn": STARTED_ON, + "errorMessage": ERROR_MESSAGE, + "someFutureField": "a value this client knows nothing about", + } + + # WHEN I read the details and serialize them back + details = UnknownTaskExecutionDetails().fill_from_dict(response) + request = details.to_synapse_request() + + # THEN the response is reproduced exactly. The status endpoint replaces + # executionDetails rather than merging, so a set_task_state that serialized + # only the fields this client understands would erase the rest server-side + assert request == response + + def test_to_synapse_request_is_a_copy(self) -> None: + # GIVEN details read from a response + response = { + "concreteType": UNKNOWN_EXECUTION_DETAILS_CONCRETE_TYPE, + "startedOn": STARTED_ON, + } + details = UnknownTaskExecutionDetails().fill_from_dict(response) + + # WHEN a caller mutates the request dict + request = details.to_synapse_request() + request["startedOn"] = "mutated" + + # THEN neither the details nor the original response are affected + assert details.to_synapse_request()["startedOn"] == STARTED_ON + assert response["startedOn"] == STARTED_ON + + def test_nested_values_are_not_shared_with_the_response(self) -> None: + """The copy must be deep: the point of raw_details is a verbatim record.""" + # GIVEN a response whose unmodelled portion is a nested structure + response = { + "concreteType": UNKNOWN_EXECUTION_DETAILS_CONCRETE_TYPE, + "someFutureField": {"nested": "original"}, + } + details = UnknownTaskExecutionDetails().fill_from_dict(response) + + # WHEN the caller mutates the nested value in the original response + response["someFutureField"]["nested"] = "mutated" + + # THEN the details still hold what Synapse actually sent + assert details.raw_details["someFutureField"]["nested"] == "original" + + def test_common_fields_are_read_from_the_raw_response(self) -> None: + # GIVEN details of an unknown type carrying every common field + details = UnknownTaskExecutionDetails().fill_from_dict( + { + "concreteType": UNKNOWN_EXECUTION_DETAILS_CONCRETE_TYPE, + "asyncJobId": ASYNC_JOB_ID, + "startedBy": STARTED_BY, + "startedOn": STARTED_ON, + "errorMessage": ERROR_MESSAGE, + "errorDetails": ERROR_DETAILS, + } + ) + + # THEN every common field is readable as a property over the raw response + assert details.concrete_type == UNKNOWN_EXECUTION_DETAILS_CONCRETE_TYPE + assert details.async_job_id == ASYNC_JOB_ID + assert details.started_by == STARTED_BY + assert details.started_on == STARTED_ON + assert details.error_message == ERROR_MESSAGE + assert details.error_details == ERROR_DETAILS + + def test_common_fields_are_none_when_absent(self) -> None: + # GIVEN details carrying nothing but a concreteType + details = UnknownTaskExecutionDetails().fill_from_dict( + {"concreteType": UNKNOWN_EXECUTION_DETAILS_CONCRETE_TYPE} + ) + + # THEN the common fields read as None rather than raising + assert details.async_job_id is None + assert details.started_by is None + assert details.started_on is None + assert details.error_message is None + assert details.error_details is None + + def test_to_synapse_request_without_a_response_raises(self) -> None: + """An empty fallback must not be sent as executionDetails. + + The class is exported, so a user can construct one; serializing it would PUT + an executionDetails with no concreteType, and the status endpoint replaces + rather than merges. + """ + # GIVEN details that were never populated from a Synapse response + details = UnknownTaskExecutionDetails() + + # WHEN I convert them to a request dict + # THEN it should raise rather than produce a payload with no concreteType + with pytest.raises( + ValueError, match="can only be serialized after being populated" + ): + details.to_synapse_request() + + def test_to_synapse_request_without_a_concrete_type_raises(self) -> None: + """Populated but unusable details must not be PUT either. + + The fallback catches a missing concreteType as well as an unrecognized one, + so raw_details can be non-empty while still carrying nothing for the server + to dispatch on. Guarding on emptiness alone would let that through. + """ + # GIVEN details read from a response that named no concreteType + details = _create_task_execution_details_from_dict({"asyncJobId": ASYNC_JOB_ID}) + + # THEN the fallback should have accepted it, non-empty but with no type + assert isinstance(details, UnknownTaskExecutionDetails) + assert details.raw_details + assert details.concrete_type == "" + + # WHEN I convert them to a request dict + # THEN it should raise here rather than PUTting a payload the server will + # reject for having no concreteType + with pytest.raises( + ValueError, match="can only be serialized after being populated" + ): + details.to_synapse_request() + + def test_is_not_an_executable_task_execution_details(self) -> None: + """An unrecognized concreteType must not claim to support execution. + + Not every TaskExecutionDetails subtype is executable (GridExecutionDetails is + not), so a type added after this client was released cannot be assumed to be. + Callers gate execute() on isinstance(details, ExecutableTaskExecutionDetails), + which would wave through a task the server will refuse. + """ + # GIVEN details of a concreteType this client does not recognize + details = UnknownTaskExecutionDetails().fill_from_dict( + { + "concreteType": UNKNOWN_EXECUTION_DETAILS_CONCRETE_TYPE, + "asyncJobId": ASYNC_JOB_ID, + "errorMessage": ERROR_MESSAGE, + } + ) - # WHEN I attempt to create task properties - # THEN it should raise a ValueError - with pytest.raises(ValueError, match="Unknown concreteType"): - _create_task_properties_from_dict(data) + # THEN they are a TaskExecutionDetails, but not an executable one + assert isinstance(details, TaskExecutionDetails) + assert not isinstance(details, ExecutableTaskExecutionDetails) + + # AND the fields common to every execution details type are still readable, + # which is the whole reason the fallback exists + assert details.async_job_id == ASYNC_JOB_ID + assert details.error_message == ERROR_MESSAGE class TestGridExecutionDetails: @@ -348,6 +914,23 @@ def test_to_synapse_request(self) -> None: assert request["concreteType"] == GRID_EXECUTION_DETAILS assert request["activeSessionId"] == SESSION_ID + def test_concrete_type(self) -> None: + # GIVEN a GridExecutionDetails + # WHEN I read its concrete type + # THEN it should be readable without serializing the object, the same as + # every other TaskExecutionDetails implementation + assert GridExecutionDetails().concrete_type == GRID_EXECUTION_DETAILS + + def test_concrete_type_is_not_a_dataclass_field(self) -> None: + # GIVEN two GridExecutionDetails with the same session id + # WHEN I compare them + # THEN they should be equal: concrete_type is a property, so it does not + # take part in the generated __init__ or __eq__ + assert GridExecutionDetails(active_session_id=SESSION_ID) == ( + GridExecutionDetails(active_session_id=SESSION_ID) + ) + assert "concrete_type" not in GridExecutionDetails.__dataclass_fields__ + class TestCurationTaskStatus: """Tests for the CurationTaskStatus dataclass.""" @@ -384,21 +967,144 @@ def test_fill_from_dict_without_execution_details(self) -> None: assert status.state == TaskState.NOT_STARTED def test_fill_from_dict_unknown_execution_details_concrete_type(self) -> None: - """An unrecognized concreteType in executionDetails raises ValueError.""" + """An unrecognized concreteType in executionDetails does not raise.""" # GIVEN a status response with an unknown executionDetails concreteType response = _get_curation_task_status_response(state="NOT_STARTED") response["executionDetails"] = { - "concreteType": "org.sagebionetworks.repo.model.curation.execution.Unknown", - "activeSessionId": SESSION_ID, + "concreteType": UNKNOWN_EXECUTION_DETAILS_CONCRETE_TYPE, + "startedOn": STARTED_ON, + "errorMessage": ERROR_MESSAGE, } # WHEN I fill a CurationTaskStatus from it - # THEN it should raise ValueError, consistent with how unknown state - # values and unknown task-properties concrete types are handled - with pytest.raises( - ValueError, match="Unknown concreteType for TaskExecutionDetails" - ): - CurationTaskStatus().fill_from_dict(response) + status = CurationTaskStatus().fill_from_dict(response) + + # THEN the status should still be readable. Execution details report on work + # the server has already done, so a subtype added after this client was + # released must not make the whole status unreadable + assert isinstance(status.execution_details, UnknownTaskExecutionDetails) + assert ( + status.execution_details.concrete_type + == UNKNOWN_EXECUTION_DETAILS_CONCRETE_TYPE + ) + assert status.execution_details.started_on == STARTED_ON + assert status.execution_details.error_message == ERROR_MESSAGE + assert status.state == TaskState.NOT_STARTED + + @pytest.mark.parametrize( + "concrete_type,details_class", + [ + ( + SAMPLE_SHEET_GENERATION_EXECUTION_DETAILS, + SampleSheetGenerationExecutionDetails, + ), + ( + RECORD_SET_GENERATION_EXECUTION_DETAILS, + RecordSetGenerationExecutionDetails, + ), + ], + ) + def test_fill_from_dict_executable_execution_details( + self, concrete_type: str, details_class + ) -> None: + """The executable execution details concrete types resolve to their classes.""" + # GIVEN a status response carrying executable execution details + response = _get_curation_task_status_response(state="EXECUTING") + response["executionDetails"] = { + "concreteType": concrete_type, + "asyncJobId": ASYNC_JOB_ID, + "startedBy": STARTED_BY, + "startedOn": STARTED_ON, + } + + # WHEN I fill a CurationTaskStatus from it + status = CurationTaskStatus().fill_from_dict(response) + + # THEN the execution details should be the matching class, fully populated + assert isinstance(status.execution_details, details_class) + assert status.execution_details.async_job_id == ASYNC_JOB_ID + assert status.execution_details.started_by == STARTED_BY + assert status.execution_details.started_on == STARTED_ON + assert status.state == TaskState.EXECUTING + + +class TestComputeTaskExecutionRequest: + """Tests for the ComputeTaskExecutionRequest async job dataclass.""" + + def test_to_synapse_request(self) -> None: + # GIVEN a request for a task + request = ComputeTaskExecutionRequest(task_id=TASK_ID) + + # WHEN I convert it to a request dict + result = request.to_synapse_request() + + # THEN it should carry the concreteType and the taskId, which the async job + # layer also uses to resolve the /curation/task/{taskId}/execute/async URI + assert result == { + "concreteType": COMPUTE_TASK_EXECUTION_REQUEST, + "taskId": TASK_ID, + } + + def test_fill_from_dict(self) -> None: + # GIVEN a ComputeTaskExecutionResponse body + response = { + "taskId": str(TASK_ID), + "executionDetails": { + "concreteType": SAMPLE_SHEET_GENERATION_EXECUTION_DETAILS, + "startedBy": STARTED_BY, + "startedOn": STARTED_ON, + }, + } + + # WHEN I fill the request from the response + request = ComputeTaskExecutionRequest().fill_from_dict(response) + + # THEN the task id is coerced to an int and the details are deserialized + assert request.task_id == TASK_ID + assert isinstance( + request.execution_details, SampleSheetGenerationExecutionDetails + ) + assert request.execution_details.started_by == STARTED_BY + + def test_fill_from_dict_without_execution_details(self) -> None: + # GIVEN a response body with no execution details + # WHEN I fill the request from the response + request = ComputeTaskExecutionRequest(task_id=TASK_ID).fill_from_dict( + {"taskId": TASK_ID} + ) + + # THEN execution_details should be None + assert request.execution_details is None + assert request.task_id == TASK_ID + + def test_fill_from_dict_unknown_execution_details_concrete_type(self) -> None: + # GIVEN a response body for a job that ran to completion, carrying an + # executionDetails concreteType this client does not know + response = { + "taskId": TASK_ID, + "executionDetails": { + "concreteType": UNKNOWN_EXECUTION_DETAILS_CONCRETE_TYPE, + "asyncJobId": ASYNC_JOB_ID, + "startedBy": STARTED_BY, + "startedOn": STARTED_ON, + }, + } + + # WHEN I fill the request from the response + request = ComputeTaskExecutionRequest().fill_from_dict(response) + + # THEN the details common to every execution type are still available. The + # computation already ran server-side, so raising here would strand the user + # with a completed job whose outcome they cannot read + assert isinstance(request.execution_details, UnknownTaskExecutionDetails) + assert ( + request.execution_details.concrete_type + == UNKNOWN_EXECUTION_DETAILS_CONCRETE_TYPE + ) + assert request.execution_details.async_job_id == ASYNC_JOB_ID + assert request.execution_details.started_by == STARTED_BY + assert request.execution_details.started_on == STARTED_ON + assert request.task_id == TASK_ID class TestCurationTask: @@ -583,6 +1289,62 @@ async def test_delete_async_without_task_id(self) -> None: with pytest.raises(ValueError, match="task_id is required to delete"): await task.delete_async(synapse_client=self.syn) + @pytest.mark.parametrize( + "task_properties", + [ + SampleSheetGenerationExecutionProperties(destination_task_id=TASK_ID_2), + RecordSetGenerationExecutionProperties( + folder_id=UPLOAD_FOLDER_ID, destination_task_id=TASK_ID_2 + ), + ], + ids=["sample_sheet_generation", "record_set_generation"], + ) + async def test_delete_async_source_for_compute_task_raises( + self, task_properties + ) -> None: + """delete_source is refused for a compute task, which owns no source.""" + # GIVEN a compute task whose properties name a destination owned by another task + task = CurationTask(task_id=TASK_ID, task_properties=task_properties) + + # WHEN I call delete_async with delete_source + # THEN it should raise a ValueError naming the actual properties type rather + # than claiming they are None, and nothing should be deleted + with patch( + "synapseclient.models.curation.delete_curation_task", + new_callable=AsyncMock, + ) as mock_delete: + with pytest.raises( + ValueError, + match=( + "delete_source is not supported for " + f"{type(task_properties).__name__}" + ), + ): + await task.delete_async(delete_source=True, synapse_client=self.syn) + + mock_delete.assert_not_called() + + async def test_delete_async_source_without_task_properties_raises(self) -> None: + """delete_source with properties that stay None reports them as None.""" + # GIVEN a task whose properties are still None after being fetched + task = CurationTask(task_id=TASK_ID) + + # WHEN I call delete_async with delete_source + # THEN it should raise the ValueError for absent properties + with ( + patch.object( + CurationTask, "get_async", new_callable=AsyncMock, return_value=task + ), + patch( + "synapseclient.models.curation.delete_curation_task", + new_callable=AsyncMock, + ) as mock_delete, + ): + with pytest.raises(ValueError, match="'task_properties' attribute is None"): + await task.delete_async(delete_source=True, synapse_client=self.syn) + + mock_delete.assert_not_called() + async def test_store_async_create_new_task(self) -> None: # GIVEN a new CurationTask with all required create fields file_props = FileBasedMetadataTaskProperties( @@ -931,6 +1693,126 @@ async def test_set_active_grid_session_async(self) -> None: assert isinstance(result.execution_details, GridExecutionDetails) assert result.execution_details.active_session_id == SESSION_ID + async def test_set_execution_details_async(self) -> None: + """Verify that set_execution_details_async PUTs the given details with a fresh etag.""" + # GIVEN a compute task with a task_id + task = CurationTask(task_id=TASK_ID) + + # AND a current status with no execution details, and an update response + # that reflects the newly attached ones + get_response = _get_curation_task_status_response(state="NOT_STARTED") + put_response = _get_curation_task_status_response(state="NOT_STARTED") + put_response["executionDetails"] = { + "concreteType": RECORD_SET_GENERATION_EXECUTION_DETAILS + } + + # WHEN I call set_execution_details_async with empty executable details + with ( + patch( + "synapseclient.models.curation.get_curation_task_status", + new_callable=AsyncMock, + return_value=get_response, + ) as mock_get_status, + patch( + "synapseclient.models.curation.update_curation_task_status", + new_callable=AsyncMock, + return_value=put_response, + ) as mock_update_status, + ): + result = await task.set_execution_details_async( + execution_details=RecordSetGenerationExecutionDetails(), + synapse_client=self.syn, + ) + + # THEN it should fetch the current status first + mock_get_status.assert_called_once_with( + task_id=TASK_ID, synapse_client=self.syn + ) + + # AND PUT a payload carrying the fresh etag, the unchanged state, and + # the new execution details + put_kwargs = mock_update_status.call_args.kwargs + assert put_kwargs["task_id"] == TASK_ID + payload = put_kwargs["curation_task_status"] + assert payload["etag"] == STATUS_ETAG + assert payload["state"] == "NOT_STARTED" + assert payload["executionDetails"] == { + "concreteType": RECORD_SET_GENERATION_EXECUTION_DETAILS + } + + # AND it should return the parsed update response + assert isinstance(result, CurationTaskStatus) + assert isinstance( + result.execution_details, RecordSetGenerationExecutionDetails + ) + + async def test_set_execution_details_async_replaces_existing_details(self) -> None: + """Verify that existing execution details are replaced rather than merged.""" + # GIVEN a task whose status already carries grid execution details + task = CurationTask(task_id=TASK_ID) + get_response = _get_curation_task_status_response( + state="NOT_STARTED", active_session_id=SESSION_ID + ) + put_response = _get_curation_task_status_response(state="NOT_STARTED") + put_response["executionDetails"] = { + "concreteType": SAMPLE_SHEET_GENERATION_EXECUTION_DETAILS + } + + # WHEN I attach details of a different type + with ( + patch( + "synapseclient.models.curation.get_curation_task_status", + new_callable=AsyncMock, + return_value=get_response, + ), + patch( + "synapseclient.models.curation.update_curation_task_status", + new_callable=AsyncMock, + return_value=put_response, + ) as mock_update_status, + ): + await task.set_execution_details_async( + execution_details=SampleSheetGenerationExecutionDetails(), + synapse_client=self.syn, + ) + + # THEN the payload carries only the new details, with no trace of the + # grid session that was there before + payload = mock_update_status.call_args.kwargs["curation_task_status"] + assert payload["executionDetails"] == { + "concreteType": SAMPLE_SHEET_GENERATION_EXECUTION_DETAILS + } + + async def test_set_execution_details_async_without_task_id(self) -> None: + """Verify that set_execution_details_async raises ValueError when task_id is not set.""" + # GIVEN a CurationTask without a task_id + task = CurationTask() + + # WHEN I call set_execution_details_async + # THEN it should raise ValueError (propagated from get_status_async) + with pytest.raises(ValueError, match="task_id is required to get"): + await task.set_execution_details_async( + execution_details=RecordSetGenerationExecutionDetails(), + synapse_client=self.syn, + ) + + async def test_set_execution_details_async_requires_a_keyword(self) -> None: + """execution_details is keyword-only, so a positional call is rejected. + + The tracing decorator builds its span name from a lambda that only accepts + self, so a positional argument reaching it would raise a TypeError naming + the lambda instead of the method. Keyword-only keeps that from happening. + """ + # GIVEN a CurationTask with a task_id + task = CurationTask(task_id=TASK_ID) + + # WHEN I pass the execution details positionally + # THEN Python should reject the call by name, before any API call + with pytest.raises(TypeError, match="set_execution_details_async"): + await task.set_execution_details_async( + RecordSetGenerationExecutionDetails(), synapse_client=self.syn + ) + async def test_set_active_grid_session_async_without_task_id(self) -> None: """Verify that set_active_grid_session_async raises ValueError when task_id is not set.""" # GIVEN a CurationTask without a task_id @@ -1010,6 +1892,95 @@ async def test_set_task_state_async( assert isinstance(result.execution_details, GridExecutionDetails) assert result.execution_details.active_session_id == SESSION_ID + async def test_set_task_state_async_preserves_executable_execution_details( + self, + ) -> None: + """set_task_state_async must not drop the fields a failed run recorded. + + The status endpoint replaces executionDetails rather than merging, so a + read-modify-write that serialized only the concreteType would delete the + failure reason server-side. + """ + # GIVEN a compute task whose last run failed, leaving the error on its + # executable execution details + task = CurationTask(task_id=TASK_ID) + get_response = _get_curation_task_status_response(state="NOT_STARTED") + get_response["executionDetails"] = { + "concreteType": RECORD_SET_GENERATION_EXECUTION_DETAILS, + "asyncJobId": ASYNC_JOB_ID, + "startedBy": STARTED_BY, + "startedOn": STARTED_ON, + "errorMessage": ERROR_MESSAGE, + "errorDetails": ERROR_DETAILS, + } + put_response = _get_curation_task_status_response(state="CANCELED") + put_response["executionDetails"] = get_response["executionDetails"] + + # WHEN I transition the task to another state + with ( + patch( + "synapseclient.models.curation.get_curation_task_status", + new_callable=AsyncMock, + return_value=get_response, + ), + patch( + "synapseclient.models.curation.update_curation_task_status", + new_callable=AsyncMock, + return_value=put_response, + ) as mock_update_status, + ): + await task.set_task_state_async( + state=TaskState.CANCELED, synapse_client=self.syn + ) + + # THEN the PUT payload carries every field that was read, so nothing the + # server recorded about the failed run is erased + payload = mock_update_status.call_args.kwargs["curation_task_status"] + assert payload["state"] == "CANCELED" + assert payload["executionDetails"] == { + "concreteType": RECORD_SET_GENERATION_EXECUTION_DETAILS, + "asyncJobId": ASYNC_JOB_ID, + "startedBy": STARTED_BY, + "startedOn": STARTED_ON, + "errorMessage": ERROR_MESSAGE, + "errorDetails": ERROR_DETAILS, + } + + async def test_set_task_state_async_accepts_a_positional_state(self) -> None: + """state is positional-or-keyword, so passing it positionally must work. + + The tracing decorator forwards positional arguments into the lambda that + builds the span name, so a lambda accepting only self would fail the call + before it reached the method body. + """ + # GIVEN a CurationTask with a task_id + task = CurationTask(task_id=TASK_ID) + get_response = _get_curation_task_status_response(state="NOT_STARTED") + put_response = _get_curation_task_status_response(state="CANCELED") + + # WHEN I pass the state positionally + with ( + patch( + "synapseclient.models.curation.get_curation_task_status", + new_callable=AsyncMock, + return_value=get_response, + ), + patch( + "synapseclient.models.curation.update_curation_task_status", + new_callable=AsyncMock, + return_value=put_response, + ) as mock_update_status, + ): + result = await task.set_task_state_async( + TaskState.CANCELED, synapse_client=self.syn + ) + + # THEN the transition should be sent, rather than the call failing inside + # the tracing decorator + payload = mock_update_status.call_args.kwargs["curation_task_status"] + assert payload["state"] == "CANCELED" + assert result.state == TaskState.CANCELED + async def test_set_task_state_async_invalid_string(self) -> None: """Verify that set_task_state_async raises ValueError before any API call when given an unrecognized state string.""" # GIVEN a CurationTask with a task_id @@ -1350,6 +2321,98 @@ async def mock_list(*args, **kwargs): synapse_client=self.syn, ) + async def test_execute_async(self) -> None: + """execute_async sends a ComputeTaskExecutionRequest and returns the details.""" + # GIVEN a CurationTask with a task_id + task = CurationTask(task_id=TASK_ID) + + # AND a completed job carrying sample sheet generation execution details + completed_request = ComputeTaskExecutionRequest(task_id=TASK_ID) + completed_request.execution_details = SampleSheetGenerationExecutionDetails( + async_job_id=ASYNC_JOB_ID, started_by=STARTED_BY, started_on=STARTED_ON + ) + + # WHEN I call execute_async + with patch.object( + ComputeTaskExecutionRequest, + "send_job_and_wait_async", + new_callable=AsyncMock, + return_value=completed_request, + ) as mock_send_job: + result = await task.execute_async(synapse_client=self.syn) + + # THEN the job should be awaited with the default timeout + mock_send_job.assert_awaited_once_with( + timeout=120, + synapse_client=self.syn, + ) + + # AND the execution details from the response should be returned + assert isinstance(result, SampleSheetGenerationExecutionDetails) + assert result.async_job_id == ASYNC_JOB_ID + assert result.started_by == STARTED_BY + assert result.started_on == STARTED_ON + + async def test_execute_async_passes_timeout(self) -> None: + """A caller-supplied timeout is forwarded to the async job.""" + # GIVEN a CurationTask with a task_id + task = CurationTask(task_id=TASK_ID) + completed_request = ComputeTaskExecutionRequest(task_id=TASK_ID) + completed_request.execution_details = RecordSetGenerationExecutionDetails( + async_job_id=ASYNC_JOB_ID + ) + + # WHEN I call execute_async with a custom timeout + with patch.object( + ComputeTaskExecutionRequest, + "send_job_and_wait_async", + new_callable=AsyncMock, + return_value=completed_request, + ) as mock_send_job: + await task.execute_async(timeout=600, synapse_client=self.syn) + + # THEN the timeout should be forwarded + mock_send_job.assert_awaited_once_with( + timeout=600, + synapse_client=self.syn, + ) + + async def test_execute_async_without_execution_details_raises(self) -> None: + """A completed job that carries no execution details raises SynapseError.""" + # GIVEN a CurationTask with a task_id + task = CurationTask(task_id=TASK_ID) + + # AND a completed job whose response carried no executionDetails + completed_request = ComputeTaskExecutionRequest(task_id=TASK_ID) + assert completed_request.execution_details is None + + # WHEN I call execute_async + # THEN it should raise SynapseError rather than returning None, which callers + # would dereference for started_on/error_message + with patch.object( + ComputeTaskExecutionRequest, + "send_job_and_wait_async", + new_callable=AsyncMock, + return_value=completed_request, + ): + with pytest.raises( + SynapseError, + match=f"execution job for CurationTask {TASK_ID} completed without", + ): + await task.execute_async(synapse_client=self.syn) + + async def test_execute_async_without_task_id(self) -> None: + """execute_async raises ValueError when task_id is not set.""" + # GIVEN a CurationTask without a task_id + task = CurationTask() + + # WHEN I call execute_async + # THEN it should raise ValueError before any API call + with pytest.raises( + ValueError, match="task_id is required to execute a CurationTask" + ): + await task.execute_async(synapse_client=self.syn) + async def test_list_async_state_filter_invalid_string_raises(self) -> None: # GIVEN a state_filter with an invalid string value # WHEN I call list_async From ae2e37c3f567a0055804c5e6c842b221734b1ab7 Mon Sep 17 00:00:00 2001 From: Andrew Lamb Date: Wed, 29 Jul 2026 12:30:11 -0700 Subject: [PATCH 2/7] fix when synapseclient.silent is None --- synapseclient/core/transfer_bar.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/synapseclient/core/transfer_bar.py b/synapseclient/core/transfer_bar.py index f7a4f3ef1..51f854e8d 100644 --- a/synapseclient/core/transfer_bar.py +++ b/synapseclient/core/transfer_bar.py @@ -171,7 +171,7 @@ def create_progress_bar( smoothing=0, postfix=postfix, leave=None, - disable=silent, + disable=bool(silent), ) From 1528b5c91d0e7f8bf10475be438bf31ba821284b Mon Sep 17 00:00:00 2001 From: Andrew Lamb Date: Wed, 29 Jul 2026 12:36:40 -0700 Subject: [PATCH 3/7] fix when synapseclient.silent is None --- synapseclient/models/mixins/asynchronous_job.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/synapseclient/models/mixins/asynchronous_job.py b/synapseclient/models/mixins/asynchronous_job.py index 028926c4a..6796ca85c 100644 --- a/synapseclient/models/mixins/asynchronous_job.py +++ b/synapseclient/models/mixins/asynchronous_job.py @@ -550,7 +550,9 @@ async def get_job_async( last_total = job_status.progress_total updated = False - if progress_bar.desc != last_message: + # A disabled bar (silent client) never stores desc, so read it + # defensively rather than assuming the attribute is present. + if getattr(progress_bar, "desc", None) != last_message: progress_bar.desc = last_message updated = True From a3f2afbf80ea660f13632f95f87d5e1de8a89a98 Mon Sep 17 00:00:00 2001 From: Andrew Lamb Date: Wed, 29 Jul 2026 12:37:21 -0700 Subject: [PATCH 4/7] fix when synapseclient.silent is None --- synapseclient/models/mixins/asynchronous_job.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/synapseclient/models/mixins/asynchronous_job.py b/synapseclient/models/mixins/asynchronous_job.py index 6796ca85c..9990fcc58 100644 --- a/synapseclient/models/mixins/asynchronous_job.py +++ b/synapseclient/models/mixins/asynchronous_job.py @@ -550,8 +550,6 @@ async def get_job_async( last_total = job_status.progress_total updated = False - # A disabled bar (silent client) never stores desc, so read it - # defensively rather than assuming the attribute is present. if getattr(progress_bar, "desc", None) != last_message: progress_bar.desc = last_message updated = True From d2c7091e56fe3d4e538046b4d8575f3d2734d68a Mon Sep 17 00:00:00 2001 From: Andrew Lamb Date: Wed, 29 Jul 2026 12:52:25 -0700 Subject: [PATCH 5/7] use schedule for cleanup --- tests/integration/conftest.py | 5 +++- .../models/async/test_curation_async.py | 25 +++++++++++-------- 2 files changed, 18 insertions(+), 12 deletions(-) diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index 25b466792..fae2d970c 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -20,9 +20,10 @@ from synapseclient.core import utils from synapseclient.core.async_utils import wrap_async_to_sync from synapseclient.core.logging_setup import DEFAULT_LOGGER_NAME, SILENT_LOGGER_NAME -from synapseclient.models import CurationTask, Evaluation, Grid +from synapseclient.models import CurationTask, Evaluation, Grid, JSONSchema from synapseclient.models import Project as Project_Model from synapseclient.models import ( + SchemaOrganization, SubmissionView, Team, WikiHeader, @@ -178,6 +179,8 @@ async def _cleanup(syn: Synapse, items): WikiOrderHint, CurationTask, Grid, + JSONSchema, + SchemaOrganization, ), ): try: diff --git a/tests/integration/synapseclient/models/async/test_curation_async.py b/tests/integration/synapseclient/models/async/test_curation_async.py index dfd668e01..3e26a66ac 100644 --- a/tests/integration/synapseclient/models/async/test_curation_async.py +++ b/tests/integration/synapseclient/models/async/test_curation_async.py @@ -3,7 +3,7 @@ import os import tempfile import uuid -from typing import Callable +from typing import AsyncGenerator, Callable import pandas as pd import pytest @@ -976,12 +976,16 @@ class TestCurationTaskExecuteAsync: @pytest.fixture(scope="function") async def sample_sheet_schema_uri( - self, syn: Synapse, request: pytest.FixtureRequest + self, syn: Synapse, schedule_for_cleanup: Callable[..., None] ) -> str: """Create a JSON schema describing the target sample sheet format.""" random_name = "".join(i for i in str(uuid.uuid4()) if i.isalpha()) organization = SchemaOrganization(name=f"SYNPY.TEST.{random_name}") await organization.store_async(synapse_client=syn) + # The cleanup list is reversed at teardown, so the organization must be + # scheduled before its schema: Synapse refuses to delete an organization that + # still owns a schema. + schedule_for_cleanup(organization) json_schema = JSONSchema( name="curation.compute.samplesheet", organization_name=organization.name @@ -1009,13 +1013,7 @@ async def sample_sheet_schema_uri( version="0.0.1", synapse_client=syn, ) - - def delete_organization() -> None: - for schema in organization.get_json_schemas(synapse_client=syn): - schema.delete(synapse_client=syn) - organization.delete(synapse_client=syn) - - request.addfinalizer(delete_organization) + schedule_for_cleanup(json_schema) return json_schema.uri @@ -1026,7 +1024,7 @@ async def destination_task( project_model: Project, schedule_for_cleanup: Callable[..., None], sample_sheet_schema_uri: str, - ) -> CurationTask: + ) -> AsyncGenerator[CurationTask, None]: """ Create the record-based CurationTask that a compute task writes its output to. @@ -1056,7 +1054,7 @@ async def destination_task( json_schema_uri=sample_sheet_schema_uri, synapse_client=syn ) - return await CurationTask( + yield await CurationTask( data_type=f"destination_{str(uuid.uuid4()).replace('-', '_')}", project_id=project_model.id, instructions="Receives the output of a compute task.", @@ -1065,6 +1063,11 @@ async def destination_task( ), ).store_async(synapse_client=syn) + # Synapse refuses to delete a schema that is still bound to an object, and the + # RecordSet is not deleted until session teardown. Unbind here, at function + # teardown, so the scheduled cleanup can delete the schema. + await record_set.unbind_schema_async(synapse_client=syn) + async def test_execute_record_set_generation_async( self, syn: Synapse, From 8120e4bdfedda7fcef63e85a14431b6a3cbcf89e Mon Sep 17 00:00:00 2001 From: Andrew Lamb Date: Thu, 30 Jul 2026 11:39:13 -0700 Subject: [PATCH 6/7] add unit test --- .../core/unit_test_transfer_bar.py | 24 +++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/tests/unit/synapseclient/core/unit_test_transfer_bar.py b/tests/unit/synapseclient/core/unit_test_transfer_bar.py index 1a9693426..9422e17bc 100644 --- a/tests/unit/synapseclient/core/unit_test_transfer_bar.py +++ b/tests/unit/synapseclient/core/unit_test_transfer_bar.py @@ -1,5 +1,7 @@ """Unit tests for the silent-aware progress bar factory in transfer_bar.""" +import io +import sys from unittest.mock import patch from synapseclient import Synapse @@ -49,6 +51,28 @@ def test_disabled_bar_accepts_mutation(self, syn: Synapse) -> None: # THEN no error is raised and nothing is rendered assert progress_bar.disable is True + def test_bar_is_usable_when_silent_is_none(self, syn: Synapse) -> None: + # GIVEN a client whose silent flag was never set to a boolean + # AND a non-tty stderr, which is what makes tqdm treat disable=None as + # "disable and skip the rest of __init__" + with ( + patch.object(syn, "silent", None), + patch.object(sys, "stderr", io.StringIO()), + ): + # WHEN a progress bar is created + progress_bar = create_progress_bar(total=None, desc="", synapse_client=syn) + + # THEN the bar is fully constructed rather than left in tqdm's + # partially-initialized disable=None state + assert progress_bar.disable is False + + # AND callers can mutate and drive it directly (as the live sites do) + progress_bar.desc = "processing" + progress_bar.total = 50 + progress_bar.update(25) + progress_bar.refresh() + progress_bar.close() + def test_shows_bar_when_no_client_available(self) -> None: # GIVEN no client is passed and none can be resolved with patch.object(Synapse, "get_client", side_effect=SynapseError("no client")): From 756f256c3b6c05225c77b7c78e5860e04fe29dc3 Mon Sep 17 00:00:00 2001 From: Andrew Lamb Date: Thu, 30 Jul 2026 11:48:21 -0700 Subject: [PATCH 7/7] improved function documentation --- synapseclient/models/mixins/asynchronous_job.py | 15 ++++++++++++--- 1 file changed, 12 insertions(+), 3 deletions(-) diff --git a/synapseclient/models/mixins/asynchronous_job.py b/synapseclient/models/mixins/asynchronous_job.py index 9990fcc58..c39ec7096 100644 --- a/synapseclient/models/mixins/asynchronous_job.py +++ b/synapseclient/models/mixins/asynchronous_job.py @@ -75,8 +75,12 @@ def _resolve_async_job_uri( Arguments: request_type: The concreteType of the asynchronous job request. - request: The request that was, or will be, sent to the server. Only required - when the URI for the request type contains placeholders. + request: The serialized job body that to_synapse_request() emits for this + request type, in the camelCase form that is sent to the server. Its keys + are what the URI placeholders are resolved against, so a URI containing + {taskId} requires a taskId key here. Only required when the URI for the + request type contains placeholders; request types with a static URI can + omit it. Returns: The asynchronous job URI with all placeholders resolved. @@ -88,8 +92,10 @@ def _resolve_async_job_uri( """ if not request_type or request_type not in ASYNC_JOB_URIS: raise ValueError(f"Unsupported request type: {request_type}") - uri = ASYNC_JOB_URIS[request_type] + + # The braced names in the URI, e.g. entityId for /entity/{entityId}/table/query/async. + # Each one names a key of the serialized request that supplies that path segment. placeholders = [ field_name for _, field_name, _, _ in Formatter().parse(uri) @@ -98,6 +104,9 @@ def _resolve_async_job_uri( if not placeholders: return uri + # Placeholders that could never name a request key, such as the positional {} or + # an indexed {0}. These are typos in ASYNC_JOB_URIS rather than caller mistakes, so + # they are caught before the request is looked at. unusable = [name for name in placeholders if not name.isidentifier()] if unusable: raise ValueError(