diff --git a/docs/reference/experimental/async/table.md b/docs/reference/experimental/async/table.md index da59e0d0a..d57e499ed 100644 --- a/docs/reference/experimental/async/table.md +++ b/docs/reference/experimental/async/table.md @@ -80,12 +80,16 @@ ::: synapseclient.models.PartialRow [](){ #partial-row-set-reference-async } ::: synapseclient.models.PartialRowSet +[](){ #table-update-request-reference-async } +::: synapseclient.models.TableUpdateRequest [](){ #table-schema-change-request-reference-async } ::: synapseclient.models.TableSchemaChangeRequest [](){ #appendable-row-set-request-reference-async } ::: synapseclient.models.AppendableRowSetRequest [](){ #upload-to-table-request-reference-async } ::: synapseclient.models.UploadToTableRequest +[](){ #table-search-change-request-reference-async } +::: synapseclient.models.TableSearchChangeRequest [](){ #table-update-transaction-reference-async } ::: synapseclient.models.TableUpdateTransaction [](){ #csv-table-descriptor-reference-async } diff --git a/docs/reference/experimental/sync/table.md b/docs/reference/experimental/sync/table.md index 2c35067ae..f3bfaad26 100644 --- a/docs/reference/experimental/sync/table.md +++ b/docs/reference/experimental/sync/table.md @@ -92,12 +92,16 @@ ::: synapseclient.models.PartialRow [](){ #partial-row-set-reference-sync } ::: synapseclient.models.PartialRowSet +[](){ #table-update-request-reference-sync } +::: synapseclient.models.TableUpdateRequest [](){ #table-schema-change-request-reference-sync } ::: synapseclient.models.TableSchemaChangeRequest [](){ #appendable-row-set-request-reference-sync } ::: synapseclient.models.AppendableRowSetRequest [](){ #upload-to-table-request-reference-sync } ::: synapseclient.models.UploadToTableRequest +[](){ #table-search-change-request-reference-sync } +::: synapseclient.models.TableSearchChangeRequest [](){ #table-update-transaction-reference-sync } ::: synapseclient.models.TableUpdateTransaction [](){ #csv-table-descriptor-reference-sync } diff --git a/synapseclient/core/constants/concrete_types.py b/synapseclient/core/constants/concrete_types.py index e24ab9da9..4366dc9d3 100644 --- a/synapseclient/core/constants/concrete_types.py +++ b/synapseclient/core/constants/concrete_types.py @@ -62,6 +62,12 @@ TABLE_SCHEMA_CHANGE_REQUEST = ( "org.sagebionetworks.repo.model.table.TableSchemaChangeRequest" ) +TABLE_SEARCH_CHANGE_RESPONSE = ( + "org.sagebionetworks.repo.model.table.TableSearchChangeResponse" +) +TABLE_SEARCH_CHANGE_REQUEST = ( + "org.sagebionetworks.repo.model.table.TableSearchChangeRequest" +) TABLE_UPDATE_TRANSACTION_REQUEST = ( "org.sagebionetworks.repo.model.table.TableUpdateTransactionRequest" ) diff --git a/synapseclient/models/__init__.py b/synapseclient/models/__init__.py index 68a0164af..452a779c6 100644 --- a/synapseclient/models/__init__.py +++ b/synapseclient/models/__init__.py @@ -72,6 +72,9 @@ ColumnExpansionStrategy, ColumnType, CsvTableDescriptor, + EntityUpdateFailureCode, + EntityUpdateResult, + EntityUpdateResults, FacetType, JsonSubColumn, PartialRow, @@ -84,13 +87,24 @@ QueryResultBundle, QueryResultOutput, Row, + RowReference, + RowReferenceSet, + RowReferenceSetResults, RowSet, SchemaStorageStrategy, SelectColumn, SumFileSizes, TableSchemaChangeRequest, + TableSchemaChangeResponse, + TableSearchChangeRequest, + TableSearchChangeResponse, + TableUpdateRequest, + TableUpdateResponse, TableUpdateTransaction, + UnknownTableUpdateResponse, UploadToTableRequest, + UploadToTableResult, + table_update_response_from_dict, ) from synapseclient.models.team import Team, TeamMember, TeamMembershipStatus from synapseclient.models.user import UserGroupHeader, UserPreference, UserProfile @@ -155,10 +169,26 @@ "ColumnChange", "PartialRow", "PartialRowSet", + # TableUpdateRequest models + "TableUpdateRequest", "TableSchemaChangeRequest", "AppendableRowSetRequest", "UploadToTableRequest", + "TableSearchChangeRequest", "TableUpdateTransaction", + # TableUpdateResponse models + "TableUpdateResponse", + "EntityUpdateResults", + "EntityUpdateResult", + "EntityUpdateFailureCode", + "RowReferenceSetResults", + "RowReferenceSet", + "RowReference", + "UploadToTableResult", + "TableSchemaChangeResponse", + "TableSearchChangeResponse", + "UnknownTableUpdateResponse", + "table_update_response_from_dict", "CsvTableDescriptor", "MaterializedView", "VirtualTable", diff --git a/synapseclient/models/mixins/table_components.py b/synapseclient/models/mixins/table_components.py index 708dce193..28e786771 100644 --- a/synapseclient/models/mixins/table_components.py +++ b/synapseclient/models/mixins/table_components.py @@ -75,6 +75,7 @@ SchemaStorageStrategy, SnapshotRequest, TableSchemaChangeRequest, + TableUpdateRequest, TableUpdateTransaction, UploadToTableRequest, ) @@ -2229,6 +2230,83 @@ async def _wait_for_eventually_consistent_changes( ) +def _log_upsert_summary( + entity: TableBase | ViewBase, + row_update_results: list[TableUpdateTransaction], + total_row_count_to_update: int, + row_count_to_insert: int, + client: Synapse, +) -> None: + """ + Log how many rows an upsert updated and inserted, along with any per-row + failures that Synapse reported. + + Arguments: + entity: The table or view that was upserted. + row_update_results: The results of every row update sent to Synapse. This + is empty for a dry run since nothing is sent. + total_row_count_to_update: The number of rows this client sent for update. + row_count_to_insert: The number of rows that are inserted after the update. + client: The Synapse client used for logging. + """ + total_rows_updated = sum( + result.total_rows_changed + for result in row_update_results + if result.total_rows_changed is not None + ) + + # Only the entities that back a view report a per-row outcome. A rejected row update + # on a table fails the asynchronous job and raises before this point, so for a table + # this list is always empty. + failed_row_updates = [ + failed_update + for result in row_update_results + for failed_update in result.failed_entity_updates + ] + + additional_message = "" + if failed_row_updates: + failure_details = [] + for failed_update in failed_row_updates: + failure_reason = ( + failed_update.failure_code.value + if failed_update.failure_code + else "UNKNOWN" + ) + if failed_update.failure_message: + failure_reason = f"{failure_reason}: {failed_update.failure_message}" + failure_details.append( + f"{failed_update.entity_id or 'unknown row'} ({failure_reason})" + ) + additional_message = ( + f". {len(failed_row_updates)} rows could not be updated:" + f" {'; '.join(failure_details)}" + ) + + reported_row_count_to_update = ( + total_rows_updated if row_update_results else total_row_count_to_update + ) + + client.logger.info( + f"[{entity.id}:{entity.name}]: Found {reported_row_count_to_update}" + f" rows to update and {row_count_to_insert} rows to insert" + additional_message + ) + + if ( + row_update_results + and not failed_row_updates + and total_rows_updated < total_row_count_to_update + ): + client.logger.debug( + f"[{entity.id}:{entity.name}]: Synapse confirmed" + f" {total_rows_updated} of the" + f" {total_row_count_to_update} rows sent for update and reported no" + " failure. This is a gap in how this client counts the responses it" + " received, most likely a response type it does not model, and not a" + " failed update." + ) + + async def _upsert_rows_async( entity: Union[TableBase, ViewBase], values: Union[str, Dict[str, Any], DATA_FRAME_TYPE], @@ -2301,7 +2379,7 @@ async def _upsert_rows_async( indexes_of_original_df_with_changes = [] indexes_of_original_df_with_no_changes = [] total_row_count_to_update = 0 - row_update_results = None + row_update_results: list[TableUpdateTransaction] = [] with logging_redirect_tqdm(loggers=[client.logger]): progress_bar = create_progress_bar( total=len(values), @@ -2341,13 +2419,15 @@ async def _upsert_rows_async( if syn_id_and_etag_dict: original_synids_and_etags_to_track.update(syn_id_and_etag_dict) if not dry_run and rows_to_update: - row_update_results = await _push_row_updates_to_synapse( - entity=entity, - rows_to_update=rows_to_update, - update_size_bytes=update_size_bytes, - progress_bar=progress_bar, - client=client, - job_timeout=job_timeout, + row_update_results.extend( + await _push_row_updates_to_synapse( + entity=entity, + rows_to_update=rows_to_update, + update_size_bytes=update_size_bytes, + progress_bar=progress_bar, + client=client, + job_timeout=job_timeout, + ) ) elif dry_run: progress_bar.update(len(rows_to_update)) @@ -2364,22 +2444,12 @@ async def _upsert_rows_async( ) ] - total_row_count_actually_updated = 0 - if row_update_results: - for result in row_update_results: - if result.entities_with_changes_applied: - total_row_count_actually_updated += len( - result.entities_with_changes_applied - ) - - additional_message = "" - if total_row_count_actually_updated < total_row_count_to_update: - additional_message = f". {total_row_count_to_update - total_row_count_actually_updated} rows could not be updated." - - client.logger.info( - f"[{entity.id}:{entity.name}]: Found {total_row_count_actually_updated or total_row_count_to_update}" - f" rows to update and {len(rows_to_insert_df)} rows to insert" - + additional_message + _log_upsert_summary( + entity=entity, + row_update_results=row_update_results, + total_row_count_to_update=total_row_count_to_update, + row_count_to_insert=len(rows_to_insert_df), + client=client, ) if wait_for_eventually_consistent_view and original_synids_and_etags_to_track: @@ -3394,13 +3464,7 @@ async def store_rows_async( schema_storage_strategy: SchemaStorageStrategy = None, column_expansion_strategy: ColumnExpansionStrategy = None, dry_run: bool = False, - additional_changes: List[ - Union[ - "TableSchemaChangeRequest", - "UploadToTableRequest", - "AppendableRowSetRequest", - ] - ] = None, + additional_changes: List["TableUpdateRequest"] = None, *, insert_size_bytes: int = 900 * MB, csv_table_descriptor: Optional[CsvTableDescriptor] = None, @@ -3586,7 +3650,10 @@ async def store_rows_async( what actions would be taken without actually performing them. additional_changes: Additional changes to the table that should execute - within the same transaction as appending or updating rows. This is used + within the same transaction as appending or updating rows. Each change + is a TableUpdateRequest, which is one of TableSchemaChangeRequest, + AppendableRowSetRequest, UploadToTableRequest, or + TableSearchChangeRequest. This is used as a part of the `upsert_rows` method call to allow for the updating of rows and the updating of the table schema in the same transaction. In most cases you will not need to use this argument. @@ -3898,13 +3965,7 @@ async def _send_update( table_descriptor: CsvTableDescriptor, job_timeout: int, file_handle_id: str = None, - changes: List[ - Union[ - "TableSchemaChangeRequest", - "UploadToTableRequest", - "AppendableRowSetRequest", - ] - ] = None, + changes: List["TableUpdateRequest"] = None, ) -> None: """ Construct the request to send to Synapse to update the table with the @@ -3919,6 +3980,7 @@ async def _send_update( file_handle_id: The file handle ID that is being uploaded to Synapse. changes: Additional changes to the table that should execute within the same transaction as appending or updating rows. + Each change is a TableUpdateRequest. """ all_changes = [] if changes: @@ -4020,13 +4082,7 @@ async def _stream_and_update_from_df( progress_bar: tqdm, wait_for_update_semaphore: asyncio.Semaphore, file_suffix: str, - changes: List[ - Union[ - "TableSchemaChangeRequest", - "UploadToTableRequest", - "AppendableRowSetRequest", - ] - ] = None, + changes: List["TableUpdateRequest"] = None, to_csv_kwargs: Optional[Dict[str, Any]] = None, ) -> None: """ @@ -4057,7 +4113,7 @@ async def _stream_and_update_from_df( file_suffix: The suffix that is being used to name the CSV file that is being uploaded. changes: Additional changes to the table that should - execute within this transaction. + execute within this transaction. Each change is a TableUpdateRequest. to_csv_kwargs: Additional arguments to pass to the `pd.DataFrame.to_csv` function when writing the data to a CSV file. """ @@ -4095,13 +4151,7 @@ async def _chunk_and_upload_csv( schema_change_request: TableSchemaChangeRequest, client: Synapse, job_timeout: int, - additional_changes: List[ - Union[ - "TableSchemaChangeRequest", - "UploadToTableRequest", - "AppendableRowSetRequest", - ] - ] = None, + additional_changes: List["TableUpdateRequest"] = None, ) -> None: """ Determines if the file we are appending to the table is larger than the @@ -4120,7 +4170,7 @@ async def _chunk_and_upload_csv( client: The Synapse client that is being used to interact with the API. job_timeout: The maximum amount of time to wait for a job to complete. additional_changes: Additional changes to the table that should execute - within this transaction. + within this transaction. Each change is a TableUpdateRequest. """ if (file_size := os.path.getsize(path_to_csv)) > insert_size_bytes: # Apply schema changes before breaking apart and uploading the file @@ -4242,13 +4292,7 @@ async def _chunk_and_upload_df( schema_change_request: TableSchemaChangeRequest, client: Synapse, job_timeout: int, - additional_changes: List[ - Union[ - "TableSchemaChangeRequest", - "UploadToTableRequest", - "AppendableRowSetRequest", - ] - ] = None, + additional_changes: List["TableUpdateRequest"] = None, to_csv_kwargs: Optional[Dict[str, Any]] = None, ) -> None: """ @@ -4268,9 +4312,9 @@ async def _chunk_and_upload_df( client: The Synapse client that is being used to interact with the API. job_timeout: The maximum amount of time to wait for a job to complete. additional_changes: Additional changes to the table that should execute - within this transaction. When there are multiple chunks to upload - the changes will be applied right away to prevent going over service - limits. + within this transaction. Each change is a TableUpdateRequest. When + there are multiple chunks to upload the changes will be applied right + away to prevent going over service limits. to_csv_kwargs: Additional arguments to pass to the `pd.DataFrame.to_csv` function when writing the data to a CSV file. """ diff --git a/synapseclient/models/table.py b/synapseclient/models/table.py index 4c916849f..cd2500355 100644 --- a/synapseclient/models/table.py +++ b/synapseclient/models/table.py @@ -17,7 +17,6 @@ from synapseclient.models import Activity, Annotations from synapseclient.models.mixins import AccessControllable, BaseJSONSchema from synapseclient.models.mixins.table_components import ( - AppendableRowSetRequest, ColumnExpansionStrategy, ColumnMixin, CsvTableDescriptor, @@ -27,11 +26,10 @@ SchemaStorageStrategy, TableBase, TableDeleteRowMixin, - TableSchemaChangeRequest, TableStoreMixin, TableStoreRowMixin, + TableUpdateRequest, TableUpsertMixin, - UploadToTableRequest, ) from synapseclient.models.table_components import Column @@ -402,13 +400,7 @@ def store_rows( schema_storage_strategy: SchemaStorageStrategy = None, column_expansion_strategy: ColumnExpansionStrategy = None, dry_run: bool = False, - additional_changes: List[ - Union[ - "TableSchemaChangeRequest", - "UploadToTableRequest", - "AppendableRowSetRequest", - ] - ] = None, + additional_changes: list["TableUpdateRequest"] = None, *, insert_size_bytes: int = 900 * MB, csv_table_descriptor: Optional[CsvTableDescriptor] = None, @@ -594,7 +586,10 @@ def store_rows( what actions would be taken without actually performing them. additional_changes: Additional changes to the table that should execute - within the same transaction as appending or updating rows. This is used + within the same transaction as appending or updating rows. Each change + is a TableUpdateRequest, which is one of TableSchemaChangeRequest, + AppendableRowSetRequest, UploadToTableRequest, or + TableSearchChangeRequest. This is used as a part of the `upsert_rows` method call to allow for the updating of rows and the updating of the table schema in the same transaction. In most cases you will not need to use this argument. diff --git a/synapseclient/models/table_components.py b/synapseclient/models/table_components.py index fdc68356f..2a7d003cd 100644 --- a/synapseclient/models/table_components.py +++ b/synapseclient/models/table_components.py @@ -1,5 +1,6 @@ import json import os +from abc import ABC, abstractmethod from dataclasses import dataclass, field, replace from enum import Enum from typing import TYPE_CHECKING, Any, AsyncGenerator, Dict, List, Optional, Union @@ -194,17 +195,72 @@ def to_synapse_request(self): } +class TableUpdateRequest(ABC): + """ + The abstract base class for a single change that may be included in a + TableUpdateTransaction. Every change that Synapse accepts within a transaction is + modeled by one of the concrete subclasses: + + - AppendableRowSetRequest: Add or update rows of a table, or the entities that + back a view. + - UploadToTableRequest: Apply the rows of an uploaded CSV file to a table. + - TableSchemaChangeRequest: Change the columns of a table or view. + - TableSearchChangeRequest: Enable or disable full text search on a table or view. + + | Request | Target | Response | + |----------------------------|------------------|---------------------------| + | AppendableRowSetRequest | Table | RowReferenceSetResults | + | AppendableRowSetRequest | View / Dataset | EntityUpdateResults | + | UploadToTableRequest | Table | UploadToTableResult | + | UploadToTableRequest | View / Dataset | EntityUpdateResults | + | TableSchemaChangeRequest | either | TableSchemaChangeResponse | + | TableSearchChangeRequest | either | TableSearchChangeResponse | + + A view has no rows of its own. Its rows are projections of the annotations on the + entities that back it. Writing a row of a view therefore updates those entities, + which can fail for one entity and succeed for another, so Synapse returns + EntityUpdateResults to report the status of each entity. + + This is modeled from: + + """ + + concrete_type: str + """The concrete type that identifies this change to Synapse.""" + + entity_id: str + """The Synapse ID of the entity that this change is applied to.""" + + @abstractmethod + def to_synapse_request(self) -> dict[str, Any]: + """Converts the request to a request expected of the Synapse REST API.""" + + @dataclass -class AppendableRowSetRequest: +class AppendableRowSetRequest(TableUpdateRequest): """ A request to append rows to a table. This is used to append rows to a table. This - request is used in the `TableUpdateTransaction` to indicate what rows should + request is used in the TableUpdateTransaction to indicate what rows should be upserted in the table. + + This is modeled from: + """ entity_id: str + """The Synapse ID of the table or view that the rows are appended to. Set it to the + same entity as the enclosing TableUpdateTransaction.""" + to_append: PartialRowSet - concrete_type: str = concrete_types.APPENDABLE_ROWSET_REQUEST + """The set of rows to append to the entity. Each PartialRow of the set names the + row to change by its row ID, and holds only the cells to write. A PartialRow with + no row ID adds a new row.""" + + concrete_type: str = field( + default=concrete_types.APPENDABLE_ROWSET_REQUEST, init=False + ) + """The concrete type that identifies this change to Synapse. This is always + APPENDABLE_ROWSET_REQUEST and cannot be given to the constructor.""" def to_synapse_request(self): """Converts the request to a request expected of the Synapse REST API.""" @@ -216,18 +272,69 @@ def to_synapse_request(self): @dataclass -class UploadToTableRequest: +class UploadToTableRequest(TableUpdateRequest): """ A request to upload a file to a table. This is used to insert any rows via a CSV file into a table. This request is used in the `TableUpdateTransaction`. + """ - table_id: str - upload_file_handle_id: str - update_etag: str + entity_id: str | None = None + """The Synapse ID of the entity that this change is applied to. This is the name + that every other change within a transaction uses for its target, and it is an + alias of table_id. Give either this field or table_id, and they are made equal + after the request is created.""" + + table_id: str | None = None + """The Synapse ID of the entity that this change is applied to. This is the name + that every other change within a transaction uses for its target, and it is an + alias of entity_id. Give either this field or table_id, and they are made equal + after the request is created.""" + + upload_file_handle_id: str = field(kw_only=True) + """The ID of the file handle of the CSV that holds the rows to apply. Upload the CSV + to Synapse first, with multipart_upload_file_async, and pass the file handle ID that + it returns.""" + + update_etag: str | None = None + """The etag of the change set that this update is applied to. Every RowSet that + Synapse returns carries the current etag of its change set, and that etag must be + given back to update the rows of that set. Leave this as None when the CSV only adds + rows.""" + lines_to_skip: int = 0 + """The number of lines to skip from the start of the file before the rows are read. + The default of 0 reads the file from its first line.""" + csv_table_descriptor: CsvTableDescriptor = field(default_factory=CsvTableDescriptor) - concrete_type: str = concrete_types.UPLOAD_TO_TABLE_REQUEST + """The separator, quote character, escape character, line end, and header flag that + describe the uploaded CSV. The default describes a comma separated file whose first + line is a header.""" + + concrete_type: str = field( + default=concrete_types.UPLOAD_TO_TABLE_REQUEST, init=False + ) + """The concrete type that identifies this change to Synapse. This is always + UPLOAD_TO_TABLE_REQUEST and cannot be given to the constructor.""" + + def __post_init__(self) -> None: + """Makes table_id and entity_id equal, because both name the table that the + rows are applied to.""" + if self.table_id is None and self.entity_id is None: + raise ValueError( + "Either table_id or entity_id must be given to name the table that " + "the rows of the uploaded CSV are applied to." + ) + elif self.table_id is None: + self.table_id = self.entity_id + elif self.entity_id is None: + self.entity_id = self.table_id + elif self.table_id != self.entity_id: + raise ValueError( + "table_id and entity_id both name the table that the rows of the " + "uploaded CSV are applied to, so they must be equal. Received " + f"table_id: {self.table_id}, entity_id: {self.entity_id}." + ) def to_synapse_request(self): """Converts the request to a request expected of the Synapse REST API.""" @@ -270,17 +377,36 @@ def to_synapse_request(self): @dataclass -class TableSchemaChangeRequest: +class TableSchemaChangeRequest(TableUpdateRequest): """ A request to change the schema of a table. This is used to change the columns in a table. This request is used in the `TableUpdateTransaction` to indicate what changes should be made to the columns in the table. + + This is modeled from: + """ entity_id: str - changes: List[ColumnChange] - ordered_column_ids: List[str] - concrete_type: str = concrete_types.TABLE_SCHEMA_CHANGE_REQUEST + """The Synapse ID of the table or view whose columns are changed. Set it to the same + entity as the enclosing TableUpdateTransaction.""" + + changes: list[ColumnChange] + """The list of changes that describes the column additions, deletions, and updates. + Each ColumnChange names the old column ID, the new column ID, or both. Give the new + column ID alone to add a column, the old column ID alone to remove one, and both to + replace one column with another.""" + + ordered_column_ids: list[str] + """The IDs of the columns in the order that the table presents them. This list must + hold the ID of every column that remains in the schema after the changes of this + request are applied. Synapse reads it to set the column order.""" + + concrete_type: str = field( + default=concrete_types.TABLE_SCHEMA_CHANGE_REQUEST, init=False + ) + """The concrete type that identifies this change to Synapse. This is always + TABLE_SCHEMA_CHANGE_REQUEST and cannot be given to the constructor.""" def to_synapse_request(self): """Converts the request to a request expected of the Synapse REST API.""" @@ -292,6 +418,36 @@ def to_synapse_request(self): } +@dataclass +class TableSearchChangeRequest(TableUpdateRequest): + """ + A request to change the full text search status of a table or view. This request is + used in the `TableUpdateTransaction` to enable or disable search on the entity. + + This is modeled from: + """ + + entity_id: str + """The Synapse ID of the table or view to change the search status of.""" + + search_enabled: bool + """Specifies if the search should be enabled or disabled on the table.""" + + concrete_type: str = field( + default=concrete_types.TABLE_SEARCH_CHANGE_REQUEST, init=False + ) + """The concrete type that identifies this change to Synapse. This is always + TABLE_SEARCH_CHANGE_REQUEST and cannot be given to the constructor.""" + + def to_synapse_request(self): + """Converts the request to a request expected of the Synapse REST API.""" + return { + "concreteType": self.concrete_type, + "entityId": self.entity_id, + "searchEnabled": self.search_enabled, + } + + @dataclass class SnapshotRequest: """A request that defines the options available when creating a snapshot of a table or view. @@ -321,32 +477,526 @@ def fill_from_dict(self, synapse_response: Dict[str, str]) -> "Self": return self +class EntityUpdateFailureCode(str, Enum): + """The reason an entity update within a view failed. Null when the update + succeeded. + + This is modeled from: + """ + + NOT_FOUND = "NOT_FOUND" + """The specified entity could not be found.""" + + UNAUTHORIZED = "UNAUTHORIZED" + """The user does not have permission to update the entity.""" + + CONCURRENT_UPDATE = "CONCURRENT_UPDATE" + """The entity was updated concurrently by another process.""" + + ILLEGAL_ARGUMENT = "ILLEGAL_ARGUMENT" + """The update request contained invalid or illegal arguments.""" + + UNKNOWN = "UNKNOWN" + """An unknown error occurred during the update.""" + + +@dataclass +class RowReference: + """ + A reference to a single version of a single row of a table. + + This result is modeled from: + """ + + row_id: int | None = None + """The immutable ID issued to a new row.""" + + version_number: int | None = None + """The version number of this row. Each row version is immutable, so when a row + is updated a new version is created.""" + + @classmethod + def fill_from_dict(cls, data: dict[str, Any]) -> "RowReference": + """Create a RowReference from a dictionary response.""" + return cls( + row_id=data.get("rowId", None), + version_number=data.get("versionNumber", None), + ) + + +@dataclass +class RowReferenceSet: + """ + Represents a set of RowReferences of a table. + + This result is modeled from: + """ + + table_id: str | None = None + """The ID of the table that owns these rows.""" + + etag: str | None = None + """When a RowReferenceSet is returned from a table update, this will be set to + the current etag of the table.""" + + headers: list["SelectColumn"] | None = None + """The list of SelectColumns that describes the rows of this set.""" + + rows: list[RowReference] | None = None + """Each RowReference of this list refers to a single version of a single row of + the table.""" + + @classmethod + def fill_from_dict(cls, data: dict[str, Any]) -> "RowReferenceSet": + """Create a RowReferenceSet from a dictionary response.""" + headers_data = data.get("headers", None) + rows_data = data.get("rows", None) + return cls( + table_id=data.get("tableId", None), + etag=data.get("etag", None), + headers=( + [SelectColumn.fill_from_dict(header) for header in headers_data] + if headers_data + else None + ), + rows=( + [RowReference.fill_from_dict(row) for row in rows_data] + if rows_data + else None + ), + ) + + +@dataclass +class EntityUpdateResult: + """ + The result of updating a single entity through a view. + + This result is modeled from: + """ + + entity_id: str | None = None + """The ID of the updated entity.""" + + failure_code: EntityUpdateFailureCode | None = None + """If the entity update failed, a failure code will be included. Null if the + update was successful.""" + + failure_message: str | None = None + """Failure message for unknown and illegal errors.""" + + @classmethod + def fill_from_dict(cls, data: dict[str, Any]) -> "EntityUpdateResult": + """Create an EntityUpdateResult from a dictionary response.""" + failure_code = None + failure_code_value = data.get("failureCode", None) + if failure_code_value: + try: + failure_code = EntityUpdateFailureCode(failure_code_value) + except ValueError: + failure_code = EntityUpdateFailureCode.UNKNOWN + return cls( + entity_id=data.get("entityId", None), + failure_code=failure_code, + failure_message=data.get("failureMessage", None), + ) + + @property + def succeeded(self) -> bool: + """True when Synapse reported neither a failure code nor a failure message.""" + return not self.failure_code and not self.failure_message + + +@dataclass +class TableUpdateResponse(ABC): + """ + The abstract base class for a single response within a + TableUpdateTransactionResponse. One response is returned for each change that was + included in the transaction request, in the same order as the requested changes. + + Each concrete subclass models one of the response types that Synapse may return. + Use table_update_response_from_dict to convert a response into the subclass that + models it. See TableUpdateRequest for the table that maps each request and target entity type to + the response that Synapse returns. + + This result is modeled from: + """ + + concrete_type: str | None + """The concrete type of this response, as reported by Synapse.""" + + @classmethod + @abstractmethod + def fill_from_dict(cls, data: dict[str, Any]) -> "TableUpdateResponse": + """ + Convert a response from the REST API into this dataclass. + + Arguments: + data: One element of the results array of a + TableUpdateTransactionResponse. + + Returns: + An instance of this class. + """ + + @property + def rows_changed(self) -> int | None: + """The number of rows that Synapse confirmed it applied, when the response + reports a row count. None when the response type carries no row count, as is + the case for a schema or a search change.""" + return None + + +@dataclass +class EntityUpdateResults(TableUpdateResponse): + """ + The response to a change applied to the entities that back a view. This is the + response type for a partial row set that was sent to a view. + + This result is modeled from: + """ + + concrete_type: str = field(default=concrete_types.ENTITY_UPDATE_RESULTS, init=False) + """The concrete type that identifies this change to Synapse. This is always + ENTITY_UPDATE_RESULTS and cannot be given to the constructor.""" + + update_results: list[EntityUpdateResult] | None = None + """The result of the update for each entity that was included in the change.""" + + @classmethod + def fill_from_dict(cls, data: dict[str, Any]) -> "EntityUpdateResults": + """Create an EntityUpdateResults from a dictionary response.""" + update_results_data = data.get("updateResults", None) + return cls( + update_results=( + [ + EntityUpdateResult.fill_from_dict(update_result) + for update_result in update_results_data + ] + if update_results_data + else None + ), + ) + + @property + def successful_entity_updates(self) -> list[EntityUpdateResult]: + """The updates for which Synapse reported no failure code and no failure + message.""" + return [ + update_result + for update_result in (self.update_results or []) + if update_result.succeeded + ] + + @property + def successful_entity_ids(self) -> list[str]: + """The IDs of the entities that were updated without a reported failure. This + is the subset of successful_entity_updates that reported an entity ID, so it + may be shorter than rows_changed.""" + return [ + update_result.entity_id + for update_result in self.successful_entity_updates + if update_result.entity_id + ] + + @property + def failed_entity_updates(self) -> list[EntityUpdateResult]: + """The updates for which Synapse reported a failure code or message.""" + return [ + update_result + for update_result in (self.update_results or []) + if not update_result.succeeded + ] + + @property + def rows_changed(self) -> int | None: + """The number of entities that were updated without a reported failure. An + update that succeeded without an entity ID is counted here, so this is not + always the length of successful_entity_ids.""" + return len(self.successful_entity_updates) + + +@dataclass +class RowReferenceSetResults(TableUpdateResponse): + """ + The response to a change applied to the rows of a table. This is the response + type for a partial row set that was sent to a table. + + This result is modeled from: + """ + + concrete_type: str = field( + default=concrete_types.ROW_REFERENCE_SET_RESULTS, init=False + ) + """The concrete type that identifies this change to Synapse. This is always + ROW_REFERENCE_SET_RESULTS and cannot be given to the constructor.""" + + row_reference_set: RowReferenceSet | None = None + """A reference to each row version that the change created.""" + + @classmethod + def fill_from_dict(cls, data: dict[str, Any]) -> "RowReferenceSetResults": + """Create a RowReferenceSetResults from a dictionary response.""" + row_reference_set_data = data.get("rowReferenceSet", None) + return cls( + row_reference_set=( + RowReferenceSet.fill_from_dict(row_reference_set_data) + if row_reference_set_data + else None + ), + ) + + @property + def rows_changed(self) -> int | None: + """The number of row versions that the change created. This response type + always carries a row count, so a change that created no row version reports a + confirmed 0 rather than None, whether Synapse omitted the row reference set or + returned one that holds no row.""" + if not self.row_reference_set: + return 0 + return len(self.row_reference_set.rows or []) + + +@dataclass +class UploadToTableResult(TableUpdateResponse): + """ + The result of a CSV table upload job. This is the response type for rows that + were inserted into a table from a file. + + This result is modeled from: + """ + + concrete_type: str = field( + default=concrete_types.UPLOAD_TO_TABLE_RESULT, init=False + ) + """The concrete type that identifies this change to Synapse. This is always + UPLOAD_TO_TABLE_RESULT and cannot be given to the constructor.""" + + rows_processed: int | None = None + """The number of rows that were read from the provided file and applied to the + table.""" + + etag: str | None = None + """The new etag of the version applied to the table.""" + + @classmethod + def fill_from_dict(cls, data: dict[str, Any]) -> "UploadToTableResult": + """Create an UploadToTableResult from a dictionary response.""" + return cls( + rows_processed=data.get("rowsProcessed", None), + etag=data.get("etag", None), + ) + + @property + def rows_changed(self) -> int | None: + """The number of rows that were read from the file and applied to the table.""" + return self.rows_processed + + +@dataclass +class TableSchemaChangeResponse(TableUpdateResponse): + """ + The response to a change applied to the schema of a table or view. + + This result is modeled from: + """ + + concrete_type: str = field( + default=concrete_types.TABLE_SCHEMA_CHANGE_RESPONSE, init=False + ) + """The concrete type that identifies this change to Synapse. This is always + TABLE_SCHEMA_CHANGE_RESPONSE and cannot be given to the constructor.""" + + schema: list["Column"] | None = None + """The resulting schema after the change.""" + + @classmethod + def fill_from_dict(cls, data: dict[str, Any]) -> "TableSchemaChangeResponse": + """Create a TableSchemaChangeResponse from a dictionary response.""" + schema_data = data.get("schema", None) + return cls( + schema=( + [Column().fill_from_dict(column) for column in schema_data] + if schema_data + else None + ), + ) + + +@dataclass +class TableSearchChangeResponse(TableUpdateResponse): + """ + The response to a change applied to the search status of a table or view. + + This result is modeled from: + """ + + concrete_type: str = field( + default=concrete_types.TABLE_SEARCH_CHANGE_RESPONSE, init=False + ) + """The concrete type that identifies this change to Synapse. This is always + TABLE_SEARCH_CHANGE_RESPONSE and cannot be given to the constructor.""" + + search_enabled: bool | None = None + """The resulting status of the search after the change.""" + + @classmethod + def fill_from_dict(cls, data: dict[str, Any]) -> "TableSearchChangeResponse": + """Create a TableSearchChangeResponse from a dictionary response.""" + return cls(search_enabled=data.get("searchEnabled", None)) + + +@dataclass +class UnknownTableUpdateResponse(TableUpdateResponse): + """ + A response that this version of the client does not model. The raw response is + held as-is so that no information is lost, and so that a response type added to + Synapse after this release does not fail the caller. + """ + + concrete_type: str | None = None + """The concrete type of this response as reported by Synapse, or None when the + response did not report one.""" + + data: dict[str, Any] | None = None + """The raw response as it was returned by Synapse.""" + + @classmethod + def fill_from_dict(cls, data: dict[str, Any]) -> "UnknownTableUpdateResponse": + """Create an UnknownTableUpdateResponse from a dictionary response.""" + return cls( + concrete_type=data.get("concreteType", None), + data=data, + ) + + +_TABLE_UPDATE_RESPONSE_TYPES: dict[str, type] = { + concrete_types.ENTITY_UPDATE_RESULTS: EntityUpdateResults, + concrete_types.ROW_REFERENCE_SET_RESULTS: RowReferenceSetResults, + concrete_types.UPLOAD_TO_TABLE_RESULT: UploadToTableResult, + concrete_types.TABLE_SCHEMA_CHANGE_RESPONSE: TableSchemaChangeResponse, + concrete_types.TABLE_SEARCH_CHANGE_RESPONSE: TableSearchChangeResponse, +} +"""Maps the concrete type reported by Synapse to the class that models it.""" + + +def table_update_response_from_dict(data: dict[str, Any]) -> TableUpdateResponse: + """ + Convert a single response of a TableUpdateTransactionResponse into the dataclass + that models it. Dispatch is on the concrete type that Synapse reported. A response + that reports no concrete type, or one that this release does not model, is held as + an UnknownTableUpdateResponse rather than guessed at from the keys it carries. + + Arguments: + data: One element of the results array of a + + + Returns: + An instance of the matching TableUpdateResponse subclass, or an + UnknownTableUpdateResponse holding the raw response when the response + cannot be identified. + """ + response_class = _TABLE_UPDATE_RESPONSE_TYPES.get(data.get("concreteType", None)) + + if response_class is None: + response_class = UnknownTableUpdateResponse + + return response_class.fill_from_dict(data) + + @dataclass class TableUpdateTransaction(AsynchronousCommunicator): """ A request to update a table. This is used to update a table with a set of changes. + - After calling the `send_job_and_wait_async` method the `results` attribute will be - filled in based off . + After calling the send_job_and_wait_async method the results attribute will + be filled in based off + . """ entity_id: str + """The Synapse ID of the table or view that this transaction is applied to. Every + change within the transaction must be applied to this same entity.""" concrete_type: str = concrete_types.TABLE_UPDATE_TRANSACTION_REQUEST + """The concrete type that identifies this request to Synapse. Leave this as the + default.""" create_snapshot: bool = False - changes: Optional[ - List[ - Union[ - TableSchemaChangeRequest, UploadToTableRequest, AppendableRowSetRequest - ] + """Whether Synapse creates a new version of the entity once every change in this + transaction has been applied. Set this to True to capture the state that the + transaction produced as an immutable version, and use snapshot_options to label + that version. The version number that Synapse assigns is reported in + snapshot_version_number.""" + changes: list[TableUpdateRequest] | None = None + """The changes to apply within this transaction, in the order that they should be + applied.""" + snapshot_options: SnapshotRequest | None = None + """The label, comment, and activity to attach to the snapshot that this transaction + creates. Sent to Synapse only when create_snapshot is True, and ignored otherwise. + Leave this as None to let Synapse pick the defaults for the new version.""" + results: list[TableUpdateResponse] | None = None + """The responses that Synapse returned for this transaction, as the dataclass that + models each one. One response per change that was included in the request, in the + same order as the requested changes. Each is a subclass of + .""" + snapshot_version_number: int | None = None + """The version number of the snapshot that Synapse created for this transaction. + This is filled in only when create_snapshot was True, and stays None otherwise. Use + it to read the entity at the version that was just captured.""" + + @property + def entities_with_changes_applied(self) -> list[str] | None: + """The Synapse IDs of the entities that Synapse confirmed it applied a change + to. This is derived from the update results of a view transaction, and holds + only the entities that reported no failure code and no failure message. An + entity that Synapse rejected is left out. This is None when the transaction + reported no such entity, which is the case for a transaction against the rows + of a table.""" + if self.results is None: + return None + + successful_entities = [ + entity_id + for response in self.results + if isinstance(response, EntityUpdateResults) + for entity_id in response.successful_entity_ids ] - ] = None - snapshot_options: Optional[SnapshotRequest] = None - results: Optional[List[Dict[str, Any]]] = None - snapshot_version_number: Optional[int] = None - entities_with_changes_applied: Optional[List[str]] = None + return successful_entities or None - """This will be an array of - .""" + @property + def failed_entity_updates(self) -> list[EntityUpdateResult]: + """The updates that Synapse rejected, as the result it reported for each one. + Each holds the entity ID together with the failure code and the failure message + that Synapse gave, so the caller can report why the update did not apply. This + is derived from the update results of a view transaction. It is always empty for + a transaction against the rows of a table, because a rejected row update fails + the asynchronous job and raises instead of reporting a per-row failure.""" + if self.results is None: + return [] + + return [ + failed_update + for response in self.results + if isinstance(response, EntityUpdateResults) + for failed_update in response.failed_entity_updates + ] + + @property + def total_rows_changed(self) -> int | None: + """The total number of rows that Synapse confirmed it changed for this + transaction, derived from the modelled responses in results. The count of every + change that reports one is added together, whether the change was applied to + the rows of a table or to the entities that back a view. A change that carries + no row count, such as a schema or a search change, contributes nothing. This is + None until the transaction has been sent, and is 0 when no row was changed.""" + if self.results is None: + return None + + return sum( + response.rows_changed + for response in self.results + if response.rows_changed is not None + ) def to_synapse_request(self): """Converts the request to a request expected of the Synapse REST API.""" @@ -365,33 +1015,27 @@ def to_synapse_request(self): return request - def fill_from_dict(self, synapse_response: Dict[str, str]) -> "Self": + def fill_from_dict(self, synapse_response: dict[str, Any]) -> "Self": """ Converts a response from the REST API into this dataclass. Arguments: - synapse_response: The response from the REST API that matches + synapse_response: The response from the REST API that matches + Returns: An instance of this class. """ - self.results = synapse_response.get("results", None) self.snapshot_version_number = synapse_response.get( "snapshotVersionNumber", None ) - if "results" in synapse_response: - successful_entities = [] - for result in synapse_response["results"]: - if "updateResults" in result: - for update_result in result["updateResults"]: - failure_code = update_result.get("failureCode", None) - failure_message = update_result.get("failureMessage", None) - entity_id = update_result.get("entityId", None) - if not failure_code and not failure_message and entity_id: - successful_entities.append(entity_id) - if successful_entities: - self.entities_with_changes_applied = successful_entities + results_data = synapse_response.get("results", None) + self.results = ( + [table_update_response_from_dict(result) for result in results_data] + if results_data is not None + else None + ) return self @@ -673,16 +1317,10 @@ class SelectColumn: @classmethod def fill_from_dict(cls, data: Dict[str, Any]) -> "SelectColumn": """Create a SelectColumn from a dictionary response.""" - column_type = None - column_type_value = data.get("columnType") - if column_type_value: - try: - column_type = ColumnType(column_type_value) - except ValueError: - column_type = None + column_type = data.get("columnType") return cls( name=data.get("name"), - column_type=column_type, + column_type=ColumnType(column_type) if column_type else None, id=data.get("id"), ) @@ -1068,19 +1706,13 @@ class JsonSubColumn: @classmethod def fill_from_dict(cls, synapse_sub_column: Dict[str, Any]) -> "JsonSubColumn": """Converts a response from the synapseclient into this dataclass.""" + column_type = synapse_sub_column.get("columnType", None) + facet_type = synapse_sub_column.get("facetType", None) return cls( name=synapse_sub_column.get("name", ""), - column_type=( - ColumnType(synapse_sub_column.get("columnType", None)) - if synapse_sub_column.get("columnType", None) - else ColumnType.STRING - ), + column_type=(ColumnType(column_type) if column_type else ColumnType.STRING), json_path=synapse_sub_column.get("jsonPath", ""), - facet_type=( - FacetType(synapse_sub_column.get("facetType", None)) - if synapse_sub_column.get("facetType", None) - else None - ), + facet_type=FacetType(facet_type) if facet_type else None, ) def to_synapse_request(self) -> Dict[str, Any]: @@ -1267,16 +1899,10 @@ def fill_from_dict( """Converts a response from the synapseclient into this dataclass.""" self.id = synapse_column.get("id", None) self.name = synapse_column.get("name", None) - self.column_type = ( - ColumnType(synapse_column.get("columnType", None)) - if synapse_column.get("columnType", None) - else None - ) - self.facet_type = ( - FacetType(synapse_column.get("facetType", None)) - if synapse_column.get("facetType", None) - else None - ) + column_type = synapse_column.get("columnType", None) + self.column_type = ColumnType(column_type) if column_type else None + facet_type = synapse_column.get("facetType", None) + self.facet_type = FacetType(facet_type) if facet_type else None self.default_value = synapse_column.get("defaultValue", None) self.maximum_size = synapse_column.get("maximumSize", None) self.maximum_list_length = synapse_column.get("maximumListLength", None) diff --git a/tests/integration/synapseclient/models/async/test_table_async.py b/tests/integration/synapseclient/models/async/test_table_async.py index 5b4fedfc0..8df960313 100644 --- a/tests/integration/synapseclient/models/async/test_table_async.py +++ b/tests/integration/synapseclient/models/async/test_table_async.py @@ -1,11 +1,13 @@ import json +import logging import os import random import re import string import tempfile import uuid -from typing import Callable +from contextlib import contextmanager +from typing import Callable, Iterator from unittest import skip import pandas as pd @@ -35,6 +37,36 @@ from tests.integration import QUERY_TIMEOUT_SEC +class _MessageCollectingHandler(logging.Handler): + """Stores the messages that are written to a logger.""" + + def __init__(self) -> None: + super().__init__() + self.messages: list[str] = [] + + def emit(self, record: logging.LogRecord) -> None: + self.messages.append(record.getMessage()) + + +@contextmanager +def capture_client_logs(syn: Synapse) -> Iterator[list[str]]: + """Collect the messages the client logs inside this block. + + A handler is attached directly to the client logger because the logger used + during the tests is silent and does not propagate to the root logger, which + is what the caplog fixture reads. + """ + handler = _MessageCollectingHandler() + original_level = syn.logger.level + syn.logger.addHandler(handler) + syn.logger.setLevel(logging.INFO) + try: + yield handler.messages + finally: + syn.logger.removeHandler(handler) + syn.logger.setLevel(original_level) + + class TestTableCreation: @pytest.fixture(autouse=True, scope="function") def init(self, syn: Synapse, schedule_for_cleanup: Callable[..., None]) -> None: @@ -1556,6 +1588,78 @@ async def test_upsert_with_large_data_and_batching( # AND multiple batch jobs should have been created due to batching settings assert spy_send_job.call_count == 7 # More batches due to small size settings + @pytest.mark.parametrize( + "rows_per_query", + [50000, 2], + ids=["single_query_chunk", "multiple_query_chunks"], + ) + async def test_upsert_reports_accurate_row_counts( + self, project_model: Project, rows_per_query: int + ) -> None: + """An upsert that fully succeeds must not report any failed updates. + + The response Synapse returns for a Table row update is a + RowReferenceSetResults, which carries row references rather than the + entityId/updateResults pairs that a View update returns. When the client + cannot account for the rows it updated it wrongly reports that every + updated row failed, even though the data was stored. + + The multiple_query_chunks case additionally covers the accumulation of + results across query chunks. + """ + # GIVEN a table in Synapse holding five rows + table = Table( + name=str(uuid.uuid4()), + parent_id=project_model.id, + columns=[ + Column(name="key", column_type=ColumnType.STRING), + Column(name="value", column_type=ColumnType.STRING), + ], + ) + table = await table.store_async(synapse_client=self.syn) + self.schedule_for_cleanup(table.id) + + await table.store_rows_async( + values=pd.DataFrame( + {"key": ["a", "b", "c", "d", "e"], "value": ["before"] * 5} + ), + schema_storage_strategy=None, + synapse_client=self.syn, + ) + + # WHEN I upsert changes for all five rows plus two new rows + with capture_client_logs(self.syn) as log_messages: + await table.upsert_rows_async( + values=pd.DataFrame( + { + "key": ["a", "b", "c", "d", "e", "f", "g"], + "value": ["after"] * 5 + ["new", "new"], + } + ), + primary_keys=["key"], + rows_per_query=rows_per_query, + synapse_client=self.syn, + ) + + # THEN every change is stored in the table + results = await query_async( + f"SELECT key, value FROM {table.id} ORDER BY key", + synapse_client=self.syn, + ) + assert len(results) == 7 + assert results["value"].tolist() == ["after"] * 5 + ["new", "new"] + + # AND the client reports the counts it actually applied + upsert_messages = [ + message for message in log_messages if "rows to update" in message + ] + assert len(upsert_messages) == 1 + upsert_message = upsert_messages[0] + + # AND no row is reported as having failed to update + assert "could not be updated" not in upsert_message + assert "Found 5 rows to update and 2 rows to insert" in upsert_message + async def test_upsert_all_data_types(self, project_model: Project) -> None: """Test upserting all supported data types to ensure type compatibility.""" # GIVEN a table in Synapse with all data types diff --git a/tests/unit/synapseclient/mixins/unit_test_table_components.py b/tests/unit/synapseclient/mixins/unit_test_table_components.py index e10270dad..3ebae5752 100644 --- a/tests/unit/synapseclient/mixins/unit_test_table_components.py +++ b/tests/unit/synapseclient/mixins/unit_test_table_components.py @@ -4,7 +4,7 @@ from dataclasses import dataclass, field from io import BytesIO from typing import Any, Dict, List, Optional -from unittest.mock import MagicMock, patch +from unittest.mock import AsyncMock, MagicMock, patch import numpy as np import pandas as pd @@ -14,9 +14,18 @@ from synapseclient import Synapse from synapseclient.api import ViewEntityType, ViewTypeMask from synapseclient.core.constants.concrete_types import ( + APPENDABLE_ROWSET_REQUEST, + ENTITY_UPDATE_RESULTS, QUERY_BUNDLE_REQUEST, QUERY_RESULT, QUERY_TABLE_CSV_REQUEST, + ROW_REFERENCE_SET_RESULTS, + TABLE_SCHEMA_CHANGE_REQUEST, + TABLE_SCHEMA_CHANGE_RESPONSE, + TABLE_SEARCH_CHANGE_REQUEST, + TABLE_SEARCH_CHANGE_RESPONSE, + UPLOAD_TO_TABLE_REQUEST, + UPLOAD_TO_TABLE_RESULT, ) from synapseclient.core.utils import MB from synapseclient.models import Activity, Column @@ -31,6 +40,7 @@ TableStoreMixin, TableUpdateTransaction, TableUpsertMixin, + ViewBase, ViewSnapshotMixin, ViewStoreMixin, ViewUpdateMixin, @@ -40,18 +50,26 @@ _construct_select_statement_for_upsert, _construct_single_key_where_statement, _format_primary_key_value_for_where, + _log_upsert_summary, _query_table_csv, _query_table_next_page, _query_table_row_set, + _upsert_rows_async, _validate_primary_keys, convert_dtypes_to_json_serializable, csv_to_pandas_df, ) from synapseclient.models.table_components import ( ActionRequiredCount, + AppendableRowSetRequest, + ColumnChange, ColumnType, CsvTableDescriptor, + EntityUpdateFailureCode, + EntityUpdateResult, + EntityUpdateResults, PartialRow, + PartialRowSet, Query, QueryBundleRequest, QueryJob, @@ -60,9 +78,22 @@ QueryResultBundle, QueryResultOutput, Row, + RowReference, + RowReferenceSet, + RowReferenceSetResults, RowSet, SelectColumn, SumFileSizes, + TableSchemaChangeRequest, + TableSchemaChangeResponse, + TableSearchChangeRequest, + TableSearchChangeResponse, + TableUpdateRequest, + TableUpdateResponse, + UnknownTableUpdateResponse, + UploadToTableRequest, + UploadToTableResult, + table_update_response_from_dict, ) POST_COLUMNS_PATCH = "synapseclient.models.mixins.table_components.post_columns" @@ -81,6 +112,9 @@ _UPSERT_ROWS_ASYNC_PATCH = ( "synapseclient.models.mixins.table_components._upsert_rows_async" ) +_PUSH_ROW_UPDATES_TO_SYNAPSE_PATCH = ( + "synapseclient.models.mixins.table_components._push_row_updates_to_synapse" +) DEFAULT_QUOTE_CHARACTER = '"' DEFAULT_SEPARATOR = "," DEFAULT_ESCAPE_CHAR = "\\" @@ -2500,6 +2534,304 @@ def test_null_primary_keys_raise(self, data, primary_keys, expected_columns): assert f"'{column}'" not in message +class TestLogUpsertSummary: + """Test suite for the _log_upsert_summary function.""" + + @pytest.fixture(autouse=True, scope="function") + def init_syn(self, syn: Synapse) -> None: + self.syn = syn + + @dataclass + class ClassForTest: + id: Optional[str] = "syn123" + name: Optional[str] = "test_table" + + @staticmethod + def _table_transaction(rows_changed: int) -> TableUpdateTransaction: + """A transaction against the rows of a table that changed the given row count.""" + return TableUpdateTransaction( + entity_id="syn123", + results=[ + RowReferenceSetResults( + row_reference_set=RowReferenceSet( + rows=[ + RowReference(row_id=index, version_number=1) + for index in range(rows_changed) + ] + ) + ) + ], + ) + + @staticmethod + def _view_transaction( + update_results: List[EntityUpdateResult], + ) -> TableUpdateTransaction: + """A transaction against the entities that back a view.""" + return TableUpdateTransaction( + entity_id="syn123", + results=[EntityUpdateResults(update_results=update_results)], + ) + + def test_no_results_reports_the_client_side_count(self): + # GIVEN no results, as is the case for a dry run + test_instance = self.ClassForTest() + with ( + patch.object(self.syn.logger, "info") as mock_logger_info, + patch.object(self.syn.logger, "debug") as mock_logger_debug, + ): + # WHEN I log the summary + _log_upsert_summary( + entity=test_instance, + row_update_results=[], + total_row_count_to_update=5, + row_count_to_insert=2, + client=self.syn, + ) + + # THEN the count this client sent for update is reported + mock_logger_info.assert_called_once_with( + "[syn123:test_table]: Found 5 rows to update and 2 rows to insert" + ) + # AND no gap is reported, because Synapse confirmed nothing + mock_logger_debug.assert_not_called() + + def test_results_report_the_count_synapse_confirmed(self): + # GIVEN results that confirm every row this client sent + test_instance = self.ClassForTest() + with ( + patch.object(self.syn.logger, "info") as mock_logger_info, + patch.object(self.syn.logger, "debug") as mock_logger_debug, + ): + # WHEN I log the summary + _log_upsert_summary( + entity=test_instance, + row_update_results=[ + self._table_transaction(2), + self._table_transaction(1), + ], + total_row_count_to_update=3, + row_count_to_insert=0, + client=self.syn, + ) + + # THEN the counts from every result are added together + mock_logger_info.assert_called_once_with( + "[syn123:test_table]: Found 3 rows to update and 0 rows to insert" + ) + # AND no gap is reported + mock_logger_debug.assert_not_called() + + @pytest.mark.parametrize( + "results,expected_count", + [ + # A transaction that has not been sent carries no count. + ([TableUpdateTransaction(entity_id="syn123", results=None)], 0), + # A schema change reports no row count. + ( + [ + TableUpdateTransaction( + entity_id="syn123", + results=[TableSchemaChangeResponse(schema=[])], + ) + ], + 0, + ), + ], + ids=["unsent_transaction", "schema_change_only"], + ) + def test_results_without_a_row_count_contribute_nothing( + self, results: List[TableUpdateTransaction], expected_count: int + ): + # GIVEN results that carry no row count + test_instance = self.ClassForTest() + with patch.object(self.syn.logger, "info") as mock_logger_info: + # WHEN I log the summary + _log_upsert_summary( + entity=test_instance, + row_update_results=results, + total_row_count_to_update=4, + row_count_to_insert=0, + client=self.syn, + ) + + # THEN those results are left out of the confirmed count + mock_logger_info.assert_called_once_with( + f"[syn123:test_table]: Found {expected_count} rows to update" + " and 0 rows to insert" + ) + + @pytest.mark.parametrize( + "failed_update,expected_detail", + [ + ( + EntityUpdateResult( + entity_id="syn456", + failure_code=EntityUpdateFailureCode.UNAUTHORIZED, + ), + "syn456 (UNAUTHORIZED)", + ), + ( + EntityUpdateResult( + entity_id="syn456", + failure_code=EntityUpdateFailureCode.ILLEGAL_ARGUMENT, + failure_message="bad value", + ), + "syn456 (ILLEGAL_ARGUMENT: bad value)", + ), + # Synapse reported a message without a code. + ( + EntityUpdateResult(entity_id="syn456", failure_message="bad value"), + "syn456 (UNKNOWN: bad value)", + ), + # Synapse reported a failure without naming the entity. + ( + EntityUpdateResult( + failure_code=EntityUpdateFailureCode.NOT_FOUND, + ), + "unknown row (NOT_FOUND)", + ), + ], + ids=["code_only", "code_and_message", "message_only", "no_entity_id"], + ) + def test_a_failed_row_update_is_described( + self, failed_update: EntityUpdateResult, expected_detail: str + ): + # GIVEN a view result that holds one failed update + test_instance = self.ClassForTest() + with patch.object(self.syn.logger, "info") as mock_logger_info: + # WHEN I log the summary + _log_upsert_summary( + entity=test_instance, + row_update_results=[self._view_transaction([failed_update])], + total_row_count_to_update=1, + row_count_to_insert=0, + client=self.syn, + ) + + # THEN the failure is described with the reason Synapse gave + mock_logger_info.assert_called_once_with( + "[syn123:test_table]: Found 0 rows to update and 0 rows to insert." + f" 1 rows could not be updated: {expected_detail}" + ) + + def test_failed_row_updates_from_every_result_are_reported(self): + # GIVEN two results that each hold a failed update alongside a successful one + test_instance = self.ClassForTest() + with ( + patch.object(self.syn.logger, "info") as mock_logger_info, + patch.object(self.syn.logger, "debug") as mock_logger_debug, + ): + # WHEN I log the summary + _log_upsert_summary( + entity=test_instance, + row_update_results=[ + self._view_transaction( + [ + EntityUpdateResult(entity_id="syn1"), + EntityUpdateResult( + entity_id="syn2", + failure_code=EntityUpdateFailureCode.NOT_FOUND, + ), + ] + ), + self._view_transaction( + [ + EntityUpdateResult( + entity_id="syn3", + failure_code=EntityUpdateFailureCode.CONCURRENT_UPDATE, + ), + ] + ), + ], + total_row_count_to_update=3, + row_count_to_insert=0, + client=self.syn, + ) + + # THEN only the successful update is counted, and both failures are listed + mock_logger_info.assert_called_once_with( + "[syn123:test_table]: Found 1 rows to update and 0 rows to insert." + " 2 rows could not be updated: syn2 (NOT_FOUND);" + " syn3 (CONCURRENT_UPDATE)" + ) + # AND the gap is not reported as an accounting gap, because the failures + # already explain it + mock_logger_debug.assert_not_called() + + def test_a_gap_without_a_reported_failure_is_logged_as_an_accounting_gap(self): + # GIVEN a result that confirms fewer rows than this client sent, with no failure + test_instance = self.ClassForTest() + with ( + patch.object(self.syn.logger, "info") as mock_logger_info, + patch.object(self.syn.logger, "debug") as mock_logger_debug, + ): + # WHEN I log the summary + _log_upsert_summary( + entity=test_instance, + row_update_results=[self._table_transaction(1)], + total_row_count_to_update=3, + row_count_to_insert=0, + client=self.syn, + ) + + # THEN the confirmed count is reported + mock_logger_info.assert_called_once_with( + "[syn123:test_table]: Found 1 rows to update and 0 rows to insert" + ) + # AND the gap is called out as a gap in this client, not a failed update + mock_logger_debug.assert_called_once() + debug_message = mock_logger_debug.call_args.args[0] + assert "Synapse confirmed 1 of the 3 rows sent for update" in debug_message + assert "not a failed update" in debug_message + + def test_a_success_with_no_entity_id_is_not_reported_as_a_gap(self): + # GIVEN a view result where every row applied, but one reported no entity ID + test_instance = self.ClassForTest() + with ( + patch.object(self.syn.logger, "info") as mock_logger_info, + patch.object(self.syn.logger, "debug") as mock_logger_debug, + ): + # WHEN I log the summary + _log_upsert_summary( + entity=test_instance, + row_update_results=[ + self._view_transaction( + [ + EntityUpdateResult(entity_id="syn1"), + EntityUpdateResult(entity_id=None), + ] + ) + ], + total_row_count_to_update=2, + row_count_to_insert=0, + client=self.syn, + ) + + # THEN both rows are counted as updated + mock_logger_info.assert_called_once_with( + "[syn123:test_table]: Found 2 rows to update and 0 rows to insert" + ) + # AND no accounting gap is reported, because nothing was lost + mock_logger_debug.assert_not_called() + + def test_more_rows_confirmed_than_sent_is_not_a_gap(self): + # GIVEN a result that confirms at least as many rows as this client sent + test_instance = self.ClassForTest() + with patch.object(self.syn.logger, "debug") as mock_logger_debug: + # WHEN I log the summary + _log_upsert_summary( + entity=test_instance, + row_update_results=[self._table_transaction(4)], + total_row_count_to_update=3, + row_count_to_insert=0, + client=self.syn, + ) + + # THEN no gap is reported + mock_logger_debug.assert_not_called() + + class TestQuery: """Test suite for the Query.to_synapse_request method.""" @@ -5028,3 +5360,1234 @@ def test_nullable_int64_with_pd_na(self): ).convert_dtypes() pd.testing.assert_frame_equal(result, expected_result, check_dtype=False) assert is_object_dtype(result.nullable_int_col) + + +def _row_reference_set_results(row_count: int) -> Dict[str, Any]: + """A RowReferenceSetResults response, as Synapse returns it for the update half of + a table upsert. Modeled on the response recorded from production for SYNPY-1912.""" + return { + "concreteType": ROW_REFERENCE_SET_RESULTS, + "rowReferenceSet": { + "tableId": "syn76890550", + "etag": "5aac0c05-c0dc-4119-b284-4c394a6044aa", + "rows": [ + {"rowId": row_id, "versionNumber": 2} + for row_id in range(1, row_count + 1) + ], + }, + } + + +def _entity_update_results(update_results: List[Dict[str, Any]]) -> Dict[str, Any]: + """An EntityUpdateResults response, as Synapse returns it for a change applied to + the entities that back a view.""" + return { + "concreteType": ENTITY_UPDATE_RESULTS, + "updateResults": update_results, + } + + +class TestTableUpdateResponseFromDict: + """Test suite for the table_update_response_from_dict dispatch function.""" + + @pytest.mark.parametrize( + "concrete_type,expected_class", + [ + (ENTITY_UPDATE_RESULTS, EntityUpdateResults), + (ROW_REFERENCE_SET_RESULTS, RowReferenceSetResults), + (UPLOAD_TO_TABLE_RESULT, UploadToTableResult), + (TABLE_SCHEMA_CHANGE_RESPONSE, TableSchemaChangeResponse), + (TABLE_SEARCH_CHANGE_RESPONSE, TableSearchChangeResponse), + ], + ids=[ + "entity_update_results", + "row_reference_set_results", + "upload_to_table_result", + "table_schema_change_response", + "table_search_change_response", + ], + ) + def test_dispatch_on_known_concrete_type(self, concrete_type, expected_class): + """Each concrete type that Synapse reports maps to the class that models it.""" + # GIVEN a response that reports a concrete type we model + data = {"concreteType": concrete_type} + + # WHEN converting it + response = table_update_response_from_dict(data) + + # THEN it is the matching subclass and the reported type is kept + assert isinstance(response, expected_class) + assert response.concrete_type == concrete_type + + @pytest.mark.parametrize( + "data", + [ + {"updateResults": []}, + {"rowReferenceSet": {}}, + {"rowsProcessed": 0}, + {"schema": []}, + {"searchEnabled": True}, + ], + ids=[ + "update_results_key", + "row_reference_set_key", + "rows_processed_key", + "schema_key", + "search_enabled_key", + ], + ) + def test_a_response_with_no_concrete_type_is_unknown(self, data): + """A response is identified only by the concrete type Synapse reports. The keys + it carries are not used to guess a type, so a response with no concrete type is + held as-is rather than reported as the type it resembles.""" + # GIVEN a response that reports no concrete type + # WHEN converting it + response = table_update_response_from_dict(data) + + # THEN it is unknown, the raw response is kept, and no row count is claimed + assert isinstance(response, UnknownTableUpdateResponse) + assert response.concrete_type is None + assert response.data == data + assert response.rows_changed is None + + def test_unrecognized_concrete_type_is_unknown_and_keeps_the_reported_type(self): + """A concrete type added to Synapse after this release does not raise, and the + type Synapse reported is preserved so that the response can be identified from + the raw data.""" + # GIVEN a response with an unmodelled concrete type + data = { + "concreteType": "org.sagebionetworks.repo.model.table.FutureResponse", + "rowReferenceSet": { + "tableId": "syn123", + "rows": [{"rowId": 1, "versionNumber": 2}], + }, + } + + # WHEN converting it + response = table_update_response_from_dict(data) + + # THEN the raw response is held as-is and the reported type is kept + assert isinstance(response, UnknownTableUpdateResponse) + assert response.data == data + assert ( + response.concrete_type + == "org.sagebionetworks.repo.model.table.FutureResponse" + ) + assert response.rows_changed is None + + def test_unidentifiable_response_is_unknown_and_keeps_the_raw_data(self): + """A response type added to Synapse after this release neither raises nor is + miscounted.""" + # GIVEN a response that can be identified neither by type nor by key + data = { + "concreteType": "org.sagebionetworks.repo.model.table.NewResponse", + "somethingNew": 5, + } + + # WHEN converting it + response = table_update_response_from_dict(data) + + # THEN the raw response is held as-is and it reports no row count + assert isinstance(response, UnknownTableUpdateResponse) + assert response.data == data + assert ( + response.concrete_type == "org.sagebionetworks.repo.model.table.NewResponse" + ) + assert response.rows_changed is None + + def test_empty_response_does_not_raise(self): + """An empty response is unknown rather than an error.""" + # GIVEN an empty response + # WHEN converting it + response = table_update_response_from_dict({}) + + # THEN it is unknown, with no concrete type and no row count + assert isinstance(response, UnknownTableUpdateResponse) + assert response.concrete_type is None + assert response.data == {} + assert response.rows_changed is None + + def test_abstract_base_class_cannot_be_instantiated(self): + """TableUpdateResponse only exists to be subclassed.""" + # GIVEN the abstract base class + # WHEN instantiating it THEN it raises + with pytest.raises(TypeError): + TableUpdateResponse() + + +class TestTableUpdateResponseRowsChanged: + """Test suite for the rows_changed property of each TableUpdateResponse subclass. + + rows_changed is the source of the row count that upsert_rows reports, so the + difference between a confirmed count of 0 and an absent count of None matters. + """ + + @pytest.mark.parametrize( + "data,expected_rows_changed", + [ + # The update half of a table upsert. + (_row_reference_set_results(5), 5), + ( + { + "concreteType": ROW_REFERENCE_SET_RESULTS, + "rowReferenceSet": {"tableId": "syn123", "rows": []}, + }, + 0, + ), + ({"concreteType": ROW_REFERENCE_SET_RESULTS, "rowReferenceSet": {}}, 0), + ({"concreteType": ROW_REFERENCE_SET_RESULTS}, 0), + # The insert half of a table upsert. + ({"concreteType": UPLOAD_TO_TABLE_RESULT, "rowsProcessed": 2}, 2), + ({"concreteType": UPLOAD_TO_TABLE_RESULT, "rowsProcessed": 0}, 0), + ({"concreteType": UPLOAD_TO_TABLE_RESULT}, None), + # A change applied to the entities that back a view. + ( + _entity_update_results( + [ + {"entityId": "syn1"}, + {"entityId": "syn2", "failureCode": "NOT_FOUND"}, + { + "entityId": "syn3", + "failureCode": "ILLEGAL_ARGUMENT", + "failureMessage": "bad value", + }, + ] + ), + 1, + ), + (_entity_update_results([]), 0), + ({"concreteType": ENTITY_UPDATE_RESULTS}, 0), + # Changes that apply no rows. + ( + { + "concreteType": TABLE_SCHEMA_CHANGE_RESPONSE, + "schema": [{"name": "col1", "columnType": "STRING"}], + }, + None, + ), + ( + {"concreteType": TABLE_SEARCH_CHANGE_RESPONSE, "searchEnabled": True}, + None, + ), + ], + ids=[ + "row_reference_set_five_rows", + "row_reference_set_no_rows_is_zero", + "row_reference_set_empty_is_zero", + "row_reference_set_absent_is_zero", + "upload_to_table_two_rows", + "upload_to_table_zero_rows_is_zero", + "upload_to_table_absent_count_is_none", + "entity_update_one_success_two_failures", + "entity_update_empty_is_zero", + "entity_update_absent_is_zero", + "schema_change_is_none", + "search_change_is_none", + ], + ) + def test_rows_changed(self, data, expected_rows_changed): + """Only a response that reports a row count contributes one.""" + # GIVEN a response from Synapse + # WHEN converting it + response = table_update_response_from_dict(data) + + # THEN rows_changed reports the confirmed count, and None when the response + # carries no count at all + assert response.rows_changed == expected_rows_changed + + def test_row_reference_set_results_fields(self): + """The full RowReferenceSetResults response is modeled, not just its count.""" + # GIVEN the response recorded from production for the update half of an upsert + # WHEN converting it + response = table_update_response_from_dict(_row_reference_set_results(5)) + + # THEN the row references and the table etag are available + assert response.row_reference_set.table_id == "syn76890550" + assert response.row_reference_set.etag == "5aac0c05-c0dc-4119-b284-4c394a6044aa" + assert response.row_reference_set.rows == [ + RowReference(row_id=row_id, version_number=2) for row_id in range(1, 6) + ] + + def test_row_reference_set_parses_headers(self): + """The optional headers of a RowReferenceSet are modeled as SelectColumns.""" + # GIVEN a row reference set that carries headers + data = { + "tableId": "syn123", + "headers": [{"name": "col1", "columnType": "STRING", "id": "1"}], + "rows": [{"rowId": 1, "versionNumber": 2}], + } + + # WHEN converting it + row_reference_set = RowReferenceSet.fill_from_dict(data) + + # THEN the headers are SelectColumn instances + assert row_reference_set.headers == [ + SelectColumn(name="col1", column_type=ColumnType.STRING, id="1") + ] + + def test_upload_to_table_result_keeps_the_etag(self): + """The etag of the version applied to the table is retained.""" + # GIVEN an upload result + data = { + "concreteType": UPLOAD_TO_TABLE_RESULT, + "rowsProcessed": 2, + "etag": "new-etag", + } + + # WHEN converting it + response = table_update_response_from_dict(data) + + # THEN the etag is available + assert response.etag == "new-etag" + + def test_schema_change_response_parses_columns(self): + """The resulting schema is modeled as Column instances.""" + # GIVEN a schema change response + data = { + "concreteType": TABLE_SCHEMA_CHANGE_RESPONSE, + "schema": [ + {"name": "col1", "columnType": "STRING", "id": "1"}, + {"name": "col2", "columnType": "INTEGER", "id": "2"}, + ], + } + + # WHEN converting it + response = table_update_response_from_dict(data) + + # THEN each column of the resulting schema is available + assert [column.name for column in response.schema] == ["col1", "col2"] + assert [column.column_type for column in response.schema] == [ + ColumnType.STRING, + ColumnType.INTEGER, + ] + + @pytest.mark.parametrize("search_enabled", [True, False]) + def test_search_change_response_parses_status(self, search_enabled): + """The resulting search status is retained, including when it is False.""" + # GIVEN a search change response + data = { + "concreteType": TABLE_SEARCH_CHANGE_RESPONSE, + "searchEnabled": search_enabled, + } + + # WHEN converting it + response = table_update_response_from_dict(data) + + # THEN the status is available + assert response.search_enabled is search_enabled + + +class TestEntityUpdateResult: + """Test suite for EntityUpdateResult and the failure detail it retains.""" + + @pytest.mark.parametrize( + "data,expected_succeeded", + [ + ({"entityId": "syn1"}, True), + ({"entityId": "syn1", "failureCode": "NOT_FOUND"}, False), + ({"entityId": "syn1", "failureMessage": "something broke"}, False), + ( + { + "entityId": "syn1", + "failureCode": "ILLEGAL_ARGUMENT", + "failureMessage": "bad value", + }, + False, + ), + ], + ids=[ + "no_failure_reported", + "failure_code_only", + "failure_message_only", + "failure_code_and_message", + ], + ) + def test_succeeded(self, data, expected_succeeded): + """An update failed when Synapse reported either a code or a message.""" + # GIVEN an entity update result + # WHEN converting it + update_result = EntityUpdateResult.fill_from_dict(data) + + # THEN succeeded reflects whether any failure was reported + assert update_result.succeeded is expected_succeeded + + @pytest.mark.parametrize( + "failure_code", [failure_code.value for failure_code in EntityUpdateFailureCode] + ) + def test_known_failure_code_is_coerced_to_the_enum(self, failure_code): + """Every documented failure code maps onto the enum.""" + # GIVEN a result with a documented failure code + # WHEN converting it + update_result = EntityUpdateResult.fill_from_dict( + {"entityId": "syn1", "failureCode": failure_code} + ) + + # THEN the code is the matching enum member + assert update_result.failure_code == EntityUpdateFailureCode(failure_code) + + def test_unrecognized_failure_code_coerces_to_unknown(self): + """A failure code added to Synapse after this release must not raise.""" + # GIVEN a result with a failure code we do not model + # WHEN converting it + update_result = EntityUpdateResult.fill_from_dict( + {"entityId": "syn1", "failureCode": "SOMETHING_NEW"} + ) + + # THEN it coerces to UNKNOWN rather than raising a ValueError + assert update_result.failure_code == EntityUpdateFailureCode.UNKNOWN + + def test_failure_detail_is_retained(self): + """The failure code and message are kept, not used as a filter and discarded.""" + # GIVEN a failed entity update + # WHEN converting it + update_result = EntityUpdateResult.fill_from_dict( + { + "entityId": "syn1", + "failureCode": "ILLEGAL_ARGUMENT", + "failureMessage": "value is not a valid date", + } + ) + + # THEN every part of the failure is available to report to the user + assert update_result.entity_id == "syn1" + assert update_result.failure_code == EntityUpdateFailureCode.ILLEGAL_ARGUMENT + assert update_result.failure_message == "value is not a valid date" + + def test_successful_and_failed_updates_are_separated(self): + """EntityUpdateResults splits the successes from the failures.""" + # GIVEN a mix of successful and failed updates + response = table_update_response_from_dict( + _entity_update_results( + [ + {"entityId": "syn1"}, + {"entityId": "syn2", "failureCode": "NOT_FOUND"}, + {"entityId": "syn3", "failureMessage": "something broke"}, + ] + ) + ) + + # THEN the successes are reported by ID and the failures are reported whole + assert response.successful_entity_ids == ["syn1"] + assert [ + update_result.entity_id for update_result in response.failed_entity_updates + ] == ["syn2", "syn3"] + + def test_successful_update_with_no_entity_id_is_still_counted(self): + """A success that Synapse reported with no entity ID cannot be reported by ID, + but it did apply, so it must still be counted as a changed row.""" + # GIVEN a successful update that carries no entity ID + response = table_update_response_from_dict( + _entity_update_results([{"entityId": "syn1"}, {}]) + ) + + # THEN only the identified success is reported by ID, nothing is treated as a + # failure, and both successes are counted + assert response.successful_entity_ids == ["syn1"] + assert response.failed_entity_updates == [] + assert response.rows_changed == 2 + + def test_absent_update_results_yields_empty_lists(self): + """A response with no update results reports empty rather than raising.""" + # GIVEN an EntityUpdateResults with no update results at all + response = EntityUpdateResults() + + # THEN both properties are empty and the count is 0 + assert response.update_results is None + assert response.successful_entity_ids == [] + assert response.failed_entity_updates == [] + assert response.rows_changed == 0 + + +class TestTableUpdateRequest: + """Test suite for the changes that may be included in a TableUpdateTransaction. + + Synapse accepts four kinds of change within one transaction, and every one of them + must be usable through TableUpdateTransaction. + """ + + @staticmethod + def _appendable_row_set_request() -> AppendableRowSetRequest: + return AppendableRowSetRequest( + entity_id="syn123", + to_append=PartialRowSet( + table_id="syn123", + rows=[PartialRow(values=[{"key": "1", "value": "a"}], row_id=1)], + ), + ) + + @staticmethod + def _upload_to_table_request() -> UploadToTableRequest: + return UploadToTableRequest( + table_id="syn123", upload_file_handle_id="456", update_etag="etag" + ) + + @staticmethod + def _table_schema_change_request() -> TableSchemaChangeRequest: + return TableSchemaChangeRequest( + entity_id="syn123", + changes=[ColumnChange(new_column_id="789")], + ordered_column_ids=["789"], + ) + + @staticmethod + def _table_search_change_request() -> TableSearchChangeRequest: + return TableSearchChangeRequest(entity_id="syn123", search_enabled=True) + + @pytest.mark.parametrize( + "request_class", + [ + AppendableRowSetRequest, + UploadToTableRequest, + TableSchemaChangeRequest, + TableSearchChangeRequest, + ], + ) + def test_every_change_is_a_table_update_request(self, request_class): + """Every change that Synapse accepts within a transaction shares the base + class, so a caller may type a change as TableUpdateRequest.""" + # GIVEN a class that models one of the changes documented for a transaction + # THEN it is a TableUpdateRequest + assert issubclass(request_class, TableUpdateRequest) + + def test_base_class_cannot_be_used_on_its_own(self): + """The base class only describes the shared contract.""" + # WHEN the base class is instantiated + # THEN it is rejected because it models no change of its own + with pytest.raises(TypeError): + TableUpdateRequest() + + def test_search_change_request_converts_to_a_synapse_request(self): + """A search change is sent with the concrete type that Synapse expects.""" + # GIVEN a request to enable search on a table + request = TableSearchChangeRequest(entity_id="syn123", search_enabled=True) + + # WHEN it is converted for the REST API + # THEN the entity, the flag, and the concrete type are all sent + assert request.to_synapse_request() == { + "concreteType": TABLE_SEARCH_CHANGE_REQUEST, + "entityId": "syn123", + "searchEnabled": True, + } + + def test_search_change_request_may_disable_search(self): + """The same request turns search off, so False must not be dropped.""" + # GIVEN a request to disable search on a table + request = TableSearchChangeRequest(entity_id="syn123", search_enabled=False) + + # WHEN it is converted for the REST API + # THEN the flag is sent as False rather than left out + assert request.to_synapse_request()["searchEnabled"] is False + + def test_upload_to_table_request_reports_its_entity_id(self): + """A CSV upload names its entity table_id, and entity_id gives every change + one way to report the entity it applies to.""" + # GIVEN a request to apply an uploaded file to a table + request = self._upload_to_table_request() + + # THEN the entity is available under the shared name + assert request.entity_id == "syn123" + + @pytest.mark.parametrize( + "table_id,entity_id", + [ + ("syn123", None), + (None, "syn123"), + ("syn123", "syn123"), + ], + ids=["table_id_only", "entity_id_only", "both_equal"], + ) + def test_upload_to_table_request_aliases_table_id_and_entity_id( + self, table_id, entity_id + ): + """table_id and entity_id are aliases, so giving either one, or both with the + same value, names the table and fills the other field.""" + # GIVEN a request that names its table through one or both of the aliases + request = UploadToTableRequest( + table_id=table_id, entity_id=entity_id, upload_file_handle_id="456" + ) + + # THEN both fields hold the table + assert request.table_id == "syn123" + assert request.entity_id == "syn123" + + @pytest.mark.parametrize( + "table_id,entity_id", + [ + (None, None), + ("syn123", "syn456"), + ], + ids=["neither_given", "both_given_but_different"], + ) + def test_upload_to_table_request_rejects_an_unnamed_or_ambiguous_table( + self, table_id, entity_id + ): + """A request that names no table, or two different tables, is rejected.""" + # WHEN a request is created without a table or with two conflicting tables + # THEN it is rejected + with pytest.raises(ValueError): + UploadToTableRequest( + table_id=table_id, entity_id=entity_id, upload_file_handle_id="456" + ) + + def test_transaction_accepts_every_kind_of_change(self): + """A single transaction may mix all four kinds of change, and each is sent in + the order it was given.""" + # GIVEN a transaction that holds one of each kind of change + changes = [ + self._table_schema_change_request(), + self._appendable_row_set_request(), + self._upload_to_table_request(), + self._table_search_change_request(), + ] + transaction = TableUpdateTransaction(entity_id="syn123", changes=changes) + + # WHEN it is converted for the REST API + request = transaction.to_synapse_request() + + # THEN every change is sent, in the order it was given + assert [change["concreteType"] for change in request["changes"]] == [ + TABLE_SCHEMA_CHANGE_REQUEST, + APPENDABLE_ROWSET_REQUEST, + UPLOAD_TO_TABLE_REQUEST, + TABLE_SEARCH_CHANGE_REQUEST, + ] + # AND each change is converted by the class that models it + assert request["changes"] == [change.to_synapse_request() for change in changes] + + +class TestTableUpdateTransactionFillFromDict: + """Test suite for the aggregates that TableUpdateTransaction.fill_from_dict fills. + + total_rows_changed is the count that upsert_rows reports, and + entities_with_changes_applied must keep its original meaning because it is used as + a dictionary key when waiting for an eventually consistent view. + """ + + def test_table_update_response_is_counted(self): + """A table update reports a row count even though it reports no entity.""" + # GIVEN the response recorded from production for a table upsert + transaction = TableUpdateTransaction(entity_id="syn76890550").fill_from_dict( + {"results": [_row_reference_set_results(5)]} + ) + + # THEN the confirmed row count is available + assert transaction.total_rows_changed == 5 + # AND entities_with_changes_applied keeps its original meaning, which is that a + # table update never fills it + assert transaction.entities_with_changes_applied is None + # AND the response is available as the class that models it + assert len(transaction.results) == 1 + assert isinstance(transaction.results[0], RowReferenceSetResults) + + def test_view_update_response_is_counted(self): + """A view update contributes both a row count and the successful entity IDs.""" + # GIVEN a view response with one success and two failures + transaction = TableUpdateTransaction(entity_id="syn123").fill_from_dict( + { + "results": [ + _entity_update_results( + [ + {"entityId": "syn1"}, + {"entityId": "syn2", "failureCode": "NOT_FOUND"}, + { + "entityId": "syn3", + "failureCode": "ILLEGAL_ARGUMENT", + "failureMessage": "bad value", + }, + ] + ) + ] + } + ) + + # THEN only the successful update is counted + assert transaction.total_rows_changed == 1 + # AND only the successful IDs are reported + assert transaction.entities_with_changes_applied == ["syn1"] + # AND the failures are reported with the detail Synapse gave for each one + assert [ + ( + failed_update.entity_id, + failed_update.failure_code, + failed_update.failure_message, + ) + for failed_update in transaction.failed_entity_updates + ] == [ + ("syn2", EntityUpdateFailureCode.NOT_FOUND, None), + ("syn3", EntityUpdateFailureCode.ILLEGAL_ARGUMENT, "bad value"), + ] + + def test_failed_entity_updates_are_collected_across_every_response(self): + """The failures of every response that reports one are flattened together, and a + response that reports no per-entity outcome contributes nothing.""" + # GIVEN a transaction whose changes returned two view responses and one table + # response + transaction = TableUpdateTransaction(entity_id="syn123").fill_from_dict( + { + "results": [ + _entity_update_results( + [ + {"entityId": "syn1"}, + {"entityId": "syn2", "failureCode": "NOT_FOUND"}, + ] + ), + _row_reference_set_results(5), + _entity_update_results( + [{"entityId": "syn3", "failureCode": "UNAUTHORIZED"}] + ), + ] + } + ) + + # THEN both failures are reported, in the order the responses were returned + assert [ + failed_update.entity_id + for failed_update in transaction.failed_entity_updates + ] == ["syn2", "syn3"] + + def test_table_update_response_reports_no_failed_entity_update(self): + """A rejected row update on a table fails the asynchronous job and raises, so a + table response never carries a per-row failure.""" + # GIVEN the response recorded from production for a table upsert + transaction = TableUpdateTransaction(entity_id="syn76890550").fill_from_dict( + {"results": [_row_reference_set_results(5)]} + ) + + # THEN no failure is reported + assert transaction.failed_entity_updates == [] + + def test_original_field_stays_none_when_no_entity_succeeded(self): + """Regression guard: entities_with_changes_applied is only set when there is + at least one success. It is used as a dictionary key at the call site, so its + behaviour must not drift.""" + # GIVEN a view response in which every update failed + transaction = TableUpdateTransaction(entity_id="syn123").fill_from_dict( + { + "results": [ + _entity_update_results( + [ + {"entityId": "syn1", "failureCode": "NOT_FOUND"}, + {"entityId": "syn2", "failureCode": "UNAUTHORIZED"}, + ] + ) + ] + } + ) + + # THEN the field is left as None + assert transaction.entities_with_changes_applied is None + assert transaction.total_rows_changed == 0 + # AND both failures are still reported + assert len(transaction.failed_entity_updates) == 2 + + def test_counts_are_summed_across_every_response(self): + """One response is returned per change in the transaction, and each that + reports a count contributes to the total.""" + # GIVEN a transaction whose changes returned three different response types + transaction = TableUpdateTransaction(entity_id="syn123").fill_from_dict( + { + "results": [ + { + "concreteType": TABLE_SCHEMA_CHANGE_RESPONSE, + "schema": [{"name": "col1", "columnType": "STRING"}], + }, + _row_reference_set_results(5), + {"concreteType": UPLOAD_TO_TABLE_RESULT, "rowsProcessed": 2}, + ] + } + ) + + # THEN the schema change contributes nothing and the row counts are summed + assert transaction.total_rows_changed == 7 + # AND the responses are kept in the order Synapse returned them + assert [type(response) for response in transaction.results] == [ + TableSchemaChangeResponse, + RowReferenceSetResults, + UploadToTableResult, + ] + + def test_a_row_change_that_created_no_row_contributes_a_confirmed_zero(self): + """A table row change always carries a row count, so one that created no row + version contributes a confirmed 0 rather than being dropped from the total as a + response that reports no count at all.""" + # GIVEN a transaction whose row change reported no row reference set + transaction = TableUpdateTransaction(entity_id="syn123").fill_from_dict( + { + "results": [ + {"concreteType": ROW_REFERENCE_SET_RESULTS}, + {"concreteType": UPLOAD_TO_TABLE_RESULT, "rowsProcessed": 2}, + ] + } + ) + + # THEN the row change is counted as 0 rather than skipped + assert transaction.results[0].rows_changed == 0 + assert transaction.total_rows_changed == 2 + + def test_unmodelled_response_does_not_break_the_count(self): + """An unmodelled response contributes nothing rather than raising.""" + # GIVEN a transaction that returned one known and one unknown response + transaction = TableUpdateTransaction(entity_id="syn123").fill_from_dict( + { + "results": [ + _row_reference_set_results(3), + { + "concreteType": "org.sagebionetworks.repo.model.table.New", + "somethingNew": 99, + }, + ] + } + ) + + # THEN only the known response is counted + assert transaction.total_rows_changed == 3 + assert isinstance(transaction.results[1], UnknownTableUpdateResponse) + + @pytest.mark.parametrize( + "synapse_response", + [{}, {"results": None}], + ids=["results_absent", "results_null"], + ) + def test_aggregates_stay_none_when_nothing_was_returned(self, synapse_response): + """The aggregates are None before anything is reported, never 0. The call site + relies on that to tell an absent count from a confirmed count of 0.""" + # GIVEN a response that carries no results + transaction = TableUpdateTransaction(entity_id="syn123").fill_from_dict( + synapse_response + ) + + # THEN nothing was counted and nothing was parsed + assert transaction.total_rows_changed is None + assert transaction.results is None + assert transaction.entities_with_changes_applied is None + # AND the failure list is empty rather than None, since there is nothing to + # tell apart: a transaction that reported nothing reported no failure + assert transaction.failed_entity_updates == [] + + def test_empty_results_array_is_kept_apart_from_an_absent_one(self): + """An empty results array means Synapse reported the transaction and changed + nothing. That is a confirmed count of 0, which the caller must be able to tell + apart from a transaction that was never sent.""" + # GIVEN a response that reports an empty results array + transaction = TableUpdateTransaction(entity_id="syn123").fill_from_dict( + {"results": []} + ) + + # THEN the empty array is kept as such, and the count is a confirmed 0 + assert transaction.results == [] + assert transaction.total_rows_changed == 0 + assert transaction.entities_with_changes_applied is None + assert transaction.failed_entity_updates == [] + + def test_a_later_send_replaces_the_results_of_an_earlier_one(self): + """The same transaction instance can be sent more than once, since + send_job_and_wait_async returns self. A later response must replace the + results of the earlier one rather than leave a stale count in place.""" + # GIVEN a transaction that already reported changed rows + transaction = TableUpdateTransaction(entity_id="syn123").fill_from_dict( + {"results": [_row_reference_set_results(5)]} + ) + assert transaction.total_rows_changed == 5 + + # WHEN the same instance is sent again and Synapse reports no results + transaction.fill_from_dict({"results": None}) + + # THEN the counts of the earlier send are gone + assert transaction.results is None + assert transaction.total_rows_changed is None + + def test_snapshot_version_number_is_filled(self): + """A transaction that created a snapshot reports the new version number + alongside the modelled responses.""" + # GIVEN a response from Synapse that reports a snapshot version + transaction = TableUpdateTransaction(entity_id="syn123").fill_from_dict( + {"results": [_row_reference_set_results(2)], "snapshotVersionNumber": 4} + ) + + # THEN the version number and the modelled responses are both available + assert transaction.snapshot_version_number == 4 + assert [type(response) for response in transaction.results] == [ + RowReferenceSetResults + ] + + +class TestUpsertRowsResultReporting: + """Test suite for how _upsert_rows_async reports what Synapse confirmed. + + Regression coverage for SYNPY-1912, where every successful table upsert logged a + contradictory message: the correct number of updated rows, followed by a claim that + the same number of rows could not be updated. + """ + + @pytest.fixture(autouse=True, scope="function") + def init_syn(self, syn: Synapse) -> None: + self.syn = syn + + COLUMNS_FOR_TEST = { + "col1": Column(name="col1", column_type=ColumnType.STRING, id="id1"), + "col2": Column(name="col2", column_type=ColumnType.INTEGER, id="id2"), + } + + @dataclass + class TableForTest(TableUpsertMixin): + """A minimal Table-like entity. The class name is deliberately not one of + CLASSES_THAT_CONTAIN_ROW_ETAG, so the rows carry no etag.""" + + id: Optional[str] = None + name: Optional[str] = None + columns: Dict[str, Column] = field(default_factory=dict) + _last_persistent_instance: Optional[Any] = True + query_results: List[Any] = field(default_factory=list) + stored_rows: Optional[Any] = None + + async def query_async(self, query: str, synapse_client=None) -> Any: + return self.query_results.pop(0) + + async def store_rows_async(self, values=None, **kwargs) -> None: + self.stored_rows = values + + @dataclass + class ViewForTest(ViewBase, TableUpsertMixin): + """A minimal View-like entity. Only the entities that back a view report a + per-row outcome, so this is the only kind of entity that can produce a failure + clause.""" + + columns: Dict[str, Column] = field(default_factory=dict) + query_results: List[Any] = field(default_factory=list) + + async def query_async(self, query: str, synapse_client=None) -> Any: + return self.query_results.pop(0) + + @staticmethod + def _existing_rows(row_ids: List[str], col2_values: List[int]) -> Any: + """The rows a query returns for the keys that are already in the table.""" + return pd.DataFrame( + { + "ROW_ID": row_ids, + "col1": [f"key{row_id}" for row_id in row_ids], + "col2": col2_values, + } + ) + + @staticmethod + def _values_to_upsert(keys: List[str]) -> Dict[str, Any]: + """New values for each of the given keys. Every value differs from the value + that _existing_rows returns, so every matched row is an update.""" + return { + "col1": [f"key{key}" for key in keys], + "col2": [int(key) * 100 for key in keys], + } + + def _table( + self, query_results: List[Any] + ) -> "TestUpsertRowsResultReporting.TableForTest": + return self.TableForTest( + id="syn123", + name="test-table", + columns=dict(self.COLUMNS_FOR_TEST), + query_results=query_results, + ) + + def _view( + self, query_results: List[Any] + ) -> "TestUpsertRowsResultReporting.ViewForTest": + view = self.ViewForTest( + id="syn456", + name="test-view", + columns=dict(self.COLUMNS_FOR_TEST), + query_results=query_results, + ) + view._last_persistent_instance = True + return view + + @staticmethod + def _table_transaction(row_count: int) -> TableUpdateTransaction: + """The transaction Synapse returns for the update half of a table upsert.""" + return TableUpdateTransaction(entity_id="syn123").fill_from_dict( + {"results": [_row_reference_set_results(row_count)]} + ) + + @staticmethod + def _view_transaction( + update_results: List[Dict[str, Any]], + ) -> TableUpdateTransaction: + """The transaction Synapse returns for a change applied to the entities that + back a view.""" + return TableUpdateTransaction(entity_id="syn456").fill_from_dict( + {"results": [_entity_update_results(update_results)]} + ) + + @staticmethod + def _upsert_message(mock_info: MagicMock) -> str: + """The single logged message that reports the upsert counts.""" + messages = [ + call.args[0] + for call in mock_info.call_args_list + if "rows to update" in call.args[0] + ] + assert len(messages) == 1 + return messages[0] + + async def test_table_upsert_reports_the_confirmed_count_with_no_failure_clause( + self, + ): + """A successful table upsert must not claim that any row failed. This is the + defect that was reported.""" + # GIVEN a table that holds 5 of the 7 keys being upserted + entity = self._table( + [self._existing_rows(["1", "2", "3", "4", "5"], [1, 2, 3, 4, 5])] + ) + + # WHEN Synapse confirms all 5 row updates + with ( + patch( + _PUSH_ROW_UPDATES_TO_SYNAPSE_PATCH, + new_callable=AsyncMock, + return_value=[self._table_transaction(5)], + ), + patch.object(self.syn.logger, "info") as mock_info, + patch.object(self.syn.logger, "debug") as mock_debug, + ): + await _upsert_rows_async( + entity=entity, + values=self._values_to_upsert(["1", "2", "3", "4", "5", "6", "7"]), + primary_keys=["col1"], + synapse_client=self.syn, + ) + + # THEN the confirmed count is reported and no failure is claimed + assert ( + self._upsert_message(mock_info) + == "[syn123:test-table]: Found 5 rows to update and 2 rows to insert" + ) + # AND no accounting gap is reported, because every row was accounted for + assert not [ + call for call in mock_debug.call_args_list if "gap in how" in call.args[0] + ] + # AND the 2 unmatched rows were inserted + assert len(entity.stored_rows) == 2 + + async def test_row_update_results_accumulate_across_query_chunks(self): + """An upsert of more than rows_per_query rows reports the total across every + chunk, not just the count from the last one.""" + # GIVEN 6 rows to upsert, queried 2 at a time, all of which already exist + entity = self._table( + [ + self._existing_rows(["1", "2"], [1, 2]), + self._existing_rows(["3", "4"], [3, 4]), + self._existing_rows(["5", "6"], [5, 6]), + ] + ) + + # WHEN Synapse confirms 2 row updates per chunk + with ( + patch( + _PUSH_ROW_UPDATES_TO_SYNAPSE_PATCH, + new_callable=AsyncMock, + side_effect=[ + [self._table_transaction(2)], + [self._table_transaction(2)], + [self._table_transaction(2)], + ], + ) as mock_push, + patch.object(self.syn.logger, "info") as mock_info, + ): + await _upsert_rows_async( + entity=entity, + values=self._values_to_upsert(["1", "2", "3", "4", "5", "6"]), + primary_keys=["col1"], + rows_per_query=2, + synapse_client=self.syn, + ) + + # THEN every chunk was pushed + assert mock_push.await_count == 3 + # AND the reported count is the total of all three chunks + assert ( + self._upsert_message(mock_info) + == "[syn123:test-table]: Found 6 rows to update and 0 rows to insert" + ) + + async def test_dry_run_reports_the_planned_count(self): + """Nothing is pushed on a dry run, so the planned count is the only meaningful + answer to what would happen.""" + # GIVEN a table that holds 5 of the 7 keys being upserted + entity = self._table( + [self._existing_rows(["1", "2", "3", "4", "5"], [1, 2, 3, 4, 5])] + ) + + # WHEN upserting as a dry run + with ( + patch( + _PUSH_ROW_UPDATES_TO_SYNAPSE_PATCH, new_callable=AsyncMock + ) as mock_push, + patch.object(self.syn.logger, "info") as mock_info, + ): + await _upsert_rows_async( + entity=entity, + values=self._values_to_upsert(["1", "2", "3", "4", "5", "6", "7"]), + primary_keys=["col1"], + dry_run=True, + synapse_client=self.syn, + ) + + # THEN nothing was sent to Synapse + mock_push.assert_not_awaited() + assert entity.stored_rows is None + # AND the planned counts are reported, with no failure claimed + assert ( + self._upsert_message(mock_info) + == "[syn123:test-table]: Found 5 rows to update and 2 rows to insert" + ) + + async def test_confirmed_count_of_zero_is_reported_as_zero(self): + """A push that Synapse confirmed changed nothing reports 0 rather than falling + back to the planned count. The fallback is what hid the original defect.""" + # GIVEN a table that holds all 5 keys being upserted + entity = self._table( + [self._existing_rows(["1", "2", "3", "4", "5"], [1, 2, 3, 4, 5])] + ) + + # WHEN Synapse reports no row references and no failure + with ( + patch( + _PUSH_ROW_UPDATES_TO_SYNAPSE_PATCH, + new_callable=AsyncMock, + return_value=[self._table_transaction(0)], + ), + patch.object(self.syn.logger, "info") as mock_info, + patch.object(self.syn.logger, "debug") as mock_debug, + ): + await _upsert_rows_async( + entity=entity, + values=self._values_to_upsert(["1", "2", "3", "4", "5"]), + primary_keys=["col1"], + synapse_client=self.syn, + ) + + # THEN the confirmed count of 0 is reported, and still no failure is claimed + assert ( + self._upsert_message(mock_info) + == "[syn123:test-table]: Found 0 rows to update and 0 rows to insert" + ) + # AND the shortfall is reported as a client accounting gap, at debug level + gap_messages = [ + call.args[0] + for call in mock_debug.call_args_list + if "gap in how" in call.args[0] + ] + assert len(gap_messages) == 1 + assert "Synapse confirmed 0 of the 5 rows sent for update" in gap_messages[0] + + @pytest.mark.parametrize( + "failed_updates,expected_clause", + [ + ( + [{"entityId": "syn2", "failureCode": "NOT_FOUND"}], + ". 1 rows could not be updated: syn2 (NOT_FOUND)", + ), + ( + [ + { + "entityId": "syn2", + "failureCode": "ILLEGAL_ARGUMENT", + "failureMessage": "bad value", + } + ], + ". 1 rows could not be updated: syn2 (ILLEGAL_ARGUMENT: bad value)", + ), + ( + [{"entityId": "syn2", "failureCode": "SOMETHING_NEW"}], + ". 1 rows could not be updated: syn2 (UNKNOWN)", + ), + ( + [{"failureMessage": "something broke"}], + ". 1 rows could not be updated: unknown row (UNKNOWN: something broke)", + ), + ( + [ + {"entityId": "syn2", "failureCode": "NOT_FOUND"}, + { + "entityId": "syn3", + "failureCode": "ILLEGAL_ARGUMENT", + "failureMessage": "bad value", + }, + ], + ". 2 rows could not be updated: syn2 (NOT_FOUND);" + " syn3 (ILLEGAL_ARGUMENT: bad value)", + ), + ], + ids=[ + "code_only", + "code_and_message", + "unrecognized_code", + "message_with_no_id_or_code", + "two_failures", + ], + ) + async def test_view_upsert_reports_the_failure_detail_synapse_returned( + self, failed_updates, expected_clause + ): + """A failure clause is built from the codes and messages Synapse reported, so + the user has something actionable rather than a bare count.""" + # GIVEN a view that holds all 3 keys being upserted + entity = self._view([self._existing_rows(["1", "2", "3"], [1, 2, 3])]) + + # WHEN Synapse reports one success and the given failures + with ( + patch( + _PUSH_ROW_UPDATES_TO_SYNAPSE_PATCH, + new_callable=AsyncMock, + return_value=[ + self._view_transaction([{"entityId": "syn1"}] + failed_updates) + ], + ), + patch.object(self.syn.logger, "info") as mock_info, + ): + await _upsert_rows_async( + entity=entity, + values=self._values_to_upsert(["1", "2", "3"]), + primary_keys=["col1"], + synapse_client=self.syn, + ) + + # THEN the confirmed count is followed by the detail of each failure + assert self._upsert_message(mock_info) == ( + "[syn456:test-view]: Found 1 rows to update and 0 rows to insert" + + expected_clause + ) + + async def test_view_upsert_with_no_failures_claims_none(self): + """A view upsert that Synapse fully applied reports no failure either.""" + # GIVEN a view that holds all 3 keys being upserted + entity = self._view([self._existing_rows(["1", "2", "3"], [1, 2, 3])]) + + # WHEN Synapse confirms all 3 entity updates + with ( + patch( + _PUSH_ROW_UPDATES_TO_SYNAPSE_PATCH, + new_callable=AsyncMock, + return_value=[ + self._view_transaction( + [ + {"entityId": "syn1"}, + {"entityId": "syn2"}, + {"entityId": "syn3"}, + ] + ) + ], + ), + patch.object(self.syn.logger, "info") as mock_info, + ): + await _upsert_rows_async( + entity=entity, + values=self._values_to_upsert(["1", "2", "3"]), + primary_keys=["col1"], + synapse_client=self.syn, + ) + + # THEN all 3 updates are reported as applied + assert ( + self._upsert_message(mock_info) + == "[syn456:test-view]: Found 3 rows to update and 0 rows to insert" + )