diff --git a/CLAUDE.md b/CLAUDE.md index 3a3d6f1..9be0e06 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -119,6 +119,8 @@ Optional extras: - `src/dp_python_lib/client/sample_status_conversions.py` - Per-sample expansion of query results (no optional extras required): `expand_data_timestamps()` (SamplingClock positions computed in **integer nanoseconds**, never float seconds — the exact-match contract depends on it), `bucket_to_rows()` / `buckets_to_rows()` / `iter_rows()` yielding `SampleStatusRow` objects with absent confidence/reason surfaced as `None` rather than fabricated `0.0`/`""` - `src/dp_python_lib/client/query_client.py` - v2 time-series query client (sample-oriented) exposed as `client.query`. Low-level wrappers `query_samples()` (unary, one resumable page) and `iter_query_samples()` (transparent paging), plus `iter_query_samples_stream()` (server-streaming, fire-and-consume, lazy). Queries are described by a kind-neutral `QueryParams` built from the `PvQuery` (`PV`) and `ConfigQuery` (`CFG`) criterion helpers; shares a `_build_query_spec()` seam so a future bucket request builder reuses it. Results wrap the raw `ColumnTable` (`.column_table`, `.next_page_token`); `.to_dataframe()`/`.to_numpy()` delegate to `query_conversions` (Phase 2, optional `[analysis]` extra) - `src/dp_python_lib/client/query_conversions.py` - Pythonic conversions for query results (optional `[analysis]` extra: pandas/numpy/openpyxl, imported lazily). `data_value_to_python()` (oneof extractor: scalars→native, timestamp→epoch-nanos, array→list, structure→dict, image→`Image` wrapper, fail-loud on unhandled arm), `column_table_to_dataframe()` (UTC datetime index + one column per DataColumn; dense-alignment and duplicate-column-name fail-loud; ColumnMetadata in `df.attrs`), `column_table_to_numpy()` (dict of 1-D arrays; complex arms stay 1-D object arrays rather than collapsing to 2-D), `dataframe_to_excel()` (thin `to_excel()` wrapper: row-limit guard, tz-drop, complex-cell stringification), and `query_samples_to_dataframe()`/`stream_query_samples_to_dataframes()` whole-query conveniences (unary concats by column name; streaming yields per-page frames lazily) +- `src/dp_python_lib/client/service_api_client_base.py` - Base class for the service clients: owns the channel and the one-per-client gRPC stub, and provides `_dispatch()`, the shared three-tier sender that all 18 unary `_send_*` methods delegate to +- `tests/unit/test_service_api_client_base.py` - Unit tests for `_dispatch` itself (success, business error, unrecognized response, `RpcError` with and without a resolvable `code()`, unexpected exception, and the `request_log`/`success_log` hooks) - `tests/unit/test_ingestion_client.py` - Unit tests for IngestionClient functionality - `tests/unit/test_pv_metadata_client.py` - Unit tests for PvMetadataClient functionality - `tests/unit/test_machine_config_client.py` - Unit tests for the Configuration side of MachineConfigClient @@ -144,6 +146,31 @@ Optional extras: - Service clients extend `ServiceApiClientBase`, which is constructed with `(channel, stub_class)` and creates the gRPC stub **once** at init time, stored as `self._stub`. `_send_*` methods reuse `self._stub` rather than creating a new stub per call. +- **Every unary `_send_*` method delegates to `ServiceApiClientBase._dispatch()`** rather than writing + the three-tier block out by hand (issue #14; `plan/tickets/14/plan.md`). A sender is now the call + itself plus its log messages: + ```python + def _send_query_pv_metadata(self, request: annotation_pb2.QueryPvMetadataRequest) -> QueryPvMetadataApiResult: + return self._dispatch( + self._stub.queryPvMetadata, # the bound stub method + request, + QueryPvMetadataApiResult, # result class; all take (is_error, message, response) + "pvMetadataResult", # the success oneof field + "queryPvMetadata", # op name, used in log and error messages + request_log=lambda: self.logger.info( + "Calling queryPvMetadata API with %d criteria", len(request.criteria)), + success_log=lambda response: self.logger.info( + "QueryPvMetadata returned %d records", len(response.pvMetadataResult.pvMetadata)), + ) + ``` + `request_log` / `success_log` are optional: omit them and `_dispatch` logs a generic + `"Calling API"` / `" completed successfully"`. Pass one whenever the message names the + entity or reports a count off the response — that detail is the reason the parameters exist, and + nothing in the test suite asserts on log content, so a dropped message fails silently. +- **The server-streaming senders deliberately do not use `_dispatch`.** + `_send_query_samples_stream()` and `_send_query_sample_statuses_stream()` *yield* one result per + streamed message — error results included, for the public `iter_*` wrapper to convert into a + `RuntimeError` — which is a different contract from returning a single result. Leave them as they are. - Where one gRPC service backs several feature areas (e.g. `DpAnnotationService` covers PV metadata, machine configuration, and annotations), use a lightweight facade (`AnnotationClient`) that owns the shared channel and exposes feature-scoped clients as attributes (`annotation.pv_metadata`). This @@ -151,12 +178,24 @@ Optional extras: ### gRPC Error Handling - Use **synchronous gRPC calls** with `DpIngestionServiceStub` for simplicity -- Implement **three-tier error handling**: +- **Three-tier error handling**, implemented once in `ServiceApiClientBase._dispatch()` and inherited by + every unary sender (see the Client Implementation Pattern above): 1. **gRPC Exceptions** (`grpc.RpcError`) - network/connection errors 2. **Business Logic Errors** - check response `exceptionalResult` field 3. **General Exceptions** - unexpected errors +- A response carrying neither `exceptionalResult` nor the expected success field is itself an error, so an + unrecognized response shape is never mistaken for a success. - Check protobuf union fields with `response.HasField('fieldName')` - Return consistent result objects with `is_error` flag and appropriate messages +- The error-message text is part of the contract — roughly 40 unit tests match on it with `assertIn`. + `_dispatch` produces `f"gRPC error: {e.details()}"`, `f"Unexpected error: {e!s}"`, and + `f"Unexpected response format: neither exceptionalResult nor {success_field} found"`. +- When logging an `RpcError`, `_dispatch` includes `e.code()` only when it resolves: a bare + `grpc.RpcError()`, which is what the test mocks raise, has no usable code. +- `_dispatch` names the operation with the raw camelCase `op_name` in all three error-log tiers. The + hand-written senders capitalized it in the business-error warning only (`SavePvMetadata API returned + business error`), while their other two error logs already used camelCase; the refactor dropped that + one inconsistency deliberately. Log text only -- returned messages are unchanged. ### Testing Best Practices - Use `@patch` decorators to mock gRPC stubs and avoid real network calls diff --git a/plan/tickets/14/plan.md b/plan/tickets/14/plan.md new file mode 100644 index 0000000..a9f495d --- /dev/null +++ b/plan/tickets/14/plan.md @@ -0,0 +1,142 @@ +# Issue #14 — Extract a shared `_dispatch` helper for the unary `_send_*` methods + +## Overview + +Collapse the 18 near-identical unary `_send_*` bodies across the five service clients into short +delegations to one `_dispatch()` helper on `ServiceApiClientBase`. The three-tier error handling +(gRPC exception / business `exceptionalResult` / unexpected exception) is written once instead of +eighteen times, and each new sender added by #6 costs ~6 lines rather than ~40. + +This is a pure refactor: no public API changes, no behavior changes, no new tests of new behavior. +It is deliberately sequenced *before* #6 (issue comment 2026-09-09, and `plan/tickets/6/plan.md` Q3) +so #6's 10 new senders are written in the collapsed form from the start. + +## Background / triage findings + +Verified against the code on `main` at b6d0b37, not taken from the ticket text. + +- **T1 — The count is 18 unary senders, and the ticket's corrected figure is right.** `grep -c "def _send_"` + reports 20 across `src/dp_python_lib/client/`; two of those are the server-streaming senders + (`_send_query_sample_statuses_stream`, `_send_query_samples_stream`), leaving 18 unary. By client: + `MachineConfigClient` 9, `PvMetadataClient` 4, `SampleStatusClient` 3, `QueryClient` 1, + `IngestionClient` 1. The issue's *original* body undercounts (it names only the 13 annotation-service + senders it knew about); its 2026-09-09 correction comment and the #6 plan's finding 5 both say 18 and + are accurate. + +- **T2 — All 18 result classes share one constructor shape.** Every `*ApiResult` extends `ApiResultBase` + with `__init__(self, is_error: bool, message: str, response: | None = None)`. So a helper + can construct any of them uniformly as `result_cls(is_error=..., message=..., response=...)`, and the + `result_cls` parameter in the ticket's sketch is sound. No result class takes extra required arguments. + +- **T3 — Success-branch logging is *not* uniform, and the ticket's signature cannot carry it.** This is + the one real gap in the ticket. Each sender logs a bespoke INFO line on success; four of them report a + count read off the response body: + - `"QueryPvMetadata returned %d records", len(response.pvMetadataResult.pvMetadata)` + - `"QueryConfigurations returned %d records", ...` + - `"QueryConfigurationActivations returned %d records", ...` + - `"GetActiveConfigurations returned %d records", ...` + - `"Successfully saved %d sample status(es)", response.saveSampleStatusesResult.savedCount` + - `"Successfully deleted %d sample status(es)", response.deleteSampleStatusesResult.deletedCount` + - `"Successfully queried %d sample status bucket(s)", ...` + + and the rest name the entity (`"Successfully saved PV metadata for: %s", request.pvName`). A literal + reading of `_dispatch(stub_call, request, result_cls, success_field, op_name)` would silently drop all + 18. Resolved by D2 below. + +- **T4 — The entry-log lines vary the same way.** `"Calling savePvMetadata API for PV: %s", request.pvName` + vs. bare `"Calling querySamples API"` vs. `"Calling saveSampleStatuses API with %d frame(s)"`. Same + problem, same resolution (D2). + +- **T5 — `IngestionClient` alone guards `e.code()`.** Its `except grpc.RpcError` block wraps `e.code()` in + a nested `try/except (AttributeError, TypeError)` with the comment "may not be available in test mocks"; + the other 17 log only `e.details()`. This is not incidental: `test_send_register_provider_grpc_error` + raises a bare `grpc.RpcError()`, on which `code()` genuinely does not resolve. Resolved by D3. + +- **T6 — No unit test asserts on log output.** `grep -rn "caplog\|assertLogs" tests/unit/` returns nothing, + so log-message changes cannot break the suite — which means the suite would *not* have caught T3's + silent log loss. That is an argument for preserving the messages deliberately (D2), not for relying on + the tests to police them. + +- **T7 — The existing tests exercise `_dispatch` as-is.** Every sender test mocks the stub method + (`mock_stub.savePvMetadata.side_effect = ...`) and calls `_send_*` directly. `_dispatch` still calls + that same stub method, so all 406 tests on `main` cover the refactored path with no edits. Baseline + before the change: 406 passed. + +## Design decisions + +- **D1 — `_dispatch` lives on `ServiceApiClientBase`, covering all five clients.** Not on an + annotation-only mixin. The shape is identical in `IngestionClient` and `QueryClient`, the base class + already owns `self._stub` and `self.logger`, and #6's senders inherit it for free. (Recorded in the #6 + plan finding 5 / Q3.) Rejected: a helper module-level function, which would have to be handed the + logger and the stub on every call. + +- **D2 — Optional `request_log` / `success_log` callables preserve every existing log line verbatim.** + `_dispatch` takes `request_log: Callable[[], None] | None` and `success_log: Callable[[Any], None] | None` + (the latter receiving the response). Callers that log a bare message pass nothing and get + `_dispatch`'s default `"Calling API"` / `" completed successfully"`; the ~13 senders + with entity- or count-bearing messages pass a one-line lambda that keeps their exact text. Each sender + ends up 4–8 lines instead of ~40. + + Rejected: (a) dropping the bespoke messages for a uniform one — operators lose record counts and PV + names from INFO-level logs, an observability regression a pure refactor has no business making, and T6 + shows nothing would flag it; (b) moving the detail into the public wrapper methods — no log content is + lost, but it relocates messages to a different call site and grows 18 public methods, which is a larger + diff than the one it avoids. + +- **D3 — The `e.code()` guard becomes shared behavior.** `_dispatch` logs the gRPC code when it resolves + and omits it when it does not, exactly as `IngestionClient` does today. This strictly improves the + other 17 (a code in the log where one exists) and cannot regress any of them, since the message text + returned in the result — `f"gRPC error: {e.details()}"` — is unchanged in every case. Rejected: + dropping the guard, which would break `test_send_register_provider_grpc_error`'s bare `grpc.RpcError()`. + +- **D4 — The two server-streaming senders are out of scope and stay as they are.** Their control flow is + a `for` loop over a response stream that *yields* results (including error results) rather than + returning one, and their documented contract — errors yielded, not raised, for the public `iter_*` + wrapper to convert — is not the unary contract. Forcing both through one helper would parameterize + away the difference that matters. (Consistent with the ticket and the #6 plan.) + +- **D5 — Error-message text is byte-identical to today's.** `f"gRPC error: {e.details()}"`, + `f"Unexpected error: {e!s}"`, and + `f"Unexpected response format: neither exceptionalResult nor {success_field} found"` are reproduced + exactly, because ~40 existing assertions match on these strings with `assertIn`. The refactor is + verified by the unchanged suite; any diff in these strings would be a behavior change smuggled into a + refactor. + +## Implementation tasks + +1. **`src/dp_python_lib/client/service_api_client_base.py`** — add `_dispatch()`, typed + `(stub_call: Callable[[Any], Any], request: Any, result_cls: type[T], success_field: str, + op_name: str, request_log=None, success_log=None) -> T`, with `T` bound to `ApiResultBase`. Implements + the three tiers per D3/D5 and the default log lines per D2. +2. **`pv_metadata_client.py`** — collapse 4 senders. +3. **`machine_config_client.py`** — collapse 9 senders. +4. **`sample_status_client.py`** — collapse 3 senders (leave the streaming one). +5. **`query_client.py`** — collapse 1 sender (leave the streaming one). +6. **`ingestion_client.py`** — collapse 1 sender. +7. **`tests/unit/test_service_api_client_base.py`** — new, direct coverage of `_dispatch` itself: success, + business error, unrecognized response, `grpc.RpcError` with and without a resolvable `code()`, general + exception, and that `request_log`/`success_log` fire (and that omitting them is fine). The 18 senders + are already covered by the existing per-client tests (T7); this file covers the helper's own branches, + including the `code()` guard that no per-client test reaches. +8. **`CLAUDE.md`** — document `_dispatch` under the "Client Implementation Pattern" and "gRPC Error + Handling" guidelines, so #6 and later work write senders in the collapsed form. + +## Out of scope + +- The two server-streaming senders (D4). +- Any change to public method signatures, result classes, or error-message text (D5). +- Relaxing the empty-`values` criterion helpers — [#40](https://github.com/osprey-dcs/dp-python-lib/issues/40). +- #6's own senders — they are written against this helper once it lands, in that ticket. + +## Dependencies and sequencing + +- Blocks nothing hard, but lands **before** #6 by decision (#6 plan Q3) so #6's 10 senders are written + collapsed. Own small PR off `main`. +- Does **not** depend on the stub sync (#39, merged as b6d0b37): `_dispatch` is generated-code-agnostic. +- Does **not** depend on #40 or #17. + +## Open questions + +- **Q1 — How to preserve the bespoke success/entry log messages. RESOLVED 2026-09-09: optional + `request_log`/`success_log` callables (D2).** Confirmed with the repo owner; the alternatives + considered and rejected are recorded in D2. diff --git a/src/dp_python_lib/client/ingestion_client.py b/src/dp_python_lib/client/ingestion_client.py index ad96c75..801a5ca 100644 --- a/src/dp_python_lib/client/ingestion_client.py +++ b/src/dp_python_lib/client/ingestion_client.py @@ -105,46 +105,20 @@ def _build_register_provider_request( def _send_register_provider(self, request: ingestion_pb2.RegisterProviderRequest) -> RegisterProviderApiResult: """ Invokes the registerProvider() API method with the supplied request object. - :param request: RegisgerProviderRequest object with parameters for call to registerProvider(). + :param request: RegisterProviderRequest object with parameters for call to registerProvider(). :return: Returns a RegisterProviderApiResult with the method response and status information. """ - self.logger.info("Calling registerProvider API for provider: %s", request.providerName) - - try: - self.logger.debug("Invoking stub.registerProvider with request") - response = self._stub.registerProvider(request) - self.logger.debug("Received response from registerProvider API") - - # Check if response contains an exceptional result (error) - if response.HasField("exceptionalResult"): - error_msg = response.exceptionalResult.message - self.logger.warning("RegisterProvider API returned business error: %s", error_msg) - return RegisterProviderApiResult(is_error=True, message=error_msg) - - # Check if response contains registration result (success) - elif response.HasField("registrationResult"): - self.logger.info("Successfully registered provider: %s", request.providerName) - return RegisterProviderApiResult(is_error=False, message="", response=response) - - # Unexpected response structure - else: - error_msg = "Unexpected response format: neither exceptionalResult nor registrationResult found" - self.logger.error(error_msg) - return RegisterProviderApiResult(is_error=True, message=error_msg) - - except grpc.RpcError as e: - error_msg = f"gRPC error: {e.details()}" - # Safely get error code - may not be available in test mocks - try: - error_code = e.code() - self.logger.error("gRPC error during registerProvider: %s (code: %s)", e.details(), error_code) - except (AttributeError, TypeError): - self.logger.error("gRPC error during registerProvider: %s", e.details()) - return RegisterProviderApiResult(is_error=True, message=error_msg) - except Exception as e: - error_msg = f"Unexpected error: {e!s}" - self.logger.exception("Unexpected error during registerProvider: %s", str(e)) - return RegisterProviderApiResult(is_error=True, message=error_msg) + return self._dispatch( + self._stub.registerProvider, + request, + RegisterProviderApiResult, + "registrationResult", + "registerProvider", + request_log=lambda: self.logger.info("Calling registerProvider API for provider: %s", request.providerName), + success_log=lambda _response: self.logger.info( + "Successfully registered provider: %s", request.providerName + ), + ) def register_provider(self, request_params: RegisterProviderRequestParams) -> RegisterProviderApiResult: """ diff --git a/src/dp_python_lib/client/machine_config_client.py b/src/dp_python_lib/client/machine_config_client.py index 2b4c6de..aa82b8b 100644 --- a/src/dp_python_lib/client/machine_config_client.py +++ b/src/dp_python_lib/client/machine_config_client.py @@ -701,35 +701,19 @@ def _send_save_configuration(self, request: annotation_pb2.SaveConfigurationRequ :param request: SaveConfigurationRequest with parameters for the call. :return: A SaveConfigurationApiResult with the method response and status information. """ - self.logger.info("Calling saveConfiguration API for configuration: %s", request.configurationName) - - try: - self.logger.debug("Invoking stub.saveConfiguration with request") - response = self._stub.saveConfiguration(request) - self.logger.debug("Received response from saveConfiguration API") - - if response.HasField("exceptionalResult"): - error_msg = response.exceptionalResult.message - self.logger.warning("SaveConfiguration API returned business error: %s", error_msg) - return SaveConfigurationApiResult(is_error=True, message=error_msg) - - elif response.HasField("saveConfigurationResult"): - self.logger.info("Successfully saved configuration: %s", request.configurationName) - return SaveConfigurationApiResult(is_error=False, message="", response=response) - - else: - error_msg = "Unexpected response format: neither exceptionalResult nor saveConfigurationResult found" - self.logger.error(error_msg) - return SaveConfigurationApiResult(is_error=True, message=error_msg) - - except grpc.RpcError as e: - error_msg = f"gRPC error: {e.details()}" - self.logger.error("gRPC error during saveConfiguration: %s", e.details()) - return SaveConfigurationApiResult(is_error=True, message=error_msg) - except Exception as e: - error_msg = f"Unexpected error: {e!s}" - self.logger.exception("Unexpected error during saveConfiguration: %s", str(e)) - return SaveConfigurationApiResult(is_error=True, message=error_msg) + return self._dispatch( + self._stub.saveConfiguration, + request, + SaveConfigurationApiResult, + "saveConfigurationResult", + "saveConfiguration", + request_log=lambda: self.logger.info( + "Calling saveConfiguration API for configuration: %s", request.configurationName + ), + success_log=lambda _response: self.logger.info( + "Successfully saved configuration: %s", request.configurationName + ), + ) def save_configuration(self, request_params: SaveConfigurationRequestParams) -> SaveConfigurationApiResult: """ @@ -776,35 +760,17 @@ def _send_get_configuration(self, request: annotation_pb2.GetConfigurationReques :param request: GetConfigurationRequest with parameters for the call. :return: A GetConfigurationApiResult with the method response and status information. """ - self.logger.info("Calling getConfiguration API for: %s", request.configurationName) - - try: - self.logger.debug("Invoking stub.getConfiguration with request") - response = self._stub.getConfiguration(request) - self.logger.debug("Received response from getConfiguration API") - - if response.HasField("exceptionalResult"): - error_msg = response.exceptionalResult.message - self.logger.warning("GetConfiguration API returned business error: %s", error_msg) - return GetConfigurationApiResult(is_error=True, message=error_msg) - - elif response.HasField("getConfigurationResult"): - self.logger.info("Successfully retrieved configuration: %s", request.configurationName) - return GetConfigurationApiResult(is_error=False, message="", response=response) - - else: - error_msg = "Unexpected response format: neither exceptionalResult nor getConfigurationResult found" - self.logger.error(error_msg) - return GetConfigurationApiResult(is_error=True, message=error_msg) - - except grpc.RpcError as e: - error_msg = f"gRPC error: {e.details()}" - self.logger.error("gRPC error during getConfiguration: %s", e.details()) - return GetConfigurationApiResult(is_error=True, message=error_msg) - except Exception as e: - error_msg = f"Unexpected error: {e!s}" - self.logger.exception("Unexpected error during getConfiguration: %s", str(e)) - return GetConfigurationApiResult(is_error=True, message=error_msg) + return self._dispatch( + self._stub.getConfiguration, + request, + GetConfigurationApiResult, + "getConfigurationResult", + "getConfiguration", + request_log=lambda: self.logger.info("Calling getConfiguration API for: %s", request.configurationName), + success_log=lambda _response: self.logger.info( + "Successfully retrieved configuration: %s", request.configurationName + ), + ) def get_configuration(self, configuration_name: str) -> GetConfigurationApiResult: """ @@ -858,38 +824,20 @@ def _send_query_configurations( :param request: QueryConfigurationsRequest with parameters for the call. :return: A QueryConfigurationsApiResult with the method response and status information. """ - self.logger.info("Calling queryConfigurations API with %d criteria", len(request.criteria)) - - try: - self.logger.debug("Invoking stub.queryConfigurations with request") - response = self._stub.queryConfigurations(request) - self.logger.debug("Received response from queryConfigurations API") - - if response.HasField("exceptionalResult"): - error_msg = response.exceptionalResult.message - self.logger.warning("QueryConfigurations API returned business error: %s", error_msg) - return QueryConfigurationsApiResult(is_error=True, message=error_msg) - - elif response.HasField("queryConfigurationsResult"): - self.logger.info( - "QueryConfigurations returned %d records", - len(response.queryConfigurationsResult.configurations), - ) - return QueryConfigurationsApiResult(is_error=False, message="", response=response) - - else: - error_msg = "Unexpected response format: neither exceptionalResult nor queryConfigurationsResult found" - self.logger.error(error_msg) - return QueryConfigurationsApiResult(is_error=True, message=error_msg) - - except grpc.RpcError as e: - error_msg = f"gRPC error: {e.details()}" - self.logger.error("gRPC error during queryConfigurations: %s", e.details()) - return QueryConfigurationsApiResult(is_error=True, message=error_msg) - except Exception as e: - error_msg = f"Unexpected error: {e!s}" - self.logger.exception("Unexpected error during queryConfigurations: %s", str(e)) - return QueryConfigurationsApiResult(is_error=True, message=error_msg) + return self._dispatch( + self._stub.queryConfigurations, + request, + QueryConfigurationsApiResult, + "queryConfigurationsResult", + "queryConfigurations", + request_log=lambda: self.logger.info( + "Calling queryConfigurations API with %d criteria", len(request.criteria) + ), + success_log=lambda response: self.logger.info( + "QueryConfigurations returned %d records", + len(response.queryConfigurationsResult.configurations), + ), + ) def query_configurations( self, @@ -967,35 +915,17 @@ def _send_delete_configuration( :param request: DeleteConfigurationRequest with parameters for the call. :return: A DeleteConfigurationApiResult with the method response and status information. """ - self.logger.info("Calling deleteConfiguration API for: %s", request.configurationName) - - try: - self.logger.debug("Invoking stub.deleteConfiguration with request") - response = self._stub.deleteConfiguration(request) - self.logger.debug("Received response from deleteConfiguration API") - - if response.HasField("exceptionalResult"): - error_msg = response.exceptionalResult.message - self.logger.warning("DeleteConfiguration API returned business error: %s", error_msg) - return DeleteConfigurationApiResult(is_error=True, message=error_msg) - - elif response.HasField("deleteConfigurationResult"): - self.logger.info("Successfully deleted configuration: %s", request.configurationName) - return DeleteConfigurationApiResult(is_error=False, message="", response=response) - - else: - error_msg = "Unexpected response format: neither exceptionalResult nor deleteConfigurationResult found" - self.logger.error(error_msg) - return DeleteConfigurationApiResult(is_error=True, message=error_msg) - - except grpc.RpcError as e: - error_msg = f"gRPC error: {e.details()}" - self.logger.error("gRPC error during deleteConfiguration: %s", e.details()) - return DeleteConfigurationApiResult(is_error=True, message=error_msg) - except Exception as e: - error_msg = f"Unexpected error: {e!s}" - self.logger.exception("Unexpected error during deleteConfiguration: %s", str(e)) - return DeleteConfigurationApiResult(is_error=True, message=error_msg) + return self._dispatch( + self._stub.deleteConfiguration, + request, + DeleteConfigurationApiResult, + "deleteConfigurationResult", + "deleteConfiguration", + request_log=lambda: self.logger.info("Calling deleteConfiguration API for: %s", request.configurationName), + success_log=lambda _response: self.logger.info( + "Successfully deleted configuration: %s", request.configurationName + ), + ) def delete_configuration(self, configuration_name: str) -> DeleteConfigurationApiResult: """ @@ -1070,41 +1000,21 @@ def _send_save_configuration_activation( :param request: SaveConfigurationActivationRequest with parameters for the call. :return: A SaveConfigurationActivationApiResult with the method response and status information. """ - self.logger.info( - "Calling saveConfigurationActivation API for configuration: %s", - request.configurationName, + return self._dispatch( + self._stub.saveConfigurationActivation, + request, + SaveConfigurationActivationApiResult, + "saveConfigurationActivationResult", + "saveConfigurationActivation", + request_log=lambda: self.logger.info( + "Calling saveConfigurationActivation API for configuration: %s", + request.configurationName, + ), + success_log=lambda _response: self.logger.info( + "Successfully saved configuration activation for: %s", request.configurationName + ), ) - try: - self.logger.debug("Invoking stub.saveConfigurationActivation with request") - response = self._stub.saveConfigurationActivation(request) - self.logger.debug("Received response from saveConfigurationActivation API") - - if response.HasField("exceptionalResult"): - error_msg = response.exceptionalResult.message - self.logger.warning("SaveConfigurationActivation API returned business error: %s", error_msg) - return SaveConfigurationActivationApiResult(is_error=True, message=error_msg) - - elif response.HasField("saveConfigurationActivationResult"): - self.logger.info("Successfully saved configuration activation for: %s", request.configurationName) - return SaveConfigurationActivationApiResult(is_error=False, message="", response=response) - - else: - error_msg = ( - "Unexpected response format: neither exceptionalResult nor saveConfigurationActivationResult found" - ) - self.logger.error(error_msg) - return SaveConfigurationActivationApiResult(is_error=True, message=error_msg) - - except grpc.RpcError as e: - error_msg = f"gRPC error: {e.details()}" - self.logger.error("gRPC error during saveConfigurationActivation: %s", e.details()) - return SaveConfigurationActivationApiResult(is_error=True, message=error_msg) - except Exception as e: - error_msg = f"Unexpected error: {e!s}" - self.logger.exception("Unexpected error during saveConfigurationActivation: %s", str(e)) - return SaveConfigurationActivationApiResult(is_error=True, message=error_msg) - def save_configuration_activation( self, request_params: SaveConfigurationActivationRequestParams ) -> SaveConfigurationActivationApiResult: @@ -1201,37 +1111,14 @@ def _send_get_configuration_activation( :param request: GetConfigurationActivationRequest with parameters for the call. :return: A GetConfigurationActivationApiResult with the method response and status information. """ - self.logger.info("Calling getConfigurationActivation API") - - try: - self.logger.debug("Invoking stub.getConfigurationActivation with request") - response = self._stub.getConfigurationActivation(request) - self.logger.debug("Received response from getConfigurationActivation API") - - if response.HasField("exceptionalResult"): - error_msg = response.exceptionalResult.message - self.logger.warning("GetConfigurationActivation API returned business error: %s", error_msg) - return GetConfigurationActivationApiResult(is_error=True, message=error_msg) - - elif response.HasField("getConfigurationActivationResult"): - self.logger.info("Successfully retrieved configuration activation") - return GetConfigurationActivationApiResult(is_error=False, message="", response=response) - - else: - error_msg = ( - "Unexpected response format: neither exceptionalResult nor getConfigurationActivationResult found" - ) - self.logger.error(error_msg) - return GetConfigurationActivationApiResult(is_error=True, message=error_msg) - - except grpc.RpcError as e: - error_msg = f"gRPC error: {e.details()}" - self.logger.error("gRPC error during getConfigurationActivation: %s", e.details()) - return GetConfigurationActivationApiResult(is_error=True, message=error_msg) - except Exception as e: - error_msg = f"Unexpected error: {e!s}" - self.logger.exception("Unexpected error during getConfigurationActivation: %s", str(e)) - return GetConfigurationActivationApiResult(is_error=True, message=error_msg) + return self._dispatch( + self._stub.getConfigurationActivation, + request, + GetConfigurationActivationApiResult, + "getConfigurationActivationResult", + "getConfigurationActivation", + success_log=lambda _response: self.logger.info("Successfully retrieved configuration activation"), + ) def get_configuration_activation( self, @@ -1295,41 +1182,20 @@ def _send_query_configuration_activations( :param request: QueryConfigurationActivationsRequest with parameters for the call. :return: A QueryConfigurationActivationsApiResult with the method response and status information. """ - self.logger.info("Calling queryConfigurationActivations API with %d criteria", len(request.criteria)) - - try: - self.logger.debug("Invoking stub.queryConfigurationActivations with request") - response = self._stub.queryConfigurationActivations(request) - self.logger.debug("Received response from queryConfigurationActivations API") - - if response.HasField("exceptionalResult"): - error_msg = response.exceptionalResult.message - self.logger.warning("QueryConfigurationActivations API returned business error: %s", error_msg) - return QueryConfigurationActivationsApiResult(is_error=True, message=error_msg) - - elif response.HasField("queryConfigurationActivationsResult"): - self.logger.info( - "QueryConfigurationActivations returned %d records", - len(response.queryConfigurationActivationsResult.configurationActivations), - ) - return QueryConfigurationActivationsApiResult(is_error=False, message="", response=response) - - else: - error_msg = ( - "Unexpected response format: neither exceptionalResult nor " - "queryConfigurationActivationsResult found" - ) - self.logger.error(error_msg) - return QueryConfigurationActivationsApiResult(is_error=True, message=error_msg) - - except grpc.RpcError as e: - error_msg = f"gRPC error: {e.details()}" - self.logger.error("gRPC error during queryConfigurationActivations: %s", e.details()) - return QueryConfigurationActivationsApiResult(is_error=True, message=error_msg) - except Exception as e: - error_msg = f"Unexpected error: {e!s}" - self.logger.exception("Unexpected error during queryConfigurationActivations: %s", str(e)) - return QueryConfigurationActivationsApiResult(is_error=True, message=error_msg) + return self._dispatch( + self._stub.queryConfigurationActivations, + request, + QueryConfigurationActivationsApiResult, + "queryConfigurationActivationsResult", + "queryConfigurationActivations", + request_log=lambda: self.logger.info( + "Calling queryConfigurationActivations API with %d criteria", len(request.criteria) + ), + success_log=lambda response: self.logger.info( + "QueryConfigurationActivations returned %d records", + len(response.queryConfigurationActivationsResult.configurationActivations), + ), + ) def query_configuration_activations( self, @@ -1430,38 +1296,14 @@ def _send_delete_configuration_activation( :param request: DeleteConfigurationActivationRequest with parameters for the call. :return: A DeleteConfigurationActivationApiResult with the method response and status information. """ - self.logger.info("Calling deleteConfigurationActivation API") - - try: - self.logger.debug("Invoking stub.deleteConfigurationActivation with request") - response = self._stub.deleteConfigurationActivation(request) - self.logger.debug("Received response from deleteConfigurationActivation API") - - if response.HasField("exceptionalResult"): - error_msg = response.exceptionalResult.message - self.logger.warning("DeleteConfigurationActivation API returned business error: %s", error_msg) - return DeleteConfigurationActivationApiResult(is_error=True, message=error_msg) - - elif response.HasField("deleteConfigurationActivationResult"): - self.logger.info("Successfully deleted configuration activation") - return DeleteConfigurationActivationApiResult(is_error=False, message="", response=response) - - else: - error_msg = ( - "Unexpected response format: neither exceptionalResult nor " - "deleteConfigurationActivationResult found" - ) - self.logger.error(error_msg) - return DeleteConfigurationActivationApiResult(is_error=True, message=error_msg) - - except grpc.RpcError as e: - error_msg = f"gRPC error: {e.details()}" - self.logger.error("gRPC error during deleteConfigurationActivation: %s", e.details()) - return DeleteConfigurationActivationApiResult(is_error=True, message=error_msg) - except Exception as e: - error_msg = f"Unexpected error: {e!s}" - self.logger.exception("Unexpected error during deleteConfigurationActivation: %s", str(e)) - return DeleteConfigurationActivationApiResult(is_error=True, message=error_msg) + return self._dispatch( + self._stub.deleteConfigurationActivation, + request, + DeleteConfigurationActivationApiResult, + "deleteConfigurationActivationResult", + "deleteConfigurationActivation", + success_log=lambda _response: self.logger.info("Successfully deleted configuration activation"), + ) def delete_configuration_activation( self, @@ -1523,40 +1365,17 @@ def _send_get_active_configurations( :param request: GetActiveConfigurationsRequest with parameters for the call. :return: A GetActiveConfigurationsApiResult with the method response and status information. """ - self.logger.info("Calling getActiveConfigurations API") - - try: - self.logger.debug("Invoking stub.getActiveConfigurations with request") - response = self._stub.getActiveConfigurations(request) - self.logger.debug("Received response from getActiveConfigurations API") - - if response.HasField("exceptionalResult"): - error_msg = response.exceptionalResult.message - self.logger.warning("GetActiveConfigurations API returned business error: %s", error_msg) - return GetActiveConfigurationsApiResult(is_error=True, message=error_msg) - - elif response.HasField("getActiveConfigurationsResult"): - self.logger.info( - "GetActiveConfigurations returned %d records", - len(response.getActiveConfigurationsResult.configurationActivations), - ) - return GetActiveConfigurationsApiResult(is_error=False, message="", response=response) - - else: - error_msg = ( - "Unexpected response format: neither exceptionalResult nor getActiveConfigurationsResult found" - ) - self.logger.error(error_msg) - return GetActiveConfigurationsApiResult(is_error=True, message=error_msg) - - except grpc.RpcError as e: - error_msg = f"gRPC error: {e.details()}" - self.logger.error("gRPC error during getActiveConfigurations: %s", e.details()) - return GetActiveConfigurationsApiResult(is_error=True, message=error_msg) - except Exception as e: - error_msg = f"Unexpected error: {e!s}" - self.logger.exception("Unexpected error during getActiveConfigurations: %s", str(e)) - return GetActiveConfigurationsApiResult(is_error=True, message=error_msg) + return self._dispatch( + self._stub.getActiveConfigurations, + request, + GetActiveConfigurationsApiResult, + "getActiveConfigurationsResult", + "getActiveConfigurations", + success_log=lambda response: self.logger.info( + "GetActiveConfigurations returned %d records", + len(response.getActiveConfigurationsResult.configurationActivations), + ), + ) def get_active_configurations(self, timestamp: TimestampInput | None = None) -> GetActiveConfigurationsApiResult: """ diff --git a/src/dp_python_lib/client/pv_metadata_client.py b/src/dp_python_lib/client/pv_metadata_client.py index 3edfae9..49fd5d4 100644 --- a/src/dp_python_lib/client/pv_metadata_client.py +++ b/src/dp_python_lib/client/pv_metadata_client.py @@ -313,35 +313,15 @@ def _send_save_pv_metadata(self, request: annotation_pb2.SavePvMetadataRequest) :param request: SavePvMetadataRequest with parameters for the call. :return: A SavePvMetadataApiResult with the method response and status information. """ - self.logger.info("Calling savePvMetadata API for PV: %s", request.pvName) - - try: - self.logger.debug("Invoking stub.savePvMetadata with request") - response = self._stub.savePvMetadata(request) - self.logger.debug("Received response from savePvMetadata API") - - if response.HasField("exceptionalResult"): - error_msg = response.exceptionalResult.message - self.logger.warning("SavePvMetadata API returned business error: %s", error_msg) - return SavePvMetadataApiResult(is_error=True, message=error_msg) - - elif response.HasField("savePvMetadataResult"): - self.logger.info("Successfully saved PV metadata for: %s", request.pvName) - return SavePvMetadataApiResult(is_error=False, message="", response=response) - - else: - error_msg = "Unexpected response format: neither exceptionalResult nor savePvMetadataResult found" - self.logger.error(error_msg) - return SavePvMetadataApiResult(is_error=True, message=error_msg) - - except grpc.RpcError as e: - error_msg = f"gRPC error: {e.details()}" - self.logger.error("gRPC error during savePvMetadata: %s", e.details()) - return SavePvMetadataApiResult(is_error=True, message=error_msg) - except Exception as e: - error_msg = f"Unexpected error: {e!s}" - self.logger.exception("Unexpected error during savePvMetadata: %s", str(e)) - return SavePvMetadataApiResult(is_error=True, message=error_msg) + return self._dispatch( + self._stub.savePvMetadata, + request, + SavePvMetadataApiResult, + "savePvMetadataResult", + "savePvMetadata", + request_log=lambda: self.logger.info("Calling savePvMetadata API for PV: %s", request.pvName), + success_log=lambda _response: self.logger.info("Successfully saved PV metadata for: %s", request.pvName), + ) def save_pv_metadata(self, request_params: SavePvMetadataRequestParams) -> SavePvMetadataApiResult: """ @@ -382,35 +362,17 @@ def _send_get_pv_metadata(self, request: annotation_pb2.GetPvMetadataRequest) -> :param request: GetPvMetadataRequest with parameters for the call. :return: A GetPvMetadataApiResult with the method response and status information. """ - self.logger.info("Calling getPvMetadata API for: %s", request.pvNameOrAlias) - - try: - self.logger.debug("Invoking stub.getPvMetadata with request") - response = self._stub.getPvMetadata(request) - self.logger.debug("Received response from getPvMetadata API") - - if response.HasField("exceptionalResult"): - error_msg = response.exceptionalResult.message - self.logger.warning("GetPvMetadata API returned business error: %s", error_msg) - return GetPvMetadataApiResult(is_error=True, message=error_msg) - - elif response.HasField("getPvMetadataResult"): - self.logger.info("Successfully retrieved PV metadata for: %s", request.pvNameOrAlias) - return GetPvMetadataApiResult(is_error=False, message="", response=response) - - else: - error_msg = "Unexpected response format: neither exceptionalResult nor getPvMetadataResult found" - self.logger.error(error_msg) - return GetPvMetadataApiResult(is_error=True, message=error_msg) - - except grpc.RpcError as e: - error_msg = f"gRPC error: {e.details()}" - self.logger.error("gRPC error during getPvMetadata: %s", e.details()) - return GetPvMetadataApiResult(is_error=True, message=error_msg) - except Exception as e: - error_msg = f"Unexpected error: {e!s}" - self.logger.exception("Unexpected error during getPvMetadata: %s", str(e)) - return GetPvMetadataApiResult(is_error=True, message=error_msg) + return self._dispatch( + self._stub.getPvMetadata, + request, + GetPvMetadataApiResult, + "getPvMetadataResult", + "getPvMetadata", + request_log=lambda: self.logger.info("Calling getPvMetadata API for: %s", request.pvNameOrAlias), + success_log=lambda _response: self.logger.info( + "Successfully retrieved PV metadata for: %s", request.pvNameOrAlias + ), + ) def get_pv_metadata(self, pv_name_or_alias: str) -> GetPvMetadataApiResult: """ @@ -462,35 +424,17 @@ def _send_query_pv_metadata(self, request: annotation_pb2.QueryPvMetadataRequest :param request: QueryPvMetadataRequest with parameters for the call. :return: A QueryPvMetadataApiResult with the method response and status information. """ - self.logger.info("Calling queryPvMetadata API with %d criteria", len(request.criteria)) - - try: - self.logger.debug("Invoking stub.queryPvMetadata with request") - response = self._stub.queryPvMetadata(request) - self.logger.debug("Received response from queryPvMetadata API") - - if response.HasField("exceptionalResult"): - error_msg = response.exceptionalResult.message - self.logger.warning("QueryPvMetadata API returned business error: %s", error_msg) - return QueryPvMetadataApiResult(is_error=True, message=error_msg) - - elif response.HasField("pvMetadataResult"): - self.logger.info("QueryPvMetadata returned %d records", len(response.pvMetadataResult.pvMetadata)) - return QueryPvMetadataApiResult(is_error=False, message="", response=response) - - else: - error_msg = "Unexpected response format: neither exceptionalResult nor pvMetadataResult found" - self.logger.error(error_msg) - return QueryPvMetadataApiResult(is_error=True, message=error_msg) - - except grpc.RpcError as e: - error_msg = f"gRPC error: {e.details()}" - self.logger.error("gRPC error during queryPvMetadata: %s", e.details()) - return QueryPvMetadataApiResult(is_error=True, message=error_msg) - except Exception as e: - error_msg = f"Unexpected error: {e!s}" - self.logger.exception("Unexpected error during queryPvMetadata: %s", str(e)) - return QueryPvMetadataApiResult(is_error=True, message=error_msg) + return self._dispatch( + self._stub.queryPvMetadata, + request, + QueryPvMetadataApiResult, + "pvMetadataResult", + "queryPvMetadata", + request_log=lambda: self.logger.info("Calling queryPvMetadata API with %d criteria", len(request.criteria)), + success_log=lambda response: self.logger.info( + "QueryPvMetadata returned %d records", len(response.pvMetadataResult.pvMetadata) + ), + ) def query_pv_metadata( self, @@ -566,35 +510,17 @@ def _send_delete_pv_metadata(self, request: annotation_pb2.DeletePvMetadataReque :param request: DeletePvMetadataRequest with parameters for the call. :return: A DeletePvMetadataApiResult with the method response and status information. """ - self.logger.info("Calling deletePvMetadata API for: %s", request.pvNameOrAlias) - - try: - self.logger.debug("Invoking stub.deletePvMetadata with request") - response = self._stub.deletePvMetadata(request) - self.logger.debug("Received response from deletePvMetadata API") - - if response.HasField("exceptionalResult"): - error_msg = response.exceptionalResult.message - self.logger.warning("DeletePvMetadata API returned business error: %s", error_msg) - return DeletePvMetadataApiResult(is_error=True, message=error_msg) - - elif response.HasField("deletePvMetadataResult"): - self.logger.info("Successfully deleted PV metadata for: %s", request.pvNameOrAlias) - return DeletePvMetadataApiResult(is_error=False, message="", response=response) - - else: - error_msg = "Unexpected response format: neither exceptionalResult nor deletePvMetadataResult found" - self.logger.error(error_msg) - return DeletePvMetadataApiResult(is_error=True, message=error_msg) - - except grpc.RpcError as e: - error_msg = f"gRPC error: {e.details()}" - self.logger.error("gRPC error during deletePvMetadata: %s", e.details()) - return DeletePvMetadataApiResult(is_error=True, message=error_msg) - except Exception as e: - error_msg = f"Unexpected error: {e!s}" - self.logger.exception("Unexpected error during deletePvMetadata: %s", str(e)) - return DeletePvMetadataApiResult(is_error=True, message=error_msg) + return self._dispatch( + self._stub.deletePvMetadata, + request, + DeletePvMetadataApiResult, + "deletePvMetadataResult", + "deletePvMetadata", + request_log=lambda: self.logger.info("Calling deletePvMetadata API for: %s", request.pvNameOrAlias), + success_log=lambda _response: self.logger.info( + "Successfully deleted PV metadata for: %s", request.pvNameOrAlias + ), + ) def delete_pv_metadata(self, pv_name_or_alias: str) -> DeletePvMetadataApiResult: """ diff --git a/src/dp_python_lib/client/query_client.py b/src/dp_python_lib/client/query_client.py index 8cd4b4f..2580d92 100644 --- a/src/dp_python_lib/client/query_client.py +++ b/src/dp_python_lib/client/query_client.py @@ -588,35 +588,14 @@ def _send_query_samples(self, request: query_pb2.QuerySamplesRequest) -> QuerySa :param request: QuerySamplesRequest with parameters for the call. :return: A QuerySamplesApiResult with the method response and status information. """ - self.logger.info("Calling querySamples API") - - try: - self.logger.debug("Invoking stub.querySamples with request") - response = self._stub.querySamples(request) - self.logger.debug("Received response from querySamples API") - - if response.HasField("exceptionalResult"): - error_msg = response.exceptionalResult.message - self.logger.warning("QuerySamples API returned business error: %s", error_msg) - return QuerySamplesApiResult(is_error=True, message=error_msg) - - elif response.HasField("sampleQueryResult"): - self.logger.info("QuerySamples returned a result page") - return QuerySamplesApiResult(is_error=False, message="", response=response) - - else: - error_msg = "Unexpected response format: neither exceptionalResult nor sampleQueryResult found" - self.logger.error(error_msg) - return QuerySamplesApiResult(is_error=True, message=error_msg) - - except grpc.RpcError as e: - error_msg = f"gRPC error: {e.details()}" - self.logger.error("gRPC error during querySamples: %s", e.details()) - return QuerySamplesApiResult(is_error=True, message=error_msg) - except Exception as e: - error_msg = f"Unexpected error: {e!s}" - self.logger.exception("Unexpected error during querySamples: %s", str(e)) - return QuerySamplesApiResult(is_error=True, message=error_msg) + return self._dispatch( + self._stub.querySamples, + request, + QuerySamplesApiResult, + "sampleQueryResult", + "querySamples", + success_log=lambda _response: self.logger.info("QuerySamples returned a result page"), + ) def query_samples(self, request_params: QueryParams, page_token: str | None = None) -> QuerySamplesApiResult: """ diff --git a/src/dp_python_lib/client/sample_status_client.py b/src/dp_python_lib/client/sample_status_client.py index cd441f6..25b0d75 100644 --- a/src/dp_python_lib/client/sample_status_client.py +++ b/src/dp_python_lib/client/sample_status_client.py @@ -501,38 +501,19 @@ def _send_save_sample_statuses( :param request: SaveSampleStatusesRequest with parameters for the call. :return: A SaveSampleStatusesApiResult with the method response and status information. """ - self.logger.info("Calling saveSampleStatuses API with %d frame(s)", len(request.frames)) - - try: - self.logger.debug("Invoking stub.saveSampleStatuses with request") - response = self._stub.saveSampleStatuses(request) - self.logger.debug("Received response from saveSampleStatuses API") - - if response.HasField("exceptionalResult"): - error_msg = response.exceptionalResult.message - self.logger.warning("SaveSampleStatuses API returned business error: %s", error_msg) - return SaveSampleStatusesApiResult(is_error=True, message=error_msg) - - elif response.HasField("saveSampleStatusesResult"): - self.logger.info( - "Successfully saved %d sample status(es)", response.saveSampleStatusesResult.savedCount - ) - return SaveSampleStatusesApiResult(is_error=False, message="", response=response) - - else: - error_msg = "Unexpected response format: neither exceptionalResult nor saveSampleStatusesResult found" - self.logger.error(error_msg) - return SaveSampleStatusesApiResult(is_error=True, message=error_msg) - - except grpc.RpcError as e: - error_msg = f"gRPC error: {e.details()}" - self.logger.error("gRPC error during saveSampleStatuses: %s", e.details()) - return SaveSampleStatusesApiResult(is_error=True, message=error_msg) - - except Exception as e: - error_msg = f"Unexpected error: {e!s}" - self.logger.exception("Unexpected error during saveSampleStatuses: %s", str(e)) - return SaveSampleStatusesApiResult(is_error=True, message=error_msg) + return self._dispatch( + self._stub.saveSampleStatuses, + request, + SaveSampleStatusesApiResult, + "saveSampleStatusesResult", + "saveSampleStatuses", + request_log=lambda: self.logger.info( + "Calling saveSampleStatuses API with %d frame(s)", len(request.frames) + ), + success_log=lambda response: self.logger.info( + "Successfully saved %d sample status(es)", response.saveSampleStatusesResult.savedCount + ), + ) def save_sample_statuses(self, request_params: SaveSampleStatusesRequestParams) -> SaveSampleStatusesApiResult: """ @@ -604,39 +585,17 @@ def _send_query_sample_statuses( :param request: QuerySampleStatusesRequest with parameters for the call. :return: A QuerySampleStatusesApiResult with the method response and status information. """ - self.logger.info("Calling querySampleStatuses API") - - try: - self.logger.debug("Invoking stub.querySampleStatuses with request") - response = self._stub.querySampleStatuses(request) - self.logger.debug("Received response from querySampleStatuses API") - - if response.HasField("exceptionalResult"): - error_msg = response.exceptionalResult.message - self.logger.warning("QuerySampleStatuses API returned business error: %s", error_msg) - return QuerySampleStatusesApiResult(is_error=True, message=error_msg) - - elif response.HasField("querySampleStatusesResult"): - self.logger.info( - "Successfully queried %d sample status bucket(s)", - len(response.querySampleStatusesResult.sampleStatusBuckets), - ) - return QuerySampleStatusesApiResult(is_error=False, message="", response=response) - - else: - error_msg = "Unexpected response format: neither exceptionalResult nor querySampleStatusesResult found" - self.logger.error(error_msg) - return QuerySampleStatusesApiResult(is_error=True, message=error_msg) - - except grpc.RpcError as e: - error_msg = f"gRPC error: {e.details()}" - self.logger.error("gRPC error during querySampleStatuses: %s", e.details()) - return QuerySampleStatusesApiResult(is_error=True, message=error_msg) - - except Exception as e: - error_msg = f"Unexpected error: {e!s}" - self.logger.exception("Unexpected error during querySampleStatuses: %s", str(e)) - return QuerySampleStatusesApiResult(is_error=True, message=error_msg) + return self._dispatch( + self._stub.querySampleStatuses, + request, + QuerySampleStatusesApiResult, + "querySampleStatusesResult", + "querySampleStatuses", + success_log=lambda response: self.logger.info( + "Successfully queried %d sample status bucket(s)", + len(response.querySampleStatusesResult.sampleStatusBuckets), + ), + ) def query_sample_statuses( self, @@ -828,38 +787,19 @@ def _send_delete_sample_statuses( :param request: DeleteSampleStatusesRequest with parameters for the call. :return: A DeleteSampleStatusesApiResult with the method response and status information. """ - self.logger.info("Calling deleteSampleStatuses API for domain=%s layer=%s", request.domain, request.layer) - - try: - self.logger.debug("Invoking stub.deleteSampleStatuses with request") - response = self._stub.deleteSampleStatuses(request) - self.logger.debug("Received response from deleteSampleStatuses API") - - if response.HasField("exceptionalResult"): - error_msg = response.exceptionalResult.message - self.logger.warning("DeleteSampleStatuses API returned business error: %s", error_msg) - return DeleteSampleStatusesApiResult(is_error=True, message=error_msg) - - elif response.HasField("deleteSampleStatusesResult"): - self.logger.info( - "Successfully deleted %d sample status(es)", response.deleteSampleStatusesResult.deletedCount - ) - return DeleteSampleStatusesApiResult(is_error=False, message="", response=response) - - else: - error_msg = "Unexpected response format: neither exceptionalResult nor deleteSampleStatusesResult found" - self.logger.error(error_msg) - return DeleteSampleStatusesApiResult(is_error=True, message=error_msg) - - except grpc.RpcError as e: - error_msg = f"gRPC error: {e.details()}" - self.logger.error("gRPC error during deleteSampleStatuses: %s", e.details()) - return DeleteSampleStatusesApiResult(is_error=True, message=error_msg) - - except Exception as e: - error_msg = f"Unexpected error: {e!s}" - self.logger.exception("Unexpected error during deleteSampleStatuses: %s", str(e)) - return DeleteSampleStatusesApiResult(is_error=True, message=error_msg) + return self._dispatch( + self._stub.deleteSampleStatuses, + request, + DeleteSampleStatusesApiResult, + "deleteSampleStatusesResult", + "deleteSampleStatuses", + request_log=lambda: self.logger.info( + "Calling deleteSampleStatuses API for domain=%s layer=%s", request.domain, request.layer + ), + success_log=lambda response: self.logger.info( + "Successfully deleted %d sample status(es)", response.deleteSampleStatusesResult.deletedCount + ), + ) def delete_sample_statuses( self, diff --git a/src/dp_python_lib/client/service_api_client_base.py b/src/dp_python_lib/client/service_api_client_base.py index 6edf08e..5f2775c 100644 --- a/src/dp_python_lib/client/service_api_client_base.py +++ b/src/dp_python_lib/client/service_api_client_base.py @@ -1,10 +1,27 @@ import logging from abc import ABC from collections.abc import Callable -from typing import Any +from typing import Any, Protocol, TypeVar import grpc +from .result import ApiResultBase + +ApiResultT = TypeVar("ApiResultT", bound=ApiResultBase, covariant=True) + + +class ApiResultFactory(Protocol[ApiResultT]): + """ + The constructor signature every concrete *ApiResult class shares. ApiResultBase itself takes only + (is_error, message); the 'response' keyword is added by each subclass, which declares it with the precise + response type for its own API method. _dispatch constructs result objects generically, so it is that shared + three-argument shape -- not the base class -- that it actually depends on, and this Protocol is what states it. + Typing result_cls with it keeps the checker honest about the 'response' keyword, which a bare + type[ApiResultT] would reject. + """ + + def __call__(self, is_error: bool, message: str, response: Any = None) -> ApiResultT: ... + class ServiceApiClientBase(ABC): """ @@ -23,3 +40,86 @@ def __init__(self, channel: grpc.Channel, stub_class: Callable[[grpc.Channel], A self._channel = channel self._stub = stub_class(channel) self.logger.debug("Initialized service client with channel: %s, stub: %s", channel, stub_class.__name__) + + def _dispatch( + self, + stub_call: Callable[[Any], Any], + request: Any, + result_cls: ApiResultFactory[ApiResultT], + success_field: str, + op_name: str, + request_log: Callable[[], None] | None = None, + success_log: Callable[[Any], None] | None = None, + ) -> ApiResultT: + """ + Invokes a unary gRPC API method and applies the standard three-tier error handling shared by every unary + _send_* method in the library: + + 1. gRPC exceptions (grpc.RpcError) -- network/connection errors. + 2. Business logic errors -- the response carries an 'exceptionalResult'. + 3. General exceptions -- anything else raised while making the call. + + A response carrying neither 'exceptionalResult' nor the expected success field is itself an error, so an + unrecognized response shape is never mistaken for a success. + + The server-streaming senders deliberately do not use this helper: they yield one result per streamed message + (error results included, for the public iter_* wrapper to convert), which is a different contract from + returning a single result. + + :param stub_call: The bound stub method to invoke, e.g. self._stub.savePvMetadata. + :param request: The request message to pass to stub_call. + :param result_cls: The *ApiResult class to construct; all of them take (is_error, message, response), + the shape stated by the ApiResultFactory protocol. + :param success_field: Name of the response's success oneof field, e.g. "savePvMetadataResult". + :param op_name: API method name used in log and error messages, e.g. "savePvMetadata". + :param request_log: Optional callable logging a method-specific message before the call. When omitted, a + generic "Calling API" is logged instead. + :param success_log: Optional callable, passed the response, logging a method-specific success message (some + methods report a record count read off the response). When omitted, a generic message is logged instead. + :return: A result_cls instance with the method response and status information. + """ + if request_log is not None: + request_log() + else: + self.logger.info("Calling %s API", op_name) + + try: + self.logger.debug("Invoking stub.%s with request", op_name) + response = stub_call(request) + self.logger.debug("Received response from %s API", op_name) + + if response.HasField("exceptionalResult"): + error_msg = response.exceptionalResult.message + # op_name is used verbatim here, so all three error tiers name the operation the same way. The + # hand-written senders capitalized it on this line only ("SavePvMetadata API returned business + # error"), while their gRPC-error and unexpected-error logs already used the camelCase name; the + # inconsistency was not worth preserving. Log text only -- the returned message is unchanged. + self.logger.warning("%s API returned business error: %s", op_name, error_msg) + return result_cls(is_error=True, message=error_msg) + + elif response.HasField(success_field): + if success_log is not None: + success_log(response) + else: + self.logger.info("%s completed successfully", op_name) + return result_cls(is_error=False, message="", response=response) + + else: + error_msg = f"Unexpected response format: neither exceptionalResult nor {success_field} found" + self.logger.error(error_msg) + return result_cls(is_error=True, message=error_msg) + + except grpc.RpcError as e: + error_msg = f"gRPC error: {e.details()}" + # Safely get the error code -- it may not be available on a bare RpcError, as raised by test mocks. + try: + error_code = e.code() + self.logger.error("gRPC error during %s: %s (code: %s)", op_name, e.details(), error_code) + except (AttributeError, TypeError): + self.logger.error("gRPC error during %s: %s", op_name, e.details()) + return result_cls(is_error=True, message=error_msg) + + except Exception as e: + error_msg = f"Unexpected error: {e!s}" + self.logger.exception("Unexpected error during %s: %s", op_name, str(e)) + return result_cls(is_error=True, message=error_msg) diff --git a/tests/unit/test_service_api_client_base.py b/tests/unit/test_service_api_client_base.py new file mode 100644 index 0000000..6d2f1d9 --- /dev/null +++ b/tests/unit/test_service_api_client_base.py @@ -0,0 +1,177 @@ +import logging +import os +import sys +import unittest +from unittest.mock import Mock + +import grpc + +# Add src directory to path for imports +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "../../src")) + +from dp_python_lib.client.result import ApiResultBase +from dp_python_lib.client.service_api_client_base import ServiceApiClientBase + + +class _FakeApiResult(ApiResultBase): + """ + Stand-in for the concrete *ApiResult classes, which all share the (is_error, message, response) constructor + that _dispatch relies on. + """ + + def __init__(self, is_error: bool, message: str, response=None) -> None: + super().__init__(is_error, message) + self.response = response + + +class _FakeClient(ServiceApiClientBase): + """Minimal concrete client, so _dispatch can be exercised without a real service stub.""" + + def __init__(self, channel) -> None: + # A factory that ignores the channel: _dispatch is handed the bound stub method by the caller, so the + # stub instance this creates is never exercised. + super().__init__(channel, lambda _channel: Mock()) + + +def _response_with_field(field_name): + """Build a Mock response whose HasField() returns True only for field_name.""" + response = Mock() + response.HasField = Mock(side_effect=lambda field: field == field_name) + return response + + +class TestDispatch(unittest.TestCase): + """ + Unit tests for ServiceApiClientBase._dispatch, the shared three-tier sender. The 18 unary _send_* methods that + delegate to it are covered by their own per-client test modules; these tests cover the helper's own branches, + including the RpcError code() guard that no per-client test reaches. + """ + + def setUp(self): + self.client = _FakeClient(Mock()) + self.request = Mock() + + def _dispatch(self, stub_call, **kwargs): + return self.client._dispatch( + stub_call, + self.request, + _FakeApiResult, + "someResult", + "someOperation", + **kwargs, + ) + + def test_success_returns_result_wrapping_response(self): + response = _response_with_field("someResult") + stub_call = Mock(return_value=response) + + result = self._dispatch(stub_call) + + stub_call.assert_called_once_with(self.request) + self.assertIsInstance(result, _FakeApiResult) + self.assertFalse(result.result_status.is_error) + self.assertEqual("", result.result_status.message) + self.assertIs(response, result.response) + + def test_business_error_returns_exceptional_result_message(self): + response = _response_with_field("exceptionalResult") + response.exceptionalResult.message = "PV not found" + stub_call = Mock(return_value=response) + + result = self._dispatch(stub_call) + + self.assertTrue(result.result_status.is_error) + self.assertEqual("PV not found", result.result_status.message) + self.assertIsNone(result.response) + + def test_unrecognized_response_is_an_error_naming_the_success_field(self): + stub_call = Mock(return_value=_response_with_field("somethingElse")) + + result = self._dispatch(stub_call) + + self.assertTrue(result.result_status.is_error) + self.assertIn("Unexpected response format", result.result_status.message) + self.assertIn("someResult", result.result_status.message) + self.assertIsNone(result.response) + + def test_grpc_error_returns_details_message(self): + error = grpc.RpcError() + error.details = Mock(return_value="Connection timeout") + error.code = Mock(return_value="UNAVAILABLE") + stub_call = Mock(side_effect=error) + + result = self._dispatch(stub_call) + + self.assertTrue(result.result_status.is_error) + self.assertEqual("gRPC error: Connection timeout", result.result_status.message) + self.assertIsNone(result.response) + + def test_grpc_error_without_resolvable_code_still_returns_details_message(self): + # A bare RpcError has no usable code(); the helper must log without it rather than raise. Test mocks raise + # exactly this shape, which is why the guard exists. + error = grpc.RpcError() + error.details = Mock(return_value="Connection timeout") + stub_call = Mock(side_effect=error) + + with self.assertLogs("dp_python_lib.client.service_api_client_base", level=logging.ERROR): + result = self._dispatch(stub_call) + + self.assertTrue(result.result_status.is_error) + self.assertEqual("gRPC error: Connection timeout", result.result_status.message) + + def test_unexpected_exception_returns_unexpected_error_message(self): + stub_call = Mock(side_effect=ValueError("Invalid parameter")) + + result = self._dispatch(stub_call) + + self.assertTrue(result.result_status.is_error) + self.assertEqual("Unexpected error: Invalid parameter", result.result_status.message) + self.assertIsNone(result.response) + + def test_request_and_success_logs_are_invoked(self): + response = _response_with_field("someResult") + stub_call = Mock(return_value=response) + request_log = Mock() + success_log = Mock() + + self._dispatch(stub_call, request_log=request_log, success_log=success_log) + + request_log.assert_called_once_with() + # The success log is handed the response, so senders can report counts read off the result. + success_log.assert_called_once_with(response) + + def test_success_log_not_invoked_on_error(self): + response = _response_with_field("exceptionalResult") + response.exceptionalResult.message = "boom" + request_log = Mock() + success_log = Mock() + + self._dispatch(Mock(return_value=response), request_log=request_log, success_log=success_log) + + request_log.assert_called_once_with() + success_log.assert_not_called() + + def test_business_error_log_names_the_operation_with_the_raw_op_name(self): + # All three error tiers name the operation with the camelCase op_name. The hand-written senders this + # helper replaced capitalized it in this one log line only; asserting the format keeps the choice + # deliberate, since nothing else in the suite covers log content. + response = _response_with_field("exceptionalResult") + response.exceptionalResult.message = "boom" + + with self.assertLogs("dp_python_lib.client.service_api_client_base", level=logging.WARNING) as captured: + self._dispatch(Mock(return_value=response)) + + self.assertIn("someOperation API returned business error: boom", "\n".join(captured.output)) + + def test_default_logs_used_when_callables_omitted(self): + stub_call = Mock(return_value=_response_with_field("someResult")) + + with self.assertLogs("dp_python_lib.client.service_api_client_base", level=logging.INFO) as captured: + self._dispatch(stub_call) + + self.assertIn("Calling someOperation API", "\n".join(captured.output)) + self.assertIn("someOperation completed successfully", "\n".join(captured.output)) + + +if __name__ == "__main__": + unittest.main()