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..3f0a6040e 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,444 @@ 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 + + # TODO: Update once https://sagebionetworks.jira.com/browse/PLFM-9831 is resolved. + @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]