diff --git a/docs/guides/extensions/curator/metadata_contribution.md b/docs/guides/extensions/curator/metadata_contribution.md index 40b07d05e..f61346f2c 100644 --- a/docs/guides/extensions/curator/metadata_contribution.md +++ b/docs/guides/extensions/curator/metadata_contribution.md @@ -128,7 +128,7 @@ print(df) # Example only: fills 4 rows with random integers regardless of column type. # Replace this with real edits that match your task's schema before importing — -# schema validation runs in Step 6 and will reject values that don't fit. +# schema validation runs in Step 7 and will reject values that don't fit. df = pd.DataFrame( np.random.randint(0, 100, size=(4, len(df.columns))), columns=df.columns, @@ -147,7 +147,74 @@ latest_grid = latest_grid.import_csv(path=edited_path) print(f"Upserted edits into grid session: https://www.synapse.org/Grid:default?sessionId={latest_grid.session_id}") ``` -### Step 6: Export the grid back to the RecordSet +### Step 6: Validate your edits in-session (before exporting) + +Before you export, you can check your edits against the JSON schema bound to the RecordSet directly on the Grid session — without creating a new RecordSet version. This lets you catch and fix problems while iterating, then export once (Step 7) with clean data. + +Open the session with `connect()` (or `connect_async()`), which binds a replica for the duration of the `with` block, then call `validate_rows()` with a `QueryRequest`. Each returned row carries its own `validation_results`. `SelectAll()` returns every column, so each row's full data comes back alongside its validation results. + +Reuse the `latest_grid` session you've been editing: + +```python +from synapseclient.models.curation import GridQuery, QueryRequest, SelectAll + +with latest_grid.connect() as grid: + query_request = QueryRequest(query=GridQuery(column_selection=[SelectAll()])) + query_result = grid.validate_rows(query_request=query_request) + + if query_result.rows is None: + print("No rows matched the query.") + else: + for row in query_result.rows: + print(f"Row ID: {row.row_id}, Validation Result: {row.validation_results}") +``` + +If you're landing here with only a `record_set_id` (no session yet), `connect()` will create — or, with `.connect(attach_to_previous_session=True)`, reattach to — a session for you: + +```python +from synapseclient.models import Grid +from synapseclient.models.curation import GridQuery, QueryRequest, SelectAll + +with Grid(record_set_id="syn123456789").connect() as grid: + query_request = QueryRequest(query=GridQuery(column_selection=[SelectAll()])) + query_result = grid.validate_rows(query_request=query_request) + + for row in query_result.rows: + print(f"Row ID: {row.row_id}, Validation Result: {row.validation_results}") +``` + +To focus on just the rows that currently fail, add a `RowIsValidFilter` and set `include_validation_messages=True` so each row’s `validation_results` includes the full `all_validation_messages` list: + +```python +from synapseclient.models.curation import ( + GridQuery, + QueryRequest, + RowIsValidFilter, + SelectAll, +) + +with latest_grid.connect() as grid: + query_request = QueryRequest( + query=GridQuery( + column_selection=[SelectAll()], + filters=[RowIsValidFilter(value=False)], + include_validation_messages=True, + ) + ) + query_result = grid.validate_rows(query_request=query_request) + + for row in query_result.rows: + print(f"Invalid row {row.row_id}: {row.validation_results}") +``` + +!!! note "Requires a bound schema" + In-session validation only produces results when the administrator has bound a + JSON schema to the RecordSet. Without one, rows still return but their + `validation_results` is `None` for each row. + +In-session `validate_rows()` checks the current Grid rows without creating a new RecordSet version — use it to iterate quickly. The export report reviewed in Step 8 (`get_detailed_validation_results()`) reflects the last export from Step 7. Each method has a matching `_async` counterpart (`connect_async`, `validate_rows_async`) for async contexts. + +### Step 7: Export the grid back to the RecordSet > **Important:** Until you call `export_to_record_set()`, your edits live only inside the Grid session — they aren't visible on the RecordSet and won't be validated. Apply changes whenever you reach a logical checkpoint. @@ -158,9 +225,9 @@ latest_grid.export_to_record_set() print(f"Exported to RecordSet version: {latest_grid.record_set_version_number}") ``` -### Step 7: Review your validation results +### Step 8: Review your validation results -When you exported the grid in Step 6, Synapse validated each row against the JSON schema bound to the RecordSet and generated a row-level report. Reviewing this report before handing the task back to the administrator lets you catch and fix problems in your own data first — saving a round trip. +When you exported the grid in Step 7, Synapse validated each row against the JSON schema bound to the RecordSet and generated a row-level report. Reviewing this report before handing the task back to the administrator lets you catch and fix problems in your own data first — saving a round trip. (For a quick check before you commit an export, use the in-session validation in Step 6 instead.) #### Prerequisites for validation results @@ -170,7 +237,7 @@ A validation report is only generated when **all** of the following are true: 2. You have entered data through a Grid session 3. The Grid session has been exported back to the RecordSet — this is the step that triggers validation and populates the RecordSet's validation_file_handle_id -If the Grid was never exported (Step 6), there is nothing to review yet. +If the Grid was never exported (Step 7), there is nothing to review yet. #### Retrieve and inspect the results @@ -185,7 +252,7 @@ if isinstance(curation_task.task_properties, RecordBasedMetadataTaskProperties): validation_df = record_set.get_detailed_validation_results() if validation_df is None: - print("No validation results yet — make sure the Grid was exported in Step 6.") + print("No validation results yet — make sure the Grid was exported in Step 7.") else: total = len(validation_df) valid = validation_df["is_valid"].sum() @@ -230,11 +297,11 @@ Row 2: #### Fix and re-export -If any rows are invalid, recreate a Grid session against the RecordSet (see Step 3), correct the offending rows, and re-run Steps 4–6 to re-export. The validation report is regenerated on each export, so iterate until the report is clean before letting the administrator know your task is ready. +If any rows are invalid, recreate a Grid session against the RecordSet (see Step 3), correct the offending rows, and re-run Steps 4–7 to re-export. The validation report is regenerated on each export, so iterate until the report is clean before letting the administrator know your task is ready. > **If get_detailed_validation_results returns None after exporting:** check that record_set.validation_file_handle_id is set after the re-fetch. If it isn't, the export did not complete — re-run export_to_record_set() on an active Grid session against the same RecordSet. -### Step 8: Mark the curation task as COMPLETED +### Step 9: Mark the curation task as COMPLETED Once your validation report is clean and you've cleaned up the Grid session, transition the curation task to COMPLETED. This signals the administrator that the task is ready for their review — they can list tasks in the project and pick up the ones whose status is COMPLETED. @@ -244,7 +311,7 @@ curation_task.set_task_state(state="COMPLETED") ## File-Based Curation Tasks -File-based tasks follow the same overall flow as record-based tasks (Steps 1–8 above), with three key differences: +File-based tasks follow the same overall flow as record-based tasks (Steps 1–9 above), with three key differences: **No CSV import.** `import_csv` is not currently supported for file-based grids. Instead, you can either: @@ -259,7 +326,20 @@ latest_grid.synchronize() This writes the Grid annotation values back to each file as Synapse annotations. There is no versioned RecordSet — the files themselves are updated in place. -**No per-row validation report.** Validation is enforced by the JSON schema bound to the folder containing the files, not by a row-level export report. After you call `synchronize()`, the administrator verifies schema compliance on their end — there is nothing to retrieve from the contributor side. If the administrator reports violations, correct the flagged annotations in the Grid UI and re-synchronize. +**No per-row export report — but in-session validation still works.** There is no versioned RecordSet, so the export report reviewed in Step 8 (`export_to_record_set()` → `get_detailed_validation_results()`) does not apply. However, the in-session validation from Step 6 works identically for file-based grids: a file-based session created from an `initial_query` still carries a bound JSON schema (`grid_json_schema_id`), so `connect()` + `validate_rows()` returns the same per-row `validation_results`. This is your primary contributor-side check for file-based tasks — run it before `synchronize()`. + +```python +from synapseclient.models.curation import GridQuery, QueryRequest, SelectAll + +with latest_grid.connect() as grid: + query_request = QueryRequest(query=GridQuery(column_selection=[SelectAll()])) + query_result = grid.validate_rows(query_request=query_request) + + for row in query_result.rows: + print(f"Row ID: {row.row_id}, Validation Result: {row.validation_results}") +``` + +After you call `synchronize()`, the administrator also verifies schema compliance on their end. If they report violations, correct the flagged annotations in the Grid UI and re-synchronize. ## Appendix @@ -280,6 +360,8 @@ Deleting is permanent — you can no longer re-export from this session. If you - [CurationTask.get][synapseclient.models.CurationTask.get] - Fetch a CurationTask by id - [CurationTask.create_grid_session][synapseclient.models.CurationTask.create_grid_session] - Create a Grid session for a CurationTask and link it to the task status - [CurationTask.set_task_state][synapseclient.models.CurationTask.set_task_state] - Set the state on a CurationTask's status +- [Grid.connect][synapseclient.models.Grid.connect] - Connect to a Grid session and bind a replica for in-session validation +- [Grid.validate_rows][synapseclient.models.Grid.validate_rows] - Validate a Grid session's rows against the bound JSON schema - [Grid.download_csv][synapseclient.models.Grid.download_csv] - Download Grid contents as a local CSV - [Grid.import_csv][synapseclient.models.Grid.import_csv] - Upsert CSV edits back into a Grid session (record-based grids only) - [Grid.export_to_record_set][synapseclient.models.Grid.export_to_record_set] - Export Grid data back to RecordSet and generate validation results diff --git a/docs/reference/experimental/async/curator.md b/docs/reference/experimental/async/curator.md index 39e62f705..7c864e7fb 100644 --- a/docs/reference/experimental/async/curator.md +++ b/docs/reference/experimental/async/curator.md @@ -68,6 +68,8 @@ - import_csv_async - delete_async - list_async + - connect_async + - validate_rows_async --- [](){ #query-reference-async } ::: synapseclient.models.Query @@ -75,3 +77,117 @@ inherited_members: true members: --- +[](){ #GridQuery-reference-async } +::: synapseclient.models.curation.GridQuery + options: + inherited_members: true + members: +--- +[](){ #QueryRequest-reference-async } +::: synapseclient.models.curation.QueryRequest + options: + inherited_members: true + members: +--- +[](){ #SelectItem-reference-async } +::: synapseclient.models.curation.SelectItem + options: + inherited_members: true + members: +--- +[](){ #SelectByName-reference-async } +::: synapseclient.models.curation.SelectByName + options: + inherited_members: true + members: +--- +[](){ #SelectAll-reference-async } +::: synapseclient.models.curation.SelectAll + options: + inherited_members: true + members: +--- +[](){ #CountStar-reference-async } +::: synapseclient.models.curation.CountStar + options: + inherited_members: true + members: +--- +[](){ #SelectSelection-reference-async } +::: synapseclient.models.curation.SelectSelection + options: + inherited_members: true + members: +--- +[](){ #Filter-reference-async } +::: synapseclient.models.curation.Filter + options: + inherited_members: true + members: +--- +[](){ #RowValidationResultFilter-reference-async } +::: synapseclient.models.curation.RowValidationResultFilter + options: + inherited_members: true + members: +--- +[](){ #CellValueFilter-reference-async } +::: synapseclient.models.curation.CellValueFilter + options: + inherited_members: true + members: +--- +[](){ #RowSelectionFilter-reference-async } +::: synapseclient.models.curation.RowSelectionFilter + options: + inherited_members: true + members: +--- +[](){ #RowIsValidFilter-reference-async } +::: synapseclient.models.curation.RowIsValidFilter + options: + inherited_members: true + members: +--- +[](){ #RowIdFilter-reference-async } +::: synapseclient.models.curation.RowIdFilter + options: + inherited_members: true + members: +--- +[](){ #ValidationOperator-reference-async } +::: synapseclient.models.curation.ValidationOperator + options: + inherited_members: true + members: +--- +[](){ #CellValueOperator-reference-async } +::: synapseclient.models.curation.CellValueOperator + options: + inherited_members: true + members: +--- +[](){ #GridQueryResult-reference-async } +::: synapseclient.models.curation.GridQueryResult + options: + inherited_members: true + members: +--- +[](){ #GridRow-reference-async } +::: synapseclient.models.curation.GridRow + options: + inherited_members: true + members: +--- +[](){ #SelectColumn-reference-async } +::: synapseclient.models.curation.SelectColumn + options: + inherited_members: true + members: +--- +[](){ #GridQueryValidationResult-reference-async } +::: synapseclient.models.curation.GridQueryValidationResult + options: + inherited_members: true + members: +--- diff --git a/docs/reference/experimental/sync/curator.md b/docs/reference/experimental/sync/curator.md index b330fc3ec..36a6dca60 100644 --- a/docs/reference/experimental/sync/curator.md +++ b/docs/reference/experimental/sync/curator.md @@ -68,6 +68,8 @@ - import_csv - delete - list + - connect + - validate_rows --- [](){ #query-reference } ::: synapseclient.models.Query @@ -75,3 +77,117 @@ inherited_members: true members: --- +[](){ #GridQuery-reference } +::: synapseclient.models.curation.GridQuery + options: + inherited_members: true + members: +--- +[](){ #QueryRequest-reference } +::: synapseclient.models.curation.QueryRequest + options: + inherited_members: true + members: +--- +[](){ #SelectItem-reference } +::: synapseclient.models.curation.SelectItem + options: + inherited_members: true + members: +--- +[](){ #SelectByName-reference } +::: synapseclient.models.curation.SelectByName + options: + inherited_members: true + members: +--- +[](){ #SelectAll-reference } +::: synapseclient.models.curation.SelectAll + options: + inherited_members: true + members: +--- +[](){ #CountStar-reference } +::: synapseclient.models.curation.CountStar + options: + inherited_members: true + members: +--- +[](){ #SelectSelection-reference } +::: synapseclient.models.curation.SelectSelection + options: + inherited_members: true + members: +--- +[](){ #Filter-reference } +::: synapseclient.models.curation.Filter + options: + inherited_members: true + members: +--- +[](){ #RowValidationResultFilter-reference } +::: synapseclient.models.curation.RowValidationResultFilter + options: + inherited_members: true + members: +--- +[](){ #CellValueFilter-reference } +::: synapseclient.models.curation.CellValueFilter + options: + inherited_members: true + members: +--- +[](){ #RowSelectionFilter-reference } +::: synapseclient.models.curation.RowSelectionFilter + options: + inherited_members: true + members: +--- +[](){ #RowIsValidFilter-reference } +::: synapseclient.models.curation.RowIsValidFilter + options: + inherited_members: true + members: +--- +[](){ #RowIdFilter-reference } +::: synapseclient.models.curation.RowIdFilter + options: + inherited_members: true + members: +--- +[](){ #ValidationOperator-reference } +::: synapseclient.models.curation.ValidationOperator + options: + inherited_members: true + members: +--- +[](){ #CellValueOperator-reference } +::: synapseclient.models.curation.CellValueOperator + options: + inherited_members: true + members: +--- +[](){ #GridQueryResult-reference } +::: synapseclient.models.curation.GridQueryResult + options: + inherited_members: true + members: +--- +[](){ #GridRow-reference } +::: synapseclient.models.curation.GridRow + options: + inherited_members: true + members: +--- +[](){ #SelectColumn-reference } +::: synapseclient.models.curation.SelectColumn + options: + inherited_members: true + members: +--- +[](){ #GridQueryValidationResult-reference } +::: synapseclient.models.curation.GridQueryValidationResult + options: + inherited_members: true + members: +--- diff --git a/synapseclient/api/__init__.py b/synapseclient/api/__init__.py index c55e43410..4e13bc94a 100644 --- a/synapseclient/api/__init__.py +++ b/synapseclient/api/__init__.py @@ -18,6 +18,7 @@ ) from .curation_services import ( create_curation_task, + create_grid_replica, delete_curation_task, delete_grid_session, get_curation_task, @@ -339,6 +340,7 @@ "list_grid_sessions", "update_curation_task", "update_curation_task_status", + "create_grid_replica", # download_list_services "add_to_download_list_async", "clear_download_list_async", diff --git a/synapseclient/api/curation_services.py b/synapseclient/api/curation_services.py index 3f13e99f5..4b16d1eb1 100644 --- a/synapseclient/api/curation_services.py +++ b/synapseclient/api/curation_services.py @@ -284,3 +284,38 @@ async def delete_grid_session( client = Synapse.get_client(synapse_client=synapse_client) await client.rest_delete_async(uri=f"/grid/session/{session_id}") + + +async def create_grid_replica( + create_replica_request: Dict[str, Any], + session_id: str, + *, + synapse_client: Optional["Synapse"] = None, +) -> dict[str, Any]: + """ + Create a new replica for a grid session. A replica represents an + 'in-memory' grid document identified by a unique replicaId. + + https://rest-docs.synapse.org/rest/POST/grid/session/sessionId/replica.html + + Note: Only the user that started the grid session may create a replica. + A user is limited to 10 replicas per-hour per-grid-session. + + Arguments: + create_replica_request: The complete CreateReplicaRequest object. + session_id: The grid session ID. + 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 CreateReplicaResponse containing information about the created replica. + """ + from synapseclient import Synapse + + client = Synapse.get_client(synapse_client=synapse_client) + + return await client.rest_post_async( + uri=f"/grid/session/{session_id}/replica", + body=json.dumps(create_replica_request), + ) diff --git a/synapseclient/core/constants/concrete_types.py b/synapseclient/core/constants/concrete_types.py index d8cbdbd59..38ec38628 100644 --- a/synapseclient/core/constants/concrete_types.py +++ b/synapseclient/core/constants/concrete_types.py @@ -157,6 +157,23 @@ ) SYNCHRONIZE_GRID_REQUEST = "org.sagebionetworks.repo.model.grid.SynchronizeGridRequest" GRID_CSV_IMPORT_REQUEST = "org.sagebionetworks.repo.model.grid.GridCsvImportRequest" +GRID_QUERY_JOB_REQUEST = "org.sagebionetworks.repo.model.grid.GridQueryJobRequest" + +# Grid Query SelectItem Types +SELECT_BY_NAME = "org.sagebionetworks.repo.model.grid.query.SelectByName" +SELECT_ALL = "org.sagebionetworks.repo.model.grid.query.SelectAll" +COUNT_STAR = "org.sagebionetworks.repo.model.grid.query.function.CountStar" +SELECT_SELECTION = "org.sagebionetworks.repo.model.grid.query.SelectSelection" + +# Grid Query Filter Types +ROW_VALIDATION_RESULT_FILTER = ( + "org.sagebionetworks.repo.model.grid.query.RowValidationResultFilter" +) +CELL_VALUE_FILTER = "org.sagebionetworks.repo.model.grid.query.CellValueFilter" +ROW_SELECTION_FILTER = "org.sagebionetworks.repo.model.grid.query.RowSelectionFilter" +ROW_IS_VALID_FILTER = "org.sagebionetworks.repo.model.grid.query.RowIsValidFilter" +ROW_ID_FILTER = "org.sagebionetworks.repo.model.grid.query.RowIdFilter" + UPLOAD_TO_TABLE_PREVIEW_REQUEST = ( "org.sagebionetworks.repo.model.table.UploadToTablePreviewRequest" ) diff --git a/synapseclient/models/curation.py b/synapseclient/models/curation.py index a38c78229..0216c9413 100644 --- a/synapseclient/models/curation.py +++ b/synapseclient/models/curation.py @@ -8,6 +8,7 @@ import asyncio import os from abc import ABC, abstractmethod +from contextlib import asynccontextmanager, contextmanager from copy import deepcopy from dataclasses import dataclass, field, replace from datetime import datetime, timezone @@ -28,6 +29,7 @@ from synapseclient import Synapse from synapseclient.api import ( create_curation_task, + create_grid_replica, delete_curation_task, delete_grid_session, get_curation_task, @@ -46,15 +48,25 @@ wrap_async_generator_to_sync_generator, ) from synapseclient.core.constants.concrete_types import ( + CELL_VALUE_FILTER, + COUNT_STAR, CREATE_GRID_REQUEST, DOWNLOAD_FROM_GRID_REQUEST, FILE_BASED_METADATA_TASK_PROPERTIES, GRID_CSV_IMPORT_REQUEST, GRID_EXECUTION_DETAILS, + GRID_QUERY_JOB_REQUEST, GRID_RECORD_SET_EXPORT_REQUEST, LIST_GRID_SESSIONS_REQUEST, LIST_GRID_SESSIONS_RESPONSE, RECORD_BASED_METADATA_TASK_PROPERTIES, + ROW_ID_FILTER, + ROW_IS_VALID_FILTER, + ROW_SELECTION_FILTER, + ROW_VALIDATION_RESULT_FILTER, + SELECT_ALL, + SELECT_BY_NAME, + SELECT_SELECTION, SYNCHRONIZE_GRID_REQUEST, UPLOAD_TO_TABLE_PREVIEW_REQUEST, ) @@ -2280,6 +2292,755 @@ def to_synapse_request(self) -> Dict[str, Any]: return request_dict +@dataclass +class SelectColumn: + """ + Information about a selected column in a grid query result. + + Represents a [Synapse SelectColumn](https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/grid/query/result/SelectColumn.html). + + Note: This is distinct from the table + [SelectColumn](https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/table/SelectColumn.html) + (`org.sagebionetworks.repo.model.table.SelectColumn`), which additionally carries + `id` and `columnType`. This grid-query `SelectColumn` only has `column_name`. + + Attributes: + column_name: The name of the column. Will be the alias if one is + provided in the select item. + """ + + column_name: Optional[str] = None + """The name of the column. Will be the alias if one is provided in the select item.""" + + def fill_from_dict(self, synapse_response: Dict[str, Any]) -> "SelectColumn": + """ + Converts a response from the REST API into this dataclass. + + Arguments: + synapse_response: The response from the REST API. + + Returns: + The SelectColumn object. + """ + self.column_name = synapse_response.get("columnName", None) + return self + + +@dataclass +class GridQueryValidationResult: + """ + Results of a grid row against a JSON schema + + Represents a [Synapse ValidationResults](https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/grid/query/result/ValidationResults.html). + + Note: This is distinct from the general-purpose + [ValidationResults](https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/schema/ValidationResults.html) + (`org.sagebionetworks.repo.model.schema.ValidationResults`), which is used + for validating entities/containers against a schema and carries additional + fields (`objectId`, `objectType`, `objectEtag`, `schema$id`, `validatedOn`, + `validationException`) that this grid-row-specific type does not have. + + Attributes: + is_valid: Will be 'true' if the row is valid according to the JSON + schema. Will be 'false' when the row is invalid. + validation_error_message: If the object is not valid according to the + schema, a simple one line error message will be provided. + all_validation_messages: If the object is not valid according to the + schema, a the flat list of error messages will be provided with one + error message per sub-schema. Included only if + includeValidationMessages was set to true in the query. Otherwise, + this array is omitted to optimize performance. + """ + + is_valid: Optional[bool] = None + """Will be 'true' if the row is valid according to the JSON schema. Will be + 'false' when the row is invalid.""" + + validation_error_message: Optional[str] = None + """If the object is not valid according to the schema, a simple one line + error message will be provided.""" + + all_validation_messages: Optional[list[str]] = None + """If the object is not valid according to the schema, a the flat list of + error messages will be provided with one error message per sub-schema. + Included only if includeValidationMessages was set to true in the query. + Otherwise, this array is omitted to optimize performance.""" + + def fill_from_dict( + self, synapse_response: Dict[str, Any] + ) -> "GridQueryValidationResult": + """ + Converts a response from the REST API into this dataclass. + + Arguments: + synapse_response: The response from the REST API. + + Returns: + The GridQueryValidationResult object. + """ + self.is_valid = synapse_response.get("isValid", None) + self.validation_error_message = synapse_response.get( + "validationErrorMessage", None + ) + self.all_validation_messages = synapse_response.get( + "allValidationMessages", None + ) + return self + + +@dataclass +class GridRow: + """ + A single row of a grid query result. + + Represents a [Synapse Row](https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/grid/query/result/Row.html). + + Note: This is distinct from the SQL-based table + [Row](https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/table/Row.html) + (`org.sagebionetworks.repo.model.table.Row`), which represents a row of a + table/view query result rather than a grid query result. + + Attributes: + row_id: Logical timestamp identifying the row in compact form + `replicaId.sequenceNumber` (e.g. `123.456`). Used in filtering + operations and update/patch procedures. + data: The JSON object representing a single row. + validation_results: Results of validating this row against a JSON + schema, if a schema is bound. + """ + + row_id: Optional[str] = None + """Logical timestamp identifying the row in compact form `replicaId.sequenceNumber`.""" + + data: Optional[Dict[str, Any]] = None + """The JSON object representing a single row.""" + + validation_results: Optional[GridQueryValidationResult] = None + """Results of validating this row against a JSON schema, if a schema is bound.""" + + def fill_from_dict(self, synapse_response: Dict[str, Any]) -> "GridRow": + """ + Converts a response from the REST API into this dataclass. + + Arguments: + synapse_response: The response from the REST API. + + Returns: + The GridRow object. + """ + self.row_id = synapse_response.get("rowId", None) + self.data = synapse_response.get("data", None) + validation_results_data = synapse_response.get("validationResults", None) + self.validation_results = ( + GridQueryValidationResult().fill_from_dict(validation_results_data) + if validation_results_data is not None + else None + ) + return self + + +@dataclass +class GridQueryResult: + """ + A single page of rows returned from a grid query. + + Represents a [Synapse QueryResult](https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/grid/query/result/QueryResult.html). + + Note: This is distinct from the SQL-based table + [QueryResult](https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/table/QueryResult.html) + (`org.sagebionetworks.repo.model.table.QueryResult`), which wraps SQL query + results rather than a grid query's SelectColumn/Row objects. + + Attributes: + select_columns: Information about the selected columns. + rows: A single page of rows. + """ + + select_columns: Optional[list[SelectColumn]] = None + """Information about the selected columns.""" + + rows: Optional[list[GridRow]] = None + """A single page of rows.""" + + def fill_from_dict(self, synapse_response: Dict[str, Any]) -> "GridQueryResult": + """ + Converts a response from the REST API into this dataclass. + + Arguments: + synapse_response: The response from the REST API. + + Returns: + The GridQueryResult object. + """ + select_columns_data = synapse_response.get("selectColumns", None) + self.select_columns = ( + [SelectColumn().fill_from_dict(col) for col in select_columns_data] + if select_columns_data is not None + else None + ) + + rows_data = synapse_response.get("rows", None) + self.rows = ( + [GridRow().fill_from_dict(row) for row in rows_data] + if rows_data is not None + else None + ) + return self + + +class ValidationOperator(str, Enum): + """ + The comparison operator. + + See . + """ + + LIKE = "LIKE" + NOT_LIKE = "NOT_LIKE" + + +class CellValueOperator(str, Enum): + """ + The comparison operator. + + See . + """ + + EQUALS = "EQUALS" + NOT_EQUALS = "NOT_EQUALS" + GREATER_THAN = "GREATER_THAN" + LESS_THAN = "LESS_THAN" + GREATER_THAN_OR_EQUALS = "GREATER_THAN_OR_EQUALS" + LESS_THAN_OR_EQUALS = "LESS_THAN_OR_EQUALS" + IN = "IN" + NOT_IN = "NOT_IN" + LIKE = "LIKE" + NOT_LIKE = "NOT_LIKE" + IS_NULL = "IS_NULL" + IS_NOT_NULL = "IS_NOT_NULL" + IS_UNDEFINED = "IS_UNDEFINED" + IS_DEFINED = "IS_DEFINED" + + +@dataclass +class SelectItem(ABC): + """ + A generic select item. + + + + Known implementations: SelectByName, SelectAll, CountStar, SelectSelection. + + The concrete subclass is determined by the concreteType field in the REST response. + """ + + @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 SelectByName(SelectItem): + """ + A SelectItem that will result in the selection of a single column by its name. + + + + Attributes: + column_name: The name of the column to include in the select. + """ + + column_name: Optional[str] = None + """The name of the column to include in the select.""" + + 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": SELECT_BY_NAME, "columnName": self.column_name} + delete_none_keys(request_dict) + return request_dict + + +@dataclass +class SelectAll(SelectItem): + """ + A SelectItem that will result in the selection of all columns. + + + """ + + 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. + """ + return {"concreteType": SELECT_ALL} + + +@dataclass +class CountStar(SelectItem): + """ + Use this to count the total number of rows that match the query. For example, + for a user request like 'how many rows are there in total?', select this item. + The alias property can be used to name the resulting count column. + + + + Attributes: + alias: Used to name the resulting count column. + """ + + alias: Optional[str] = None + """Used to name the resulting count column.""" + + 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": COUNT_STAR, "alias": self.alias} + delete_none_keys(request_dict) + return request_dict + + +@dataclass +class SelectSelection(SelectItem): + """ + A SelectItem that will result in the selection of the columns the user has + actively selected in the interface. + + + """ + + 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. + """ + return {"concreteType": SELECT_SELECTION} + + +@dataclass +class Filter(ABC): + """ + There are five different types of filters that can be applied to a grid query. + + + + Known implementations: RowValidationResultFilter, CellValueFilter, + RowSelectionFilter, RowIsValidFilter, RowIdFilter. + + The concrete subclass is determined by the concreteType field in the REST response. + """ + + @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 RowValidationResultFilter(Filter, EnumCoercionMixin): + """ + Use this filter to find rows that have data quality issues or schema + validation errors. + + + + To find type errors, use the LIKE operator with '%expected type:%' as the + validation result value. + + Attributes: + operator: The comparison operator. + validation_result_value: A validation result value. For wildcards use + '%' to represents zero or more characters, and '_' to represents a + single character. + """ + + _ENUM_FIELDS: ClassVar[dict[str, type]] = {"operator": ValidationOperator} + + operator: Optional[Union[ValidationOperator, str]] = None + """The comparison operator.""" + + validation_result_value: Optional[str] = None + """A validation result value. For wildcards use '%' to represents zero or + more characters, and '_' to represents a single character.""" + + 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": ROW_VALIDATION_RESULT_FILTER, + "operator": self.operator.value if self.operator is not None else None, + "validationResultValue": self.validation_result_value, + } + delete_none_keys(request_dict) + return request_dict + + +@dataclass +class CellValueFilter(Filter, EnumCoercionMixin): + """ + A filter used to select rows based on cell values. For example, to handle a + user request like 'find all rows where the Project column is Alpha', you + would set 'columnName' to 'Project', 'operator' to 'EQUALS', and 'value' to + ['Alpha']. + + + + Attributes: + column_name: The name of the column to filter by. + operator: The comparison operator. + value: Use operators like 'EQUALS' or 'LIKE' with the 'value' property + for standard comparisons. The 'IS_NULL' operator can be used to + find null values. The 'IS_UNDEFINED' operator can be used to find + undefined values. When using IN or NOT_IN operators, value should + be an array of values to compare against. When using either 'LIKE' + or 'NOT_LIKE', the wildcard character '%' is used to represents + zero or more characters, and '_' is used to represent a single + character. + """ + + _ENUM_FIELDS: ClassVar[dict[str, type]] = {"operator": CellValueOperator} + + column_name: Optional[str] = None + """The name of the column to filter by.""" + + operator: Optional[Union[CellValueOperator, str]] = None + """The comparison operator.""" + + value: Optional[Any] = None + """Use operators like 'EQUALS' or 'LIKE' with the 'value' property for + standard comparisons. The 'IS_NULL' operator can be used to find null + values. The 'IS_UNDEFINED' operator can be used to find undefined values. + When using IN or NOT_IN operators, value should be an array of values to + compare against. When using either 'LIKE' or 'NOT_LIKE', the wildcard + character '%' is used to represents zero or more characters, and '_' is + used to represent a single character.""" + + 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": CELL_VALUE_FILTER, + "columnName": self.column_name, + "operator": self.operator.value if self.operator is not None else None, + "value": self.value, + } + delete_none_keys(request_dict) + return request_dict + + +@dataclass +class RowSelectionFilter(Filter): + """ + Use this filter to narrow down results based on rows the user has actively + selected in the interface. For user requests like 'show me only my selected + items' or 'run this analysis on the rows I've checked', set the + 'isSelected' property to true. + + + + Attributes: + is_selected: When true, only rows that the user has selected will be + returned. When false, rows that the user has selected will be + excluded. + """ + + is_selected: Optional[bool] = None + """When true, only rows that the user has selected will be returned. When + false, rows that the user has selected will be excluded.""" + + 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": ROW_SELECTION_FILTER, + "isSelected": self.is_selected, + } + delete_none_keys(request_dict) + return request_dict + + +@dataclass +class RowIsValidFilter(Filter): + """ + Use this filter for simple requests to find all 'valid' or 'invalid' rows + based on their overall validation status. + + + + Attributes: + value: Set to true to find rows that are valid according to the + schema. Set to false to find rows that are invalid and have + validation errors. + """ + + value: Optional[bool] = None + """Set to true to find rows that are valid according to the schema. Set to + false to find rows that are invalid and have validation errors.""" + + 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": ROW_IS_VALID_FILTER, "value": self.value} + delete_none_keys(request_dict) + return request_dict + + +@dataclass +class RowIdFilter(Filter): + """ + Row ID inclusion filter. Use when you need to operate on specific existing + rows by their explicit row IDs obtained from a prior grid query (e.g., an + update). The filter matches any row whose ID is in the provided list + (logical OR semantics). Do not use for pattern matching or broad selection; + supply only the exact row IDs you intend to modify or retrieve. + + + + Attributes: + row_ids_in: Array of explicit row IDs. The result will include any row + whose ID appears in this list (logical OR). Provide only IDs + previously obtained from a grid query. Omit this filter if you do + not know the IDs. Do not include duplicates or IDs not present in + the current grid. + """ + + row_ids_in: Optional[list[str]] = None + """Array of explicit row IDs. The result will include any row whose ID + appears in this list (logical OR). Provide only IDs previously obtained + from a grid query. Omit this filter if you do not know the IDs. Do not + include duplicates or IDs not present in the current grid.""" + + 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": ROW_ID_FILTER, "rowIdsIn": self.row_ids_in} + delete_none_keys(request_dict) + return request_dict + + +@dataclass +class GridQuery: + """ + A structured grid query, expressed with JSON SelectItem and Filter objects + rather than SQL syntax. + + Represents a [Synapse Query](https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/grid/query/Query.html). + + Note: This is distinct from the SQL-based table + [Query](https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/table/Query.html) + (`org.sagebionetworks.repo.model.table.Query`, imported here as `Query`), which + takes a SQL string rather than structured select items and filters. + + Attributes: + column_selection: One or more SelectItem is required to define the + columns that will be returned by this query (e.g. SelectAll, + CountStar). + filters: Each filter must be a complete JSON object with the required + 'concreteType' property. Multiple filters are combined with AND logic. + limit: Limit of the number of rows returned to avoid loading more data + than needed into your context window. + offset: Specifies where the first returned row begins in the result set. + include_validation_messages: Controls whether the + 'allValidationMessages' array appears in the response. Defaults to + false to conserve token usage. + """ + + column_selection: list[SelectItem] = field(default_factory=list) + """One or more SelectItem is required to define the columns that will be + returned by this query.""" + + limit: Optional[int] = None + """Limit of the number of rows returned to avoid loading more data than + needed into your context window.""" + + filters: Optional[list[Filter]] = None + """Each filter must be a complete JSON object with the required + 'concreteType' property. Multiple filters are combined with AND logic.""" + + offset: Optional[int] = None + """Specifies where the first returned row begins in the result set.""" + + include_validation_messages: Optional[bool] = None + """Controls whether the 'allValidationMessages' array appears in the response.""" + + 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. + + Raises: + ValueError: If column_selection is empty. + """ + if not self.column_selection: + raise ValueError( + "column_selection is required and must contain at least one " + "SelectItem." + ) + + request_dict = { + "columnSelection": [ + item.to_synapse_request() for item in self.column_selection + ], + "filters": ( + [item.to_synapse_request() for item in self.filters] + if self.filters is not None + else None + ), + "limit": self.limit, + "offset": self.offset, + "includeValidationMessages": self.include_validation_messages, + } + delete_none_keys(request_dict) + return request_dict + + +@dataclass +class QueryRequest: + """ + Request to run a query. + + Represents a [Synapse QueryRequest](https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/grid/query/QueryRequest.html). + + Attributes: + query: Defines a structured query using JSON SelectItems and Filters objects - NOT SQL syntax. + """ + + query: Optional[GridQuery] = None + """Defines a structured query using JSON SelectItems and Filters objects (not SQL syntax).""" + + 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 = { + "query": self.query.to_synapse_request() if self.query else None, + } + delete_none_keys(request_dict) + return request_dict + + +@dataclass +class GridQueryJobRequest(AsynchronousCommunicator): + """ + Asynchronous job request to query a grid session and return a single page of + rows with optional per-row validation data. + + This request is modeled from: + + The response is modeled from: + """ + + session_id: str + """The grid session ID for querying.""" + + replica_id: int + """The caller's replica ID. Used for row-selection filtering and must be owned by the caller.""" + + query_request: QueryRequest = field(default_factory=QueryRequest) + """Request to run a query.""" + + concrete_type: str = GRID_QUERY_JOB_REQUEST + """The concrete type for this request.""" + + # Response fields (populated by fill_from_dict) + query_result: Optional[GridQueryResult] = field(default=None, compare=False) + """Results of a query against a grid session.""" + + def fill_from_dict(self, synapse_response: Dict[str, Any]) -> "GridQueryJobRequest": + """ + Converts a response from the REST API into this dataclass. + + Arguments: + synapse_response: The response from the REST API. + + Returns: + The GridQueryJobRequest object. + """ + query_result_data = synapse_response.get("queryResult", None) + self.query_result = ( + GridQueryResult().fill_from_dict(query_result_data) + if query_result_data 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. + + Raises: + ValueError: If query_request or its nested query is not set, since + Synapse would reject the request regardless. + """ + if not self.query_request or not self.query_request.query: + raise ValueError( + "query_request.query is required to query the grid. " + "Set query_request=QueryRequest(query=GridQuery(...)) before " + "sending this request." + ) + + request_dict = { + "concreteType": self.concrete_type, + "sessionId": self.session_id, + "replicaId": self.replica_id, + "queryRequest": self.query_request.to_synapse_request(), + } + delete_none_keys(request_dict) + return request_dict + + @dataclass class DownloadFromGridRequest(AsynchronousCommunicator): """ @@ -2719,6 +3480,84 @@ def fill_from_dict( return self +@dataclass +class GridReplica: + """ + Information about a replica. + + Represents a [Synapse GridReplica](https://rest-docs.synapse.org/rest/org/sagebionetworks/repo/model/grid/GridReplica.html). + + Attributes: + grid_session_id: The ID of the grid session. + replica_id: The unique identifier for the new replica. + created_by: The user that created this replica. + is_agent_replica: When true, this replica belongs to the createdBy + user's agent. + created_on: The date-time when the user created this replica. + """ + + grid_session_id: Optional[str] = None + """The ID of the grid session.""" + + replica_id: Optional[int] = None + """The unique identifier for the new replica.""" + + created_by: Optional[str] = None + """The user that created this replica.""" + + is_agent_replica: Optional[bool] = None + """When true, this replica belongs to the createdBy user's agent.""" + + created_on: Optional[str] = None + """The date-time when the user created this replica.""" + + def fill_from_dict(self, synapse_response: Dict[str, Any]) -> "GridReplica": + """ + Converts a response from the REST API into this dataclass. + + Arguments: + synapse_response: The response from the REST API. + + Returns: + The GridReplica object. + """ + self.grid_session_id = synapse_response.get("gridSessionId", None) + self.replica_id = synapse_response.get("replicaId", None) + self.created_by = synapse_response.get("createdBy", None) + self.is_agent_replica = synapse_response.get("isAgentReplica", None) + self.created_on = synapse_response.get("createdOn", None) + return self + + +@dataclass +class CreateReplicaRequest: + """ + Request to create a new replica. A replica represents an 'in-memory' grid + document identified by a unique replicaId. + + This request is modeled from: + + Attributes: + grid_session_id: The ID of the grid session. + replica: Information about a replica. Populated after calling the + create-replica endpoint. + """ + + grid_session_id: Optional[str] = None + """The ID of the grid session.""" + + 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 = {"gridSessionId": self.grid_session_id} + delete_none_keys(request_dict) + return request_dict + + class GridSynchronousProtocol(Protocol): """ The protocol for methods that are asynchronous but also @@ -2843,33 +3682,166 @@ def synchronize( `Synapse.allow_client_caching(False)` this will use the last created instance from the Synapse class constructor. - Returns: - Grid: The Grid object. + Returns: + Grid: The Grid object. + + Raises: + ValueError: If session_id is not provided. + + Example: Synchronize a grid session created from a file view +   + + ```python + from synapseclient import Synapse + from synapseclient.models import Grid + from synapseclient.models.table_components import Query + + syn = Synapse() + syn.login() + + # First create a grid session from a file view + query = Query(sql="SELECT * FROM syn1234567") + grid = Grid(initial_query=query) + grid = grid.create() + + # Synchronize the grid with the latest state of the file view + grid = grid.synchronize() + ``` + """ + return self + + def _create_replica( + self, *, synapse_client: Optional[Synapse] = None + ) -> "GridReplica": + """ + Creates a new replica for this grid session. + + A grid replica is an in-memory document that represents a 'copy' of the + grid. Each replica is identified by a unique replicaId, issued by the + 'hub'. A user can have more than one replica at a time (i.e. using + multiple browser tabs/machines). A user is limited to 10 replicas + per-hour per-grid-session. Only the user that started the grid session + may create a replica. + + Arguments: + 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 newly created GridReplica. + + Raises: + ValueError: If session_id is not provided, or if the Synapse response + did not contain replica information. + + Example: Create a replica for a grid session +   + + ```python + from synapseclient import Synapse + from synapseclient.models import Grid + + syn = Synapse() + syn.login() + + grid = Grid(record_set_id="syn1234567") + grid = grid.create() + + replica = grid._create_replica() + print(f"Replica created with ID: {replica.replica_id}") + ``` + """ + return None + + def validate_rows( + self, + *, + timeout: int = 120, + query_request: "QueryRequest", + synapse_client: Optional[Synapse] = None, + ) -> "GridQueryResult": + """ + Queries this grid session's rows and returns their per-row validation + results against the grid's bound JSON schema. + + This Grid must have been obtained from `connect_async` (or `connect`), + which binds a replica to it. The given query_request is then submitted + to the grid session using that replica, and this waits for the job to + complete. + + Arguments: + timeout: The number of seconds to wait for the job to complete or progress + before raising a SynapseTimeoutError. Defaults to 120. + query_request: The structured query to run against the grid, wrapping a + GridQuery that defines the column selection, filters, and whether to + include the detailed `all_validation_messages` list on each row. + 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 GridQueryResult containing the selected columns and rows, each with its own validation_results. Returns an empty GridQueryResult (and logs a warning) if the job completed but no rows matched the query. + + Raises: + ValueError: If session_id is not provided, or if no replica is bound + to this Grid (see `connect_async`/`connect`). + + Example: Validate every row of a grid session +   + + ```python + from synapseclient import Synapse + from synapseclient.models import Grid + from synapseclient.models.curation import GridQuery, QueryRequest, SelectAll + + syn = Synapse() + syn.login() + + with Grid(record_set_id="syn1234567").connect() as grid: + # SelectAll() selects every column in the grid, so each row's full + # data is returned alongside its validation results. + query_request = QueryRequest(query=GridQuery(column_selection=[SelectAll()])) + query_result = grid.validate_rows(query_request=query_request) - Raises: - ValueError: If session_id is not provided. + for row in query_result.rows: + validation = row.validation_results + print(f"Row ID: {row.row_id}, Validation Result: {validation}") + ``` - Example: Synchronize a grid session created from a file view + Example: Validate only the currently invalid rows of a grid session   ```python from synapseclient import Synapse from synapseclient.models import Grid - from synapseclient.models.table_components import Query + from synapseclient.models.curation import ( + GridQuery, + QueryRequest, + RowIsValidFilter, + SelectAll, + ) syn = Synapse() syn.login() - # First create a grid session from a file view - query = Query(sql="SELECT * FROM syn1234567") - grid = Grid(initial_query=query) - grid = grid.create() + with Grid(record_set_id="syn1234567").connect() as grid: + # Filter to only the rows that are currently invalid, and request the + # detailed allValidationMessages list on each one. + query_request = QueryRequest( + query=GridQuery( + column_selection=[SelectAll()], + filters=[RowIsValidFilter(value=False)], + include_validation_messages=True, + ) + ) + query_result = grid.validate_rows(query_request=query_request) - # Synchronize the grid with the latest state of the file view - grid = grid.synchronize() + for row in query_result.rows: + print(f"Invalid row {row.row_id}: {row.validation_results}") ``` """ - return self + return None def delete(self, *, synapse_client: Optional[Synapse] = None) -> None: """ @@ -3120,6 +4092,7 @@ class Grid(EnumCoercionMixin, GridSynchronousProtocol): ```python from synapseclient import Synapse from synapseclient.models import Grid + from synapseclient.models.curation import GridQuery, QueryRequest, SelectAll syn = Synapse() syn.login() @@ -3129,6 +4102,14 @@ class Grid(EnumCoercionMixin, GridSynchronousProtocol): grid = grid.create() print(f"Created grid session: {grid.session_id}") + # Validate rows. SelectAll() selects every column, so each row's full + # data is returned alongside its validation results. + with grid.connect() as session: + query_request = QueryRequest(query=GridQuery(column_selection=[SelectAll()])) + result = session.validate_rows(query_request=query_request) + for row in result.rows: + print(f"Row ID: {row.row_id}, Validation Result: {row.validation_results}") + # Later, export the modified data back to the record set grid = grid.export_to_record_set() print(f"Exported to version: {grid.record_set_version_number}") @@ -3144,6 +4125,7 @@ class Grid(EnumCoercionMixin, GridSynchronousProtocol): from synapseclient import Synapse from synapseclient.models import Grid from synapseclient.models.table_components import Query + from synapseclient.models.curation import GridQuery, QueryRequest, SelectAll syn = Synapse() syn.login() @@ -3153,7 +4135,14 @@ class Grid(EnumCoercionMixin, GridSynchronousProtocol): grid = Grid(initial_query=query) grid = grid.create() - # Work with the grid session... + # Validate rows. SelectAll() selects every column, so each row's full + # data is returned alongside its validation results. + with grid.connect() as session: + query_request = QueryRequest(query=GridQuery(column_selection=[SelectAll()])) + result = session.validate_rows(query_request=query_request) + for row in result.rows: + print(f"Row ID: {row.row_id}, Validation Result: {row.validation_results}") + # Export when ready grid = grid.export_to_record_set() ``` @@ -3208,6 +4197,10 @@ class Grid(EnumCoercionMixin, GridSynchronousProtocol): validation_summary_statistics: Optional[ValidationSummary] = None """Summary statistics for validation results""" + _replica_id: Optional[int] = field(default=None, repr=False, compare=False) + """The replica ID bound to this instance by `connect_async`. Reused by + `validate_rows_async` so repeated calls do not each create a new replica.""" + _ENUM_FIELDS: ClassVar[dict[str, type]] = {"authorization_mode": AuthorizationMode} async def create_async( @@ -3837,3 +4830,407 @@ async def main(): ) return self + + @otel_trace_method( + method_to_trace_name=lambda self, **kwargs: f"Grid_Create_Replica_Session_ID: {self.session_id}" + ) + async def _create_replica_async( + self, *, synapse_client: Optional[Synapse] = None + ) -> "GridReplica": + """ + Creates a new replica for this grid session. + + A grid replica is an in-memory document that represents a 'copy' of the + grid. Each replica is identified by a unique replicaId, issued by the + 'hub'. A user can have more than one replica at a time (i.e. using + multiple browser tabs/machines). A user is limited to 10 replicas + per-hour per-grid-session. Only the user that started the grid session + may create a replica. + + Arguments: + 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 newly created GridReplica. + + Raises: + ValueError: If session_id is not provided, or if the Synapse response + did not contain replica information. + + Example: Create a replica for a grid session +   + + ```python + import asyncio + from synapseclient import Synapse + from synapseclient.models import Grid + + syn = Synapse() + syn.login() + + async def main(): + grid = Grid(record_set_id="syn1234567") + grid = await grid.create_async() + + replica = await grid._create_replica_async() + print(f"Replica created with ID: {replica.replica_id}") + + asyncio.run(main()) + ``` + """ + if not self.session_id: + raise ValueError( + "session_id is required to create a replica for a GridSession" + ) + + grid_replica = await create_grid_replica( + session_id=self.session_id, + create_replica_request=CreateReplicaRequest( + self.session_id + ).to_synapse_request(), + synapse_client=synapse_client, + ) + replica_data = grid_replica.get("replica") + if not replica_data: + raise ValueError( + f"Replica could not be created for grid session '{self.session_id}': " + f"no replica was returned in the Synapse response." + ) + return GridReplica().fill_from_dict(replica_data) + + @skip_async_to_sync + @asynccontextmanager + async def connect_async( + self, + *, + attach_to_previous_session: bool = False, + timeout: int = 120, + synapse_client: Optional[Synapse] = None, + ) -> AsyncGenerator["Grid", None]: + """ + Connects to a grid session and binds a single replica to it for the + duration of the `async with` block. + + If `session_id` is not already set (e.g. from `create_grid_session`), + creates a new grid session first via `record_set_id` or + `initial_query`, same as `create_async`. Either way, one replica is + then created (see `_create_replica_async`) that is reused by every + `validate_rows_async` call made within the block. + + Arguments: + attach_to_previous_session: Only applies when creating a new + session from `record_set_id`. If True, attaches to an existing + active session instead of creating a new one. Defaults to False. + timeout: The number of seconds to wait for the 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. + + Yields: + The connected Grid, with a replica bound to it. + + Example: Validate rows using a newly created grid session +   + + ```python + import asyncio + from synapseclient import Synapse + from synapseclient.models import Grid + from synapseclient.models.curation import GridQuery, QueryRequest, SelectAll + + syn = Synapse() + syn.login() + + async def main(): + async with Grid(record_set_id="syn1234567").connect_async() as session: + query_request = QueryRequest( + query=GridQuery(column_selection=[SelectAll()]) + ) + result = await session.validate_rows_async(query_request=query_request) + for row in result.rows: + print(f"Row ID: {row.row_id}, Validation Result: {row.validation_results}") + + asyncio.run(main()) + ``` + + Example: Validate rows using an existing grid session +   + + If a session_id is already set, connect_async will not create a new + grid session + + ```python + import asyncio + from synapseclient import Synapse + from synapseclient.models import CurationTask, Grid + from synapseclient.models.curation import GridQuery, QueryRequest, SelectAll + + syn = Synapse() + syn.login() + + task = CurationTask(task_id="1234") + grid = task.create_grid_session() + + async def main(): + async with grid.connect_async() as session: + query_request = QueryRequest( + query=GridQuery(column_selection=[SelectAll()]) + ) + result = await session.validate_rows_async(query_request=query_request) + for row in result.rows: + print(f"Row ID: {row.row_id}, Validation Result: {row.validation_results}") + + asyncio.run(main()) + ``` + """ + trace.get_current_span().set_attributes( + { + "synapse.record_set_id": self.record_set_id or "", + "synapse.session_id": self.session_id or "", + } + ) + + if not self.session_id: + await self.create_async( + attach_to_previous_session=attach_to_previous_session, + timeout=timeout, + synapse_client=synapse_client, + ) + + replica = await self._create_replica_async(synapse_client=synapse_client) + self._replica_id = replica.replica_id + try: + yield self + finally: + self._replica_id = None + + @contextmanager + def connect( + self, + *, + attach_to_previous_session: bool = False, + timeout: int = 120, + synapse_client: Optional[Synapse] = None, + ) -> Generator["Grid", None, None]: + """ + Synchronous equivalent of `connect_async`. + + Connects to a grid session and binds a single replica to it for the + duration of the `with` block. + + If `session_id` is not already set (e.g. from `create_grid_session`), + creates a new grid session first via `record_set_id` or + `initial_query`, same as `create`. Either way, one replica is then + created (see `_create_replica`) that is reused by every + `validate_rows` call made within the block. + + Arguments: + attach_to_previous_session: Only applies when creating a new + session from `record_set_id`. If True, attaches to an existing + active session instead of creating a new one. Defaults to False. + timeout: The number of seconds to wait for the 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. + + Yields: + The connected Grid, with a replica bound to it. + + Example: Validate rows using a newly created grid session +   + + ```python + from synapseclient import Synapse + from synapseclient.models import Grid + from synapseclient.models.curation import GridQuery, QueryRequest, SelectAll + + syn = Synapse() + syn.login() + + with Grid(record_set_id="syn1234567").connect() as session: + query_request = QueryRequest( + query=GridQuery(column_selection=[SelectAll()]) + ) + result = session.validate_rows(query_request=query_request) + for row in result.rows: + print(f"Row ID: {row.row_id}, Validation Result: {row.validation_results}") + ``` + + Example: Validate rows using an existing grid session +   + + If a session_id is already set, connect will not create a new grid + session. + + ```python + from synapseclient import Synapse + from synapseclient.models import CurationTask, Grid + from synapseclient.models.curation import GridQuery, QueryRequest, SelectAll + + syn = Synapse() + syn.login() + + task = CurationTask(task_id="1234") + grid = task.create_grid_session() + + with grid.connect() as session: + query_request = QueryRequest( + query=GridQuery(column_selection=[SelectAll()]) + ) + result = session.validate_rows(query_request=query_request) + for row in result.rows: + print(f"Row ID: {row.row_id}, Validation Result: {row.validation_results}") + ``` + """ + trace.get_current_span().set_attributes( + { + "synapse.record_set_id": self.record_set_id or "", + "synapse.session_id": self.session_id or "", + } + ) + + if not self.session_id: + self.create( + attach_to_previous_session=attach_to_previous_session, + timeout=timeout, + synapse_client=synapse_client, + ) + + replica = self._create_replica(synapse_client=synapse_client) + self._replica_id = replica.replica_id + try: + yield self + finally: + self._replica_id = None + + @otel_trace_method( + method_to_trace_name=lambda self, **kwargs: f"Grid_Validate_Rows_Session_ID: {self.session_id}" + ) + async def validate_rows_async( + self, + *, + timeout: int = 120, + query_request: QueryRequest, + synapse_client: Optional[Synapse] = None, + ) -> "GridQueryResult": + """ + Queries this grid session's rows and returns their per-row validation + results against the grid's bound JSON schema. + + This Grid must have been obtained from `connect_async` (or `connect`), + which binds a replica to it. The given query_request is then submitted + to the grid session using that replica, and this waits for the job to + complete. + + Arguments: + timeout: The number of seconds to wait for the job to complete or progress + before raising a SynapseTimeoutError. Defaults to 120. + query_request: The structured query to run against the grid, wrapping a + GridQuery that defines the column selection, filters, and whether to + include the detailed `all_validation_messages` list on each row. + 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 GridQueryResult containing the selected columns and rows, each with its own validation_results. Returns an empty GridQueryResult (and logs a warning) if the job completed but no rows matched the query. + + Raises: + ValueError: If session_id is not provided, or if no replica is bound + to this Grid (see `connect_async`/`connect`). + + Example: Validate every row of a grid session +   + + ```python + import asyncio + from synapseclient import Synapse + from synapseclient.models import Grid + from synapseclient.models.curation import GridQuery, QueryRequest, SelectAll + + syn = Synapse() + syn.login() + + async def main(): + async with Grid(record_set_id="syn1234567").connect_async() as grid: + # SelectAll() selects every column in the grid, so each row's + # full data is returned alongside its validation results. + query_request = QueryRequest( + query=GridQuery(column_selection=[SelectAll()]) + ) + query_result = await grid.validate_rows_async(query_request=query_request) + + for row in query_result.rows: + validation = row.validation_results + print(f"Row ID: {row.row_id}, Validation Result: {validation}") + + asyncio.run(main()) + ``` + + Example: Validate only the currently invalid rows of a grid session +   + + ```python + import asyncio + from synapseclient import Synapse + from synapseclient.models import Grid + from synapseclient.models.curation import ( + GridQuery, + QueryRequest, + RowIsValidFilter, + SelectAll, + ) + + syn = Synapse() + syn.login() + + async def main(): + async with Grid(record_set_id="syn1234567").connect_async() as grid: + # Filter to only the rows that are currently invalid, and request + # the detailed allValidationMessages list on each one. + query_request = QueryRequest( + query=GridQuery( + column_selection=[SelectAll()], + filters=[RowIsValidFilter(value=False)], + include_validation_messages=True, + ) + ) + query_result = await grid.validate_rows_async(query_request=query_request) + + for row in query_result.rows: + print(f"Invalid row {row.row_id}: {row.validation_results}") + + asyncio.run(main()) + ``` + """ + if not self.session_id: + raise ValueError("session_id is required to validate rows") + + if self._replica_id is None: + raise ValueError( + "No replica is bound to this Grid. Use `connect_async` (or " + "`connect`) to connect to a grid session before calling " + "validate_rows_async." + ) + + request = GridQueryJobRequest( + session_id=self.session_id, + replica_id=self._replica_id, + query_request=query_request, + ) + request = await request.send_job_and_wait_async( + timeout=timeout, synapse_client=synapse_client + ) + + if not request.query_result or not request.query_result.rows: + client = Synapse.get_client(synapse_client=synapse_client) + client.logger.warning( + f"Validation job for grid session '{self.session_id}' completed but " + "did not return any row validation results. This grid may not have " + "any rows matching the query." + ) + return request.query_result diff --git a/synapseclient/models/mixins/asynchronous_job.py b/synapseclient/models/mixins/asynchronous_job.py index 88cce73be..cccda8400 100644 --- a/synapseclient/models/mixins/asynchronous_job.py +++ b/synapseclient/models/mixins/asynchronous_job.py @@ -17,6 +17,7 @@ DOWNLOAD_LIST_MANIFEST_REQUEST, GET_VALIDATION_SCHEMA_REQUEST, GRID_CSV_IMPORT_REQUEST, + GRID_QUERY_JOB_REQUEST, GRID_RECORD_SET_EXPORT_REQUEST, QUERY_BUNDLE_REQUEST, QUERY_TABLE_CSV_REQUEST, @@ -44,6 +45,7 @@ QUERY_TABLE_CSV_REQUEST: "/entity/{entityId}/table/download/csv/async", QUERY_BUNDLE_REQUEST: "/entity/{entityId}/table/query/async", GRID_CSV_IMPORT_REQUEST: "/grid/import/csv/async", + GRID_QUERY_JOB_REQUEST: "/grid/session/query/async", UPLOAD_TO_TABLE_PREVIEW_REQUEST: "/table/upload/csv/preview/async", } diff --git a/tests/integration/synapseclient/models/async/test_grid_async.py b/tests/integration/synapseclient/models/async/test_grid_async.py index b4eb32bf4..3355d3766 100644 --- a/tests/integration/synapseclient/models/async/test_grid_async.py +++ b/tests/integration/synapseclient/models/async/test_grid_async.py @@ -1,9 +1,10 @@ """Integration tests for the synapseclient.models.Grid class (async).""" +import asyncio import os import tempfile import uuid -from typing import Callable +from typing import AsyncGenerator, Callable, Generator, Tuple import pandas as pd import pytest @@ -21,7 +22,23 @@ ViewTypeMask, query_async, ) +from synapseclient.models.curation import ( + CellValueFilter, + CellValueOperator, + CountStar, + GridQuery, + GridReplica, + QueryRequest, + RowIdFilter, + RowIsValidFilter, + RowSelectionFilter, + RowValidationResultFilter, + SelectAll, + SelectByName, + ValidationOperator, +) from synapseclient.models.table_components import Query +from synapseclient.services.json_schema import JsonSchemaOrganization from tests.integration import ASYNC_JOB_TIMEOUT_SEC, QUERY_TIMEOUT_SEC from tests.integration.helpers import wait_for_condition @@ -385,3 +402,443 @@ async def test_download_csv_async(self, record_set_fixture: RecordSet) -> None: } ) pd.testing.assert_frame_equal(df, expected_df, check_dtype=False) + + async def test_create_replica_async(self, record_set_fixture: RecordSet) -> None: + # GIVEN: A grid session created first + grid = Grid(record_set_id=record_set_fixture.id) + created_grid = await grid.create_async( + timeout=ASYNC_JOB_TIMEOUT_SEC, synapse_client=self.syn + ) + self.schedule_for_cleanup(created_grid) + + # WHEN: Creating a replica for the grid session + replica = await created_grid._create_replica_async(synapse_client=self.syn) + + # THEN: A populated GridReplica should be returned + assert isinstance(replica, GridReplica) + assert replica.replica_id is not None + assert replica.grid_session_id == created_grid.session_id + assert replica.created_by == created_grid.started_by + + async def test_connect_async_creates_session_and_binds_replica( + self, record_set_fixture: RecordSet + ) -> None: + # GIVEN: A Grid instance with no existing session + grid = Grid(record_set_id=record_set_fixture.id) + + # WHEN: Connecting to the grid + async with grid.connect_async(synapse_client=self.syn) as connected_grid: + self.schedule_for_cleanup(connected_grid) + + # THEN: A session should have been created and a replica bound + assert connected_grid is grid + assert connected_grid.session_id is not None + assert connected_grid._replica_id is not None + + # AND: The replica should be unbound once the block exits + assert grid._replica_id is None + + +CELL_VALUE_FILTER_CASES = [ + pytest.param("category", CellValueOperator.EQUALS, "A", {1, 3}, id="equals"), + pytest.param( + "category", CellValueOperator.NOT_EQUALS, "A", {2, 4, 5}, id="not_equals" + ), + pytest.param("category", CellValueOperator.IN, ["A", "B"], {1, 2, 3, 5}, id="in"), + pytest.param("category", CellValueOperator.NOT_IN, ["A", "B"], {4}, id="not_in"), + pytest.param("value", CellValueOperator.GREATER_THAN, 100, {4}, id="greater_than"), + pytest.param( + "value", CellValueOperator.LESS_THAN, 100, {1, 2, 3, 5}, id="less_than" + ), + pytest.param( + "value", + CellValueOperator.GREATER_THAN_OR_EQUALS, + 50.9, + {4, 5}, + id="greater_than_or_equals", + ), + pytest.param( + "value", + CellValueOperator.LESS_THAN_OR_EQUALS, + 20.3, + {1, 2}, + id="less_than_or_equals", + ), + pytest.param("name", CellValueOperator.LIKE, "%lpha", {1}, id="like"), + pytest.param( + "name", CellValueOperator.NOT_LIKE, "%lpha", {2, 3, 4, 5}, id="not_like" + ), + pytest.param("category", CellValueOperator.IS_NULL, None, set(), id="is_null"), + pytest.param( + "category", + CellValueOperator.IS_NOT_NULL, + None, + {1, 2, 3, 4, 5}, + id="is_not_null", + ), + pytest.param( + "category", CellValueOperator.IS_UNDEFINED, None, set(), id="is_undefined" + ), + pytest.param( + "category", + CellValueOperator.IS_DEFINED, + None, + {1, 2, 3, 4, 5}, + id="is_defined", + ), +] + + +class TestGridValidateRowsAsync: + """Tests for Grid.validate_rows_async, covering every SelectItem and + Filter option, using one shared bound-schema grid session for the whole + class. Every query in this class is read-only, so + sharing the same session across tests is safe.""" + + @pytest.fixture(autouse=True, scope="function") + def init(self, syn: Synapse, schedule_for_cleanup: Callable[..., None]) -> None: + self.syn = syn + self.schedule_for_cleanup = schedule_for_cleanup + + @pytest.fixture(scope="class") + def create_test_schema( + self, syn: Synapse + ) -> Generator[Tuple[JsonSchemaOrganization, str, list], None, None]: + """Create a test JSON schema matching the RecordSet's columns, shared + for the lifetime of this class.""" + org_name = "gridtest" + uuid.uuid4().hex[:6] + schema_name = "grid.validation.schema" + + js = syn.service("json_schema") + created_org = js.create_organization(org_name) + record_set_ids = [] # Track RecordSets that need schema unbinding + + try: + schema = { + "$schema": "http://json-schema.org/draft-07/schema#", + "$id": f"https://example.com/schema/{schema_name}.json", + "title": "Grid ValidateRows Schema", + "type": "object", + "properties": { + "id": {"type": "integer"}, + "name": { + "description": "Name of the record (min 3 characters)", + "type": "string", + "minLength": 3, + }, + "value": { + "description": "Numeric value (must be >= 0 and <= 1000)", + "type": "number", + "minimum": 0, + "maximum": 1000, + }, + "category": { + "description": "Category classification (A, B, C, or D only)", + "type": "string", + "enum": ["A", "B", "C", "D"], + }, + "active": {"type": "boolean"}, + }, + "required": ["id", "name"], + } + + test_org = js.JsonSchemaOrganization(org_name) + created_schema = test_org.create_json_schema(schema, schema_name, "0.0.1") + yield test_org, created_schema.uri, record_set_ids + finally: + for record_set_id in record_set_ids: + try: + RecordSet(id=record_set_id).unbind_schema(synapse_client=syn) + except Exception: + pass # Ignore errors if already unbound or deleted + + try: + js.delete_json_schema(created_schema.uri) + except Exception: + pass # Ignore if schema can't be deleted + + try: + js.delete_organization(created_org["id"]) + except Exception: + pass # Ignore if org can't be deleted + + @pytest.fixture(scope="class") + async def record_set_with_schema_fixture( + self, + project_model: Project, + syn: Synapse, + schedule_for_cleanup: Callable[..., None], + create_test_schema: Tuple[JsonSchemaOrganization, str, list], + ) -> RecordSet: + """Create one RecordSet with a bound JSON schema and a mix of + valid/invalid rows (ids 3 and 4 invalid), shared by every test in + this class.""" + _, schema_uri, record_set_ids = create_test_schema + + # Row 3: INVALID - "name" below the schema's minLength of 3 + # Row 4: INVALID - "value" above the schema's maximum of 1000, and + # "category" not in the schema's enum + test_data = pd.DataFrame( + { + "id": [1, 2, 3, 4, 5], + "name": ["Alpha", "Beta", "AB", "Delta", "Epsilon"], + "value": [10.5, 20.3, 30.7, 1500.0, 50.9], + "category": ["A", "B", "A", "X", "B"], + "active": [True, False, True, True, False], + } + ) + + temp_fd, filename = tempfile.mkstemp(suffix=".csv") + try: + os.close(temp_fd) # Close the file descriptor + test_data.to_csv(filename, index=False) + schedule_for_cleanup(filename) + + record_set = RecordSet( + path=filename, + name=str(uuid.uuid4()), + description="Test RecordSet with bound schema for Grid testing", + version_comment="Grid schema test version", + version_label=str(uuid.uuid4()), + upsert_keys=["id", "name"], + ) + + stored_record_set = await record_set.store_async( + parent=project_model, synapse_client=syn + ) + schedule_for_cleanup(stored_record_set.id) + record_set_ids.append(stored_record_set.id) # Track for schema cleanup + + await asyncio.sleep(3) + + await stored_record_set.bind_schema_async( + json_schema_uri=schema_uri, + enable_derived_annotations=False, + synapse_client=syn, + ) + + # Wait for schema binding to be fully processed by backend + await asyncio.sleep(5) + + return stored_record_set + except Exception: + # Clean up the temp file if something goes wrong + if os.path.exists(filename): + os.unlink(filename) + raise + + @pytest.fixture(scope="class") + async def connected_grid( + self, + record_set_with_schema_fixture: RecordSet, + syn: Synapse, + schedule_for_cleanup: Callable[..., None], + ) -> AsyncGenerator[Grid, None]: + """Connect once, binding a single replica that every test method in + this class reuses for its (read-only) validate_rows_async calls.""" + grid = Grid(record_set_id=record_set_with_schema_fixture.id) + async with grid.connect_async(synapse_client=syn) as connected: + schedule_for_cleanup(connected) + yield connected + + async def test_select_all_returns_full_data_and_validation_results( + self, connected_grid: Grid + ) -> None: + # WHEN: Selecting every column for every row + result = await connected_grid.validate_rows_async( + query_request=QueryRequest(query=GridQuery(column_selection=[SelectAll()])), + synapse_client=self.syn, + ) + + # THEN: Every row has full data and real per-row validation results + assert len(result.rows) == 5 + for row in result.rows: + assert {"id", "name", "value", "category", "active"} <= set(row.data.keys()) + assert row.validation_results is not None + + results_by_id = {row.data["id"]: row.validation_results for row in result.rows} + assert results_by_id[1].is_valid is True + assert results_by_id[2].is_valid is True + assert results_by_id[3].is_valid is False + assert results_by_id[4].is_valid is False + assert results_by_id[5].is_valid is True + + async def test_select_by_name_returns_only_requested_columns( + self, connected_grid: Grid + ) -> None: + # WHEN: Selecting only "id" and "category" by name + result = await connected_grid.validate_rows_async( + query_request=QueryRequest( + query=GridQuery( + column_selection=[ + SelectByName(column_name="id"), + SelectByName(column_name="category"), + ] + ) + ), + synapse_client=self.syn, + ) + + # THEN: Every row's data contains exactly those two columns + assert len(result.rows) == 5 + for row in result.rows: + assert set(row.data.keys()) == {"id", "category"} + + async def test_row_is_valid_filter_returns_only_invalid_rows( + self, connected_grid: Grid + ) -> None: + # WHEN: Filtering to only invalid rows + result = await connected_grid.validate_rows_async( + query_request=QueryRequest( + query=GridQuery( + column_selection=[SelectAll()], + filters=[RowIsValidFilter(value=False)], + ) + ), + synapse_client=self.syn, + ) + + # THEN: Only the two rows that violate the schema are returned + assert {row.data["id"] for row in result.rows} == {3, 4} + + @pytest.mark.parametrize( + "operator,validation_result_value,expected_ids", + [ + # LIKE '%' matches any non-empty message, so only the invalid + # rows (which have a message) match. + pytest.param(ValidationOperator.LIKE, "%", {3, 4}, id="like"), + # NOT_LIKE '%' never matches: invalid rows have a message that + # matches '%' (and so are excluded), while valid rows have no + # message at all to compare against. + pytest.param(ValidationOperator.NOT_LIKE, "%", set(), id="not_like"), + ], + ) + async def test_row_validation_result_filter_operators( + self, + connected_grid: Grid, + operator: ValidationOperator, + validation_result_value: str, + expected_ids: set, + ) -> None: + # WHEN: Filtering rows by their validation message + result = await connected_grid.validate_rows_async( + query_request=QueryRequest( + query=GridQuery( + column_selection=[SelectAll()], + filters=[ + RowValidationResultFilter( + operator=operator, + validation_result_value=validation_result_value, + ) + ], + include_validation_messages=True, + ) + ), + synapse_client=self.syn, + ) + + # THEN: Only the rows matching that operator/value are returned + assert {row.data["id"] for row in result.rows} == expected_ids + if expected_ids: + assert all( + row.validation_results.all_validation_messages for row in result.rows + ) + + async def test_row_id_filter_returns_only_specified_rows( + self, connected_grid: Grid + ) -> None: + # GIVEN: row_ids captured from an initial unfiltered query + all_rows_result = await connected_grid.validate_rows_async( + query_request=QueryRequest(query=GridQuery(column_selection=[SelectAll()])), + synapse_client=self.syn, + ) + row_ids_by_data_id = { + row.data["id"]: row.row_id for row in all_rows_result.rows + } + + # WHEN: Filtering by two explicit row_ids (ids 1 and 5) + result = await connected_grid.validate_rows_async( + query_request=QueryRequest( + query=GridQuery( + column_selection=[SelectAll()], + filters=[ + RowIdFilter( + row_ids_in=[ + row_ids_by_data_id[1], + row_ids_by_data_id[5], + ] + ) + ], + ) + ), + synapse_client=self.syn, + ) + + # THEN: Only those two rows are returned + assert {row.data["id"] for row in result.rows} == {1, 5} + + async def test_row_selection_filter_excludes_unselected_rows( + self, connected_grid: Grid + ) -> None: + # WHEN: Filtering to rows the user has actively selected in the + # interface. Nothing has been marked as selected through this API + # path, so no rows should match. + result = await connected_grid.validate_rows_async( + query_request=QueryRequest( + query=GridQuery( + column_selection=[SelectAll()], + filters=[RowSelectionFilter(is_selected=True)], + ) + ), + synapse_client=self.syn, + ) + + # THEN: No rows are selected + assert len(result.rows) == 0 + + async def test_count_star_returns_aggregate_count( + self, connected_grid: Grid + ) -> None: + # WHEN: Counting only the valid rows instead of selecting them + result = await connected_grid.validate_rows_async( + query_request=QueryRequest( + query=GridQuery( + column_selection=[CountStar(alias="total")], + filters=[RowIsValidFilter(value=True)], + ) + ), + synapse_client=self.syn, + ) + + # THEN: A single aggregate row with the count is returned + assert len(result.rows) == 1 + assert result.rows[0].data["count"] == 3 + + @pytest.mark.parametrize( + "column_name,operator,value,expected_ids", CELL_VALUE_FILTER_CASES + ) + async def test_cell_value_filter_operators( + self, + connected_grid: Grid, + column_name: str, + operator: CellValueOperator, + value, + expected_ids: set, + ) -> None: + # WHEN: Filtering by a single column/operator/value combination + result = await connected_grid.validate_rows_async( + query_request=QueryRequest( + query=GridQuery( + column_selection=[SelectAll()], + filters=[ + CellValueFilter( + column_name=column_name, operator=operator, value=value + ) + ], + ) + ), + synapse_client=self.syn, + ) + + # THEN: Only the rows matching that operator/value are returned + assert {row.data["id"] for row in result.rows} == expected_ids 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..09d78f92b 100644 --- a/tests/unit/synapseclient/models/async/unit_test_curation_async.py +++ b/tests/unit/synapseclient/models/async/unit_test_curation_async.py @@ -7,16 +7,30 @@ from synapseclient import Synapse from synapseclient.core.constants.concrete_types import ( + CELL_VALUE_FILTER, + COUNT_STAR, FILE_BASED_METADATA_TASK_PROPERTIES, GRID_CSV_IMPORT_REQUEST, GRID_EXECUTION_DETAILS, + GRID_QUERY_JOB_REQUEST, RECORD_BASED_METADATA_TASK_PROPERTIES, + ROW_ID_FILTER, + ROW_IS_VALID_FILTER, + ROW_SELECTION_FILTER, + ROW_VALIDATION_RESULT_FILTER, + SELECT_ALL, + SELECT_BY_NAME, + SELECT_SELECTION, UPLOAD_TO_TABLE_PREVIEW_REQUEST, ) from synapseclient.models import EntityView, RecordSet from synapseclient.models.curation import ( AuthorizationMode, + CellValueFilter, + CellValueOperator, + CountStar, CreateGridRequest, + CreateReplicaRequest, CurationTask, CurationTaskStatus, DownloadFromGridRequest, @@ -24,8 +38,23 @@ Grid, GridCsvImportRequest, GridExecutionDetails, + GridQuery, + GridQueryJobRequest, + GridQueryResult, + GridQueryValidationResult, GridRecordSetExportRequest, + GridReplica, + GridRow, + QueryRequest, RecordBasedMetadataTaskProperties, + RowIdFilter, + RowIsValidFilter, + RowSelectionFilter, + RowValidationResultFilter, + SelectAll, + SelectByName, + SelectColumn, + SelectSelection, SynchronizeGridRequest, TaskState, UploadToTablePreviewRequest, @@ -55,6 +84,7 @@ STARTED_ON = "2024-03-01T00:00:00.000Z" FILE_HANDLE_ID = "1234567" OWNER_PRINCIPAL_ID = 987654 +REPLICA_ID = 12345 def _get_file_based_task_api_response(): @@ -2447,3 +2477,721 @@ async def test_synchronize_grid_async_with_errors(self) -> None: error_message = mock_logger.error.call_args[0][0] assert "sync_error_1" in error_message assert "sync_error_2" in error_message + + +class TestSelectColumn: + """Tests for the SelectColumn dataclass.""" + + def test_fill_from_dict(self) -> None: + # GIVEN a response with a column name + response = {"columnName": "diagnosis"} + + # WHEN I fill a SelectColumn from the response + result = SelectColumn().fill_from_dict(response) + + # THEN the column_name should be populated + assert result.column_name == "diagnosis" + + def test_fill_from_dict_missing_column_name(self) -> None: + # GIVEN a response without a column name + # WHEN I fill a SelectColumn from the response + result = SelectColumn().fill_from_dict({}) + + # THEN the column_name should be None + assert result.column_name is None + + +class TestGridRow: + """Tests for the GridRow dataclass.""" + + def test_fill_from_dict(self) -> None: + # GIVEN a response with row data and validation results + response = { + "rowId": "1.2", + "data": {"diagnosis": "flu"}, + "validationResults": {"isValid": True}, + } + + # WHEN I fill a GridRow from the response + result = GridRow().fill_from_dict(response) + + # THEN the fields should be populated + assert result.row_id == "1.2" + assert result.data == {"diagnosis": "flu"} + assert isinstance(result.validation_results, GridQueryValidationResult) + assert result.validation_results.is_valid is True + + def test_fill_from_dict_without_validation_results(self) -> None: + # GIVEN a response without validation results + response = {"rowId": "1.3", "data": {"diagnosis": "cold"}} + + # WHEN I fill a GridRow from the response + result = GridRow().fill_from_dict(response) + + # THEN validation_results should be None + assert result.row_id == "1.3" + assert result.validation_results is None + + +class TestGridQueryValidationResult: + """Tests for the GridQueryValidationResult dataclass.""" + + def test_fill_from_dict(self) -> None: + # GIVEN a response with validation result data + response = { + "isValid": False, + "validationErrorMessage": "#: only 1 subschema matches out of 2", + "allValidationMessages": ["error one", "error two"], + } + + # WHEN I fill a GridQueryValidationResult from the response + result = GridQueryValidationResult().fill_from_dict(response) + + # THEN the fields should be populated + assert result.is_valid is False + assert result.validation_error_message == ( + "#: only 1 subschema matches out of 2" + ) + assert result.all_validation_messages == ["error one", "error two"] + + def test_fill_from_dict_valid_row(self) -> None: + # GIVEN a response for a valid row with no error messages + response = {"isValid": True} + + # WHEN I fill a GridQueryValidationResult from the response + result = GridQueryValidationResult().fill_from_dict(response) + + # THEN is_valid should be True and the message fields should be None + assert result.is_valid is True + assert result.validation_error_message is None + assert result.all_validation_messages is None + + +class TestGridQueryResult: + """Tests for the GridQueryResult dataclass.""" + + def test_fill_from_dict(self) -> None: + # GIVEN a response with select columns and rows + response = { + "selectColumns": [{"columnName": "diagnosis"}, {"columnName": "age"}], + "rows": [ + {"rowId": "1.1", "data": {"diagnosis": "flu", "age": 30}}, + {"rowId": "1.2", "data": {"diagnosis": "cold", "age": 40}}, + ], + } + + # WHEN I fill a GridQueryResult from the response + result = GridQueryResult().fill_from_dict(response) + + # THEN the select_columns and rows should be populated + assert len(result.select_columns) == 2 + assert all(isinstance(col, SelectColumn) for col in result.select_columns) + assert result.select_columns[0].column_name == "diagnosis" + assert len(result.rows) == 2 + assert all(isinstance(row, GridRow) for row in result.rows) + assert result.rows[1].row_id == "1.2" + + def test_fill_from_dict_empty_response(self) -> None: + # GIVEN a response with no select columns or rows + # WHEN I fill a GridQueryResult from the response + result = GridQueryResult().fill_from_dict({}) + + # THEN both fields should be None + assert result.select_columns is None + assert result.rows is None + + +class TestSelectItemSubclasses: + """Tests for the SelectItem subclasses: SelectByName, SelectAll, CountStar, + and SelectSelection.""" + + def test_select_by_name_to_synapse_request(self) -> None: + # GIVEN a SelectByName with a column name + item = SelectByName(column_name="diagnosis") + + # WHEN I convert it to a synapse request + result = item.to_synapse_request() + + # THEN it should contain the concreteType and columnName + assert result == {"concreteType": SELECT_BY_NAME, "columnName": "diagnosis"} + + def test_select_all_to_synapse_request(self) -> None: + # GIVEN a SelectAll item + item = SelectAll() + + # WHEN I convert it to a synapse request + result = item.to_synapse_request() + + # THEN it should only contain the concreteType + assert result == {"concreteType": SELECT_ALL} + + def test_count_star_to_synapse_request_without_alias(self) -> None: + # GIVEN a CountStar with no alias + item = CountStar() + + # WHEN I convert it to a synapse request + result = item.to_synapse_request() + + # THEN the alias key should be omitted + assert result == {"concreteType": COUNT_STAR} + + def test_select_selection_to_synapse_request(self) -> None: + # GIVEN a SelectSelection item + item = SelectSelection() + + # WHEN I convert it to a synapse request + result = item.to_synapse_request() + + # THEN it should only contain the concreteType + assert result == {"concreteType": SELECT_SELECTION} + + +class TestFilterSubclasses: + """Tests for the Filter subclasses: RowValidationResultFilter, CellValueFilter, + RowSelectionFilter, RowIsValidFilter, and RowIdFilter.""" + + def test_row_validation_result_filter_to_synapse_request(self) -> None: + # GIVEN a RowValidationResultFilter constructed with a string operator + item = RowValidationResultFilter( + operator="LIKE", validation_result_value="%expected type:%" + ) + + # WHEN I convert it to a synapse request + result = item.to_synapse_request() + + # THEN the operator should be serialized as its string value + assert result == { + "concreteType": ROW_VALIDATION_RESULT_FILTER, + "operator": "LIKE", + "validationResultValue": "%expected type:%", + } + + def test_cell_value_filter_to_synapse_request(self) -> None: + # GIVEN a CellValueFilter with an enum operator + item = CellValueFilter( + column_name="Project", + operator=CellValueOperator.EQUALS, + value=["Alpha"], + ) + + # WHEN I convert it to a synapse request + result = item.to_synapse_request() + + # THEN it should contain the correct fields + assert result == { + "concreteType": CELL_VALUE_FILTER, + "columnName": "Project", + "operator": "EQUALS", + "value": ["Alpha"], + } + + def test_row_selection_filter_to_synapse_request(self) -> None: + # GIVEN a RowSelectionFilter + item = RowSelectionFilter(is_selected=False) + + # WHEN I convert it to a synapse request + result = item.to_synapse_request() + + # THEN it should contain the correct fields + assert result == { + "concreteType": ROW_SELECTION_FILTER, + "isSelected": False, + } + + def test_row_is_valid_filter_to_synapse_request(self) -> None: + # GIVEN a RowIsValidFilter + item = RowIsValidFilter(value=True) + + # WHEN I convert it to a synapse request + result = item.to_synapse_request() + + # THEN it should contain the correct fields + assert result == {"concreteType": ROW_IS_VALID_FILTER, "value": True} + + def test_row_id_filter_to_synapse_request(self) -> None: + # GIVEN a RowIdFilter + item = RowIdFilter(row_ids_in=["1.1", "1.2"]) + + # WHEN I convert it to a synapse request + result = item.to_synapse_request() + + # THEN it should contain the correct fields + assert result == { + "concreteType": ROW_ID_FILTER, + "rowIdsIn": ["1.1", "1.2"], + } + + +class TestGridQuery: + """Tests for the GridQuery dataclass.""" + + def test_to_synapse_request(self) -> None: + # GIVEN a GridQuery with select items and filters + query = GridQuery( + column_selection=[SelectAll(), SelectByName(column_name="diagnosis")], + filters=[RowIsValidFilter(value=True)], + limit=25, + offset=5, + include_validation_messages=False, + ) + + # WHEN I convert it to a synapse request + result = query.to_synapse_request() + + # THEN it should contain the serialized select items and filters + assert result["columnSelection"] == [ + {"concreteType": SELECT_ALL}, + {"concreteType": SELECT_BY_NAME, "columnName": "diagnosis"}, + ] + assert result["filters"] == [ + {"concreteType": ROW_IS_VALID_FILTER, "value": True} + ] + assert result["limit"] == 25 + assert result["offset"] == 5 + assert result["includeValidationMessages"] is False + + def test_to_synapse_request_without_filters(self) -> None: + # GIVEN a GridQuery with no filters set + query = GridQuery(column_selection=[SelectAll()], limit=10) + + # WHEN I convert it to a synapse request + result = query.to_synapse_request() + + # THEN the filters key should be omitted + assert "filters" not in result + assert result["columnSelection"] == [{"concreteType": SELECT_ALL}] + assert result["limit"] == 10 + + def test_to_synapse_request_with_empty_column_selection_raises(self) -> None: + # GIVEN a GridQuery with no column_selection set + query = GridQuery() + + # WHEN I convert it to a synapse request + # THEN it should raise ValueError + with pytest.raises(ValueError, match="column_selection is required"): + query.to_synapse_request() + + +class TestQueryRequest: + """Tests for the QueryRequest dataclass.""" + + def test_to_synapse_request(self) -> None: + # GIVEN a QueryRequest with a GridQuery + request = QueryRequest( + query=GridQuery(column_selection=[SelectAll()], limit=10) + ) + + # WHEN I convert it to a synapse request + result = request.to_synapse_request() + + # THEN it should contain the serialized query + assert result["query"]["columnSelection"] == [{"concreteType": SELECT_ALL}] + assert result["query"]["limit"] == 10 + + def test_to_synapse_request_without_query(self) -> None: + # GIVEN a QueryRequest with no query set + request = QueryRequest() + + # WHEN I convert it to a synapse request + result = request.to_synapse_request() + + # THEN the query key should be omitted + assert result == {} + + +class TestGridQueryJobRequest: + """Tests for the GridQueryJobRequest dataclass.""" + + def test_to_synapse_request(self) -> None: + # GIVEN a GridQueryJobRequest with a query + job_request = GridQueryJobRequest( + session_id=SESSION_ID, + replica_id=REPLICA_ID, + query_request=QueryRequest( + query=GridQuery(column_selection=[SelectAll()], limit=50) + ), + ) + + # WHEN I convert it to a synapse request + result = job_request.to_synapse_request() + + # THEN it should contain the correct fields + assert result["concreteType"] == GRID_QUERY_JOB_REQUEST + assert result["sessionId"] == SESSION_ID + assert result["replicaId"] == REPLICA_ID + assert result["queryRequest"]["query"]["columnSelection"] == [ + {"concreteType": SELECT_ALL} + ] + assert result["queryRequest"]["query"]["limit"] == 50 + + def test_to_synapse_request_with_default_query_request_raises(self) -> None: + # GIVEN a GridQueryJobRequest with no query set on its query_request + job_request = GridQueryJobRequest(session_id=SESSION_ID, replica_id=REPLICA_ID) + + # WHEN I convert it to a synapse request + # THEN it should raise ValueError since Synapse would reject the request + with pytest.raises(ValueError, match="query_request.query is required"): + job_request.to_synapse_request() + + def test_to_synapse_request_with_query_request_none_raises(self) -> None: + # GIVEN a GridQueryJobRequest with query_request explicitly set to None + job_request = GridQueryJobRequest( + session_id=SESSION_ID, replica_id=REPLICA_ID, query_request=None + ) + + # WHEN I convert it to a synapse request + # THEN it should raise ValueError since Synapse would reject the request + with pytest.raises(ValueError, match="query_request.query is required"): + job_request.to_synapse_request() + + def test_fill_from_dict(self) -> None: + # GIVEN a response with a queryResult + response = { + "concreteType": "org.sagebionetworks.repo.model.grid.GridQueryJobResponse", + "queryResult": { + "selectColumns": [{"columnName": "diagnosis"}], + "rows": [{"rowId": "1.1", "data": {"diagnosis": "flu"}}], + }, + } + + # WHEN I fill a GridQueryJobRequest from the response + job_request = GridQueryJobRequest(session_id=SESSION_ID, replica_id=REPLICA_ID) + job_request.fill_from_dict(response) + + # THEN query_result should be populated as a GridQueryResult + assert isinstance(job_request.query_result, GridQueryResult) + assert job_request.query_result.select_columns[0].column_name == "diagnosis" + assert job_request.query_result.rows[0].row_id == "1.1" + + def test_fill_from_dict_without_query_result(self) -> None: + # GIVEN a response without a queryResult + response = { + "concreteType": "org.sagebionetworks.repo.model.grid.GridQueryJobResponse" + } + + # WHEN I fill a GridQueryJobRequest from the response + job_request = GridQueryJobRequest(session_id=SESSION_ID, replica_id=REPLICA_ID) + job_request.fill_from_dict(response) + + # THEN query_result should be None + assert job_request.query_result is None + + +class TestGridReplica: + """Tests for the GridReplica dataclass.""" + + def test_fill_from_dict(self) -> None: + # GIVEN a response with replica data + response = { + "gridSessionId": SESSION_ID, + "replicaId": REPLICA_ID, + "createdBy": CREATED_BY, + "isAgentReplica": False, + "createdOn": CREATED_ON, + } + + # WHEN I fill a GridReplica from the response + result = GridReplica().fill_from_dict(response) + + # THEN the fields should be populated + assert result.grid_session_id == SESSION_ID + assert result.replica_id == REPLICA_ID + assert result.created_by == CREATED_BY + assert result.is_agent_replica is False + assert result.created_on == CREATED_ON + + +class TestCreateReplicaRequest: + """Tests for the CreateReplicaRequest dataclass.""" + + def test_to_synapse_request(self) -> None: + # GIVEN a CreateReplicaRequest with a grid_session_id + request = CreateReplicaRequest(grid_session_id=SESSION_ID) + + # WHEN I convert it to a synapse request + result = request.to_synapse_request() + + # THEN it should contain the gridSessionId + assert result == {"gridSessionId": SESSION_ID} + + +class TestGridCreateReplica: + """Tests for Grid._create_replica_async.""" + + @pytest.fixture(autouse=True, scope="function") + def init_syn(self, syn: Synapse) -> None: + self.syn = syn + + async def test_create_replica_async_without_session_id_raises(self) -> None: + # GIVEN a Grid without a session_id + grid = Grid() + + # WHEN I call _create_replica_async + # THEN it should raise ValueError + with pytest.raises( + ValueError, match="session_id is required to create a replica" + ): + await grid._create_replica_async(synapse_client=self.syn) + + async def test_create_replica_async_returns_grid_replica(self) -> None: + # GIVEN a Grid with a session_id and a mocked API response + grid = Grid(session_id=SESSION_ID) + mock_response = { + "replica": { + "gridSessionId": SESSION_ID, + "replicaId": REPLICA_ID, + "createdBy": CREATED_BY, + "isAgentReplica": False, + "createdOn": CREATED_ON, + } + } + + # WHEN I call _create_replica_async + with patch( + "synapseclient.models.curation.create_grid_replica", + new_callable=AsyncMock, + return_value=mock_response, + ) as mock_create: + result = await grid._create_replica_async(synapse_client=self.syn) + + # THEN the API should be called with the session_id and request body + mock_create.assert_called_once_with( + session_id=SESSION_ID, + create_replica_request={"gridSessionId": SESSION_ID}, + synapse_client=self.syn, + ) + + # THEN the result should be a populated GridReplica + assert isinstance(result, GridReplica) + assert result.replica_id == REPLICA_ID + assert result.grid_session_id == SESSION_ID + assert result.created_by == CREATED_BY + assert result.is_agent_replica is False + assert result.created_on == CREATED_ON + + async def test_create_replica_async_raises_without_replica_in_response( + self, + ) -> None: + # GIVEN a Grid with a session_id and a response with no replica data + grid = Grid(session_id=SESSION_ID) + + # WHEN I call _create_replica_async + # THEN it should raise ValueError since no replica was returned + with patch( + "synapseclient.models.curation.create_grid_replica", + new_callable=AsyncMock, + return_value={}, + ): + with pytest.raises(ValueError, match="Replica could not be created"): + await grid._create_replica_async(synapse_client=self.syn) + + +class TestGridConnect: + """Tests for Grid.connect_async.""" + + @pytest.fixture(autouse=True, scope="function") + def init_syn(self, syn: Synapse) -> None: + self.syn = syn + + async def test_connect_async_creates_session_when_no_session_id(self) -> None: + # GIVEN a Grid with no session_id, so a new session needs to be created + grid = Grid(record_set_id=RECORD_SET_ID) + + with ( + patch.object( + Grid, + "create_async", + new_callable=AsyncMock, + side_effect=lambda **kwargs: grid, + ) as mock_create_async, + patch.object( + Grid, + "_create_replica_async", + new_callable=AsyncMock, + return_value=GridReplica(replica_id=REPLICA_ID), + ) as mock_create_replica, + ): + # WHEN I connect to the grid + async with grid.connect_async(synapse_client=self.syn) as session: + # THEN create_async should be called since no session_id was set + mock_create_async.assert_called_once_with( + attach_to_previous_session=False, + timeout=120, + synapse_client=self.syn, + ) + # AND _create_replica_async should be called to bind a replica + mock_create_replica.assert_called_once_with(synapse_client=self.syn) + # AND the replica_id should be bound on the yielded Grid + assert session._replica_id == REPLICA_ID + + # THEN the replica_id should be cleared after exiting the block + assert grid._replica_id is None + + async def test_connect_async_does_not_create_session_when_session_id_provided( + self, + ) -> None: + # GIVEN a Grid that already has a session_id (e.g. an existing session) + grid = Grid(session_id=SESSION_ID) + + with ( + patch.object( + Grid, "create_async", new_callable=AsyncMock + ) as mock_create_async, + patch.object( + Grid, + "_create_replica_async", + new_callable=AsyncMock, + return_value=GridReplica(replica_id=REPLICA_ID), + ) as mock_create_replica, + ): + # WHEN I connect to the grid + async with grid.connect_async(synapse_client=self.syn) as session: + # THEN create_async should NOT be called since session_id was + # already set + mock_create_async.assert_not_called() + # AND _create_replica_async should still be called to bind a + # replica to the existing session + mock_create_replica.assert_called_once_with(synapse_client=self.syn) + assert session.session_id == SESSION_ID + assert session._replica_id == REPLICA_ID + + # THEN the replica_id should be cleared after exiting the block + assert grid._replica_id is None + + +class TestGridValidateRows: + """Tests for Grid.validate_rows_async.""" + + @pytest.fixture(autouse=True, scope="function") + def init_syn(self, syn: Synapse) -> None: + self.syn = syn + + async def test_validate_rows_async_without_session_id_raises(self) -> None: + # GIVEN a Grid without a session_id + grid = Grid() + + # WHEN I call validate_rows_async + # THEN it should raise ValueError + with pytest.raises(ValueError, match="session_id is required to validate rows"): + await grid.validate_rows_async( + synapse_client=self.syn, query_request=QueryRequest() + ) + + async def test_validate_rows_async_without_replica_id_raises(self) -> None: + # GIVEN a Grid with a session_id but no replica bound to it + grid = Grid(session_id=SESSION_ID) + + # WHEN I call validate_rows_async + # THEN it should raise ValueError + with pytest.raises(ValueError, match="No replica is bound to this Grid"): + await grid.validate_rows_async( + synapse_client=self.syn, query_request=QueryRequest() + ) + + async def test_validate_rows_async_returns_query_result(self) -> None: + # GIVEN a Grid with a session_id and a replica already bound to it (as + # would be the case after `connect_async`/`connect`), and a mocked API + # response + grid = Grid(session_id=SESSION_ID) + grid._replica_id = REPLICA_ID + + # Build a GridQueryJobRequest with query_result already populated + mock_job_request = GridQueryJobRequest( + session_id=SESSION_ID, replica_id=REPLICA_ID + ) + grid_query_result = GridQueryResult().fill_from_dict( + { + "selectColumns": [ + {"columnName": "Sex"}, + {"columnName": "Diagnosis"}, + ], + "rows": [ + { + "rowId": "123", + "data": {"Sex": "Female", "Diagnosis": "Cancer"}, + "validationResults": { + "isValid": False, + "validationErrorMessage": "#: only 1 subschema matches out of 2", + }, + }, + { + "rowId": "456", + "data": {"Sex": "Male", "Diagnosis": "Cancer"}, + "validationResults": { + "isValid": False, + "validationErrorMessage": "#: only 1 subschema matches out of 2", + }, + }, + ], + } + ) + mock_job_request.query_result = grid_query_result + + # WHEN I call validate_rows_async + with patch.object( + GridQueryJobRequest, + "send_job_and_wait_async", + new_callable=AsyncMock, + return_value=mock_job_request, + ): + query_request = QueryRequest( + query=GridQuery(column_selection=[SelectAll()]) + ) + result = await grid.validate_rows_async( + synapse_client=self.syn, + query_request=query_request, + ) + + # THEN the result should be a populated GridQueryResult + assert isinstance(result, GridQueryResult) + assert result.select_columns[0].column_name == "Sex" + assert result.select_columns[1].column_name == "Diagnosis" + assert result.rows[0].row_id == "123" + assert result.rows[0].data == {"Sex": "Female", "Diagnosis": "Cancer"} + assert result.rows[0].validation_results.is_valid is False + assert result.rows[1].row_id == "456" + assert result.rows[1].data == {"Sex": "Male", "Diagnosis": "Cancer"} + assert result.rows[1].validation_results.is_valid is False + # AND the replica_id is cached on the Grid instance + assert grid._replica_id == REPLICA_ID + + async def test_validate_rows_async_no_rows_returns_existing_result_and_warns( + self, + ) -> None: + # GIVEN a Grid with a session_id and a replica already bound to it, and + # a mocked API response with a query_result that has no rows + grid = Grid(session_id=SESSION_ID) + grid._replica_id = REPLICA_ID + + mock_job_request = GridQueryJobRequest( + session_id=SESSION_ID, replica_id=REPLICA_ID + ) + mock_job_request.query_result = GridQueryResult().fill_from_dict( + {"selectColumns": [{"columnName": "Sex"}], "rows": []} + ) + + # WHEN I call validate_rows_async + with ( + patch.object( + GridQueryJobRequest, + "send_job_and_wait_async", + new_callable=AsyncMock, + return_value=mock_job_request, + ), + patch.object(self.syn.logger, "warning") as mock_warning, + ): + query_request = QueryRequest( + query=GridQuery(column_selection=[SelectAll()]) + ) + result = await grid.validate_rows_async( + synapse_client=self.syn, + query_request=query_request, + ) + + # THEN it should return the (empty-rows) query_result rather than + # discarding it, and log a warning instead of raising + assert result is mock_job_request.query_result + assert result.rows == [] + assert result.select_columns[0].column_name == "Sex" + mock_warning.assert_called_once() + assert SESSION_ID in mock_warning.call_args[0][0]