diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 35ea6a8..6794c1e 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -20,32 +20,11 @@ permissions: env: # rhosocial-activerecord and rhosocial-activerecord-testsuite dev releases are # not published to PyPI; install them from git. - # The schema-object refactor this branch depends on was developed on the core - # library's expression-schema-object-refactor branch and merged to - # release/v1.0.0.dev30 (PR #125), and the introspection ownership rule this - # branch completes was merged there too (PR #128). CORE_PKG therefore pins the - # release branch, and TESTSUITE_PKG pins the matching testsuite release branch, - # so the two move together. Two of the - # four names a previous revision of this comment listed -- QualifiedNameMixin - # and CatalogSupport -- are ones that work *deleted*, along with the - # single-`supports_schema_qualification` object model this backend used to - # assume. What the dialect actually imports from core is: - # - # * the per-kind object mixins and NamespaceMixin, which supply - # format__object for the fourteen kinds that carry a namespace and - # the levels behind them (a Database is the one kind that does not -- it - # is never qualified, so format_database_object is name only); - # * RelationSourceMixin, which renders the FROM side of a named object -- - # without it, rendering any aliased row source raises AttributeError; - # * the object protocols (ViewObjectSupport, IndexObjectSupport), which - # derive from NamespaceSupport and so must follow this backend's own - # namespace mixin in the base list (C3). - # - # The previous pin, the fix/schema-name-propagation-gaps branch, predates all - # of that layer, so the imports this backend now makes do not resolve against - # it. - CORE_PKG: "rhosocial-activerecord @ git+https://github.com/rhosocial/python-activerecord.git@release/v1.0.0.dev30" - TESTSUITE_PKG: "rhosocial-activerecord-testsuite @ git+https://github.com/rhosocial/python-activerecord-testsuite.git@release/v1.0.0.dev17" + # This branch pins the core and testsuite feature branches so the error-taxonomy + # and execute-contract changes under review are installed together. Restore the + # release-branch pins once those branches are merged. + CORE_PKG: "rhosocial-activerecord @ git+https://github.com/rhosocial/python-activerecord.git@feature/error-taxonomy" + TESTSUITE_PKG: "rhosocial-activerecord-testsuite @ git+https://github.com/rhosocial/python-activerecord-testsuite.git@feature/error-taxonomy" jobs: test: diff --git a/changelog.d/10.added.md b/changelog.d/10.added.md new file mode 100644 index 0000000..399466f --- /dev/null +++ b/changelog.d/10.added.md @@ -0,0 +1 @@ +Classified BigQuery driver errors into the semantic error taxonomy (data, record, policy, concurrency, infrastructure) and aligned `execute` / `execute_many` with the execute-phase error contract: framework errors pass through, auto-commit failures raise `TransactionError`, and only transient connection failures are retried. Added contract tests covering the mapping and phases. diff --git a/src/rhosocial/activerecord/backend/impl/bigquery/backend/async_backend.py b/src/rhosocial/activerecord/backend/impl/bigquery/backend/async_backend.py index 46f3139..1d82077 100644 --- a/src/rhosocial/activerecord/backend/impl/bigquery/backend/async_backend.py +++ b/src/rhosocial/activerecord/backend/impl/bigquery/backend/async_backend.py @@ -1,12 +1,12 @@ """BigQuery asynchronous backend implementation.""" import logging -from typing import Any, Dict, List, Optional, Tuple +from typing import Any, Dict, Optional, Tuple from rhosocial.activerecord.backend.base import AsyncStorageBackend from rhosocial.activerecord.backend.base.execution_logging import ( log_execution_error, log_execution_start, ) -from rhosocial.activerecord.backend.errors import ConnectionError, DatabaseError +from rhosocial.activerecord.backend.errors import DatabaseError from rhosocial.activerecord.backend.options import ExecutionOptions, LogContext from rhosocial.activerecord.backend.result import QueryResult from rhosocial.activerecord.backend.schema import StatementType @@ -18,7 +18,10 @@ from ..config import BigQueryConnectionConfig from ..dialect import BigQueryDialect from ..async_transaction import AsyncBigQueryTransactionManager -from .backend import log_result_summary +from .backend import ( + _classify_bigquery_error, _translate_bigquery_error, + _translate_connect_error, _translate_job_errors, log_result_summary, +) from .qualified_names import BigQueryQualifiedNameMixin @@ -85,7 +88,9 @@ async def connect(self) -> None: self._connected = True self.log(logging.DEBUG, "Connected to BigQuery") except Exception as e: - raise ConnectionError(f"Failed to connect to BigQuery: {e}") from e + # The connection phase always surfaces ``ConnectionError``; the + # translator only decides whether the failure is transient. + raise _translate_connect_error(e) from e async def disconnect(self) -> None: if hasattr(self, '_client') and self._client: @@ -149,6 +154,18 @@ async def execute( is replaced (there is no DB-API cursor), so this method emits the shared ``DEBUG`` SQL record and ``INFO`` summary through :mod:`rhosocial.activerecord.backend.base.execution_logging`. + + Transaction semantics: every BigQuery query job commits its own + statement asynchronously, so there is no connection-level + commit/autocommit hook and the auto-commit phase cannot raise + ``TransactionError``. Explicit transactions issue ``BEGIN + TRANSACTION`` / ``COMMIT TRANSACTION`` through this method (see + :class:`~..async_transaction.AsyncBigQueryTransactionManager`). + + Raises: + DatabaseError: the phase-specific taxonomy error: driver + failures are mapped by :func:`_translate_bigquery_error` and + framework errors pass through unchanged. """ from .backend import BigQueryBackend @@ -183,7 +200,9 @@ async def execute( exc = job.exception() if exc is not None: raise exc - raise DatabaseError(str(job.errors)) + # No driver exception to chain (emulator-style job errors): + # classify the message like a status-less driver reply. + raise _translate_job_errors(job.errors) if is_select: rows_iter = job.result() # See BigQueryBackend.execute: rows are converted to dicts and @@ -207,6 +226,9 @@ async def execute( except Exception as e: log_execution_error(self, e, options) self._handle_error(e) + # _handle_error() is required to raise. The fallback keeps a + # backend that forgot from turning a failure into a None result. + raise DatabaseError(str(e)) from e # -- Client-side PK generation (see BigQueryBackend for rationale) ---------- @@ -294,7 +316,7 @@ async def bulk_update(self, options) -> QueryResult: assignments = {} for field_name, values in options.field_values.items(): cases = [] - for pk_val, field_val in zip(options.pk_values, values): + for pk_val, field_val in zip(options.pk_values, values): # noqa: B905 - Python 3.8 has no zip(strict=) condition = ComparisonPredicate(self.dialect, "=", pk_col, Literal(self.dialect, pk_val)) result_expr = Literal(self.dialect, field_val) cases.append((condition, result_expr)) @@ -370,62 +392,34 @@ async def delete(self, options) -> QueryResult: return result def _handle_error(self, error: Exception) -> None: - from rhosocial.activerecord.backend.errors import ( - ConnectionError, IntegrityError, QueryError, DatabaseError, - ) - category = self._classify_error(error) - if category == 'connection': - raise ConnectionError(str(error)) from error - elif category == 'integrity': - raise IntegrityError(str(error)) from error - elif category == 'query': - raise QueryError(str(error)) from error - else: - raise DatabaseError(str(error)) from error + """Translate a driver exception into a taxonomy error and raise it. + + Delegates to :func:`~.backend._translate_bigquery_error` so the sync + and async backends classify identically; the original driver error + stays chained as the cause. Framework errors are re-raised unchanged, + preserving their phase-specific type. + """ + if isinstance(error, DatabaseError): + raise error + raise _translate_bigquery_error(error) from error + + async def _handle_auto_commit(self) -> None: + """No-op: a BigQuery query job commits its own statement. + + There is no connection-level commit or autocommit hook, so the + auto-commit phase has no failure mode and never raises + ``TransactionError``. Explicit transactions are driven with + ``BEGIN TRANSACTION`` / ``COMMIT TRANSACTION`` statements by + :class:`~..async_transaction.AsyncBigQueryTransactionManager`. + """ def _classify_error(self, error: Exception) -> str: - """Classify a google-cloud / google-api-core exception into an - error category used by :meth:`_handle_error`. + """Return the legacy category name for *error*. - Returns one of: 'connection', 'integrity', 'query', 'generic'. + Kept for callers that used this method before the taxonomy split; + :func:`~.backend._translate_bigquery_error` is authoritative. """ - try: - from google.api_core import exceptions as gexc - except ImportError: # google-cloud-bigquery not installed - return 'generic' - - if isinstance(error, gexc.InternalServerError): - # BigQuery (and the emulator) often reports query analysis / - # execution failures as jobInternalError 500s; only treat a bare - # internal error as a connection problem. - msg = str(error).lower() - integrity_hints = ('constraint failed', 'duplicate', 'not null', - 'primary key', 'foreign key', 'unique') - query_hints = ('failed to analyze', 'failed to parse', 'failed to exec', - 'not found', 'syntax error', 'unrecognized', 'invalid', - 'type mismatch', 'no such table', 'already exists') - if any(h in msg for h in integrity_hints): - return 'integrity' - if any(h in msg for h in query_hints): - return 'query' - return 'connection' - if isinstance(error, ( - gexc.ServiceUnavailable, gexc.DeadlineExceeded, gexc.GatewayTimeout, - gexc.Aborted, gexc.Unauthenticated, - gexc.ResourceExhausted, gexc.TooManyRequests, - )): - return 'connection' - if isinstance(error, (gexc.AlreadyExists, gexc.FailedPrecondition, - gexc.PermissionDenied)): - return 'integrity' - if isinstance(error, (gexc.BadRequest, gexc.InvalidArgument, - gexc.NotFound)): - msg = str(error).lower() - if any(k in msg for k in ('duplicate', 'constraint', 'primary key', - 'foreign key', 'not null')): - return 'integrity' - return 'query' - return 'generic' + return _classify_bigquery_error(error) def get_server_version(self) -> Tuple[int, ...]: return self._version diff --git a/src/rhosocial/activerecord/backend/impl/bigquery/backend/backend.py b/src/rhosocial/activerecord/backend/impl/bigquery/backend/backend.py index 4381d2f..19993f8 100644 --- a/src/rhosocial/activerecord/backend/impl/bigquery/backend/backend.py +++ b/src/rhosocial/activerecord/backend/impl/bigquery/backend/backend.py @@ -1,4 +1,5 @@ """BigQuery synchronous backend implementation.""" +import builtins import logging from typing import Any, Dict, List, Optional, Tuple @@ -7,7 +8,8 @@ log_execution_error, log_execution_start, log_execution_summary, ) from rhosocial.activerecord.backend.errors import ( - ConnectionError, DatabaseError, IntegrityError, QueryError, + ConnectionError, DatabaseError, IntegrityError, OperationalError, QueryError, + attach_diagnostics, ) from rhosocial.activerecord.backend.options import ExecutionOptions, LogContext from rhosocial.activerecord.backend.result import QueryResult @@ -62,6 +64,242 @@ def log_result_summary( ) +# -- Driver error taxonomy ---------------------------------------------------- +# google-cloud-bigquery 3.x reports service failures as +# ``google.api_core.exceptions`` instances carrying an HTTP status in +# ``code``. Recognition follows the taxonomy rule of status/class first and +# message text second: the class decides the category, and a message only +# refines a 4xx/5xx reply that the class alone cannot place. +# +# BigQuery has no deadlock or lock-wait concept. Concurrent DML is serialized +# by the service or rejected with a 409 ``Aborted`` reply, which is mapped to +# a transient ``OperationalError`` -- this backend never raises +# ``DeadlockError`` or ``LockError``. + +_INTEGRITY_MESSAGE_HINTS = ( + "already exists", + "constraint", + "duplicate", + "foreign key", + "not null", + "primary key", + "unique", +) +_QUERY_MESSAGE_HINTS = ( + "failed to analyze", + "failed to exec", + "failed to parse", + "invalid table", + "no such table", + "not found", + "syntax error", + "type mismatch", + "unrecognized column", + "unrecognized name", +) + + +def _driver_status(error: Exception) -> Optional[int]: + """Read the HTTP status a google.api_core exception carries. + + ``google.api_core.exceptions`` set ``code`` to an ``http.HTTPStatus``; + ``isinstance(code, int)`` is true because ``HTTPStatus`` is an ``int`` + subclass. Non-driver exceptions answer ``None``. + """ + code = getattr(error, "code", None) + if isinstance(code, int) and not isinstance(code, bool): + return int(code) + return None + + +def _is_transport_error(error: Exception) -> bool: + """Whether *error* is a lost connection or a socket-level timeout.""" + if isinstance(error, (builtins.ConnectionError, builtins.TimeoutError)): + return True + try: + from requests import exceptions as requests_exceptions + except ImportError: # google-cloud-bigquery not installed + requests_exceptions = None + if requests_exceptions is not None and isinstance( + error, (requests_exceptions.ConnectionError, requests_exceptions.Timeout) + ): + return True + try: + from google.auth.exceptions import TransportError + except ImportError: + return False + return isinstance(error, TransportError) + + +def _message_category(message: str) -> Optional[str]: + """Classify a driver message as ``"integrity"``, ``"query"`` or ``None``.""" + lower = message.lower() + if any(hint in lower for hint in _INTEGRITY_MESSAGE_HINTS): + return "integrity" + if any(hint in lower for hint in _QUERY_MESSAGE_HINTS): + return "query" + return None + + +def _translate_bigquery_error(error: Exception) -> DatabaseError: + """Translate a BigQuery driver exception into a taxonomy error. + + The returned instance is annotated with ``operation="execute"``, the + driver HTTP status as ``driver_code``, ``backend="bigquery"``, and an + explicit ``transient`` flag for infrastructure failures. Callers raise it + with ``from error`` so the original driver traceback stays chained. + + Framework :class:`DatabaseError` subclasses are returned unchanged: they + already carry the phase-specific type (``ConnectionError``, + ``TypeConversionError``, ``TransactionError`` ...) and re-translating them + would flatten that type into a generic ``DatabaseError``. + + Mapping (HTTP status / api_core class -> taxonomy error): + + * 401 ``Unauthorized`` / ``Unauthenticated``, 403 ``Forbidden`` / + ``PermissionDenied`` -> ``DatabaseError`` (no retry). + * 404 ``NotFound`` -> ``QueryError`` (unknown table, dataset or column). + * 409 ``AlreadyExists`` / ``Conflict`` -> ``IntegrityError``; 409 + ``Aborted`` -> transient ``OperationalError`` (no lock semantics). + * 429 ``TooManyRequests`` / ``ResourceExhausted``, 503 + ``ServiceUnavailable``, 504 ``GatewayTimeout`` / ``DeadlineExceeded``, + 500 ``InternalServerError`` -> transient ``OperationalError``. + * 400 ``BadRequest`` / ``InvalidArgument`` / ``FailedPrecondition`` -> + ``QueryError``, or ``IntegrityError`` when the message names a + constraint or an already-existing object. + * Connection/socket failures -> ``ConnectionError`` (transient). + * Anything else -> ``DatabaseError``. + """ + if isinstance(error, DatabaseError): + # Framework errors keep their type and diagnostics. Translating them + # again would turn ConnectionError / TypeConversionError / + # TransactionError into a generic DatabaseError. + return error + + status = _driver_status(error) + message = str(error) + operation = "execute" + backend = "bigquery" + + try: + from google.api_core import exceptions as gexc + except ImportError: # google-cloud-bigquery not installed + gexc = None + + def translated(error_class: type, transient: Optional[bool] = None) -> DatabaseError: + return attach_diagnostics( + error_class(message), + operation=operation, + driver_code=status, + backend=backend, + transient=transient, + ) + + if _is_transport_error(error): + return translated(ConnectionError, transient=True) + + if status in (401, 403): + # Authentication and permission refusals do not improve on retry. + return translated(DatabaseError, transient=False) + + if status == 404: + return translated(QueryError) + + if status == 409: + if gexc is not None and isinstance(error, gexc.Aborted): + # BigQuery has no deadlock/lock-wait concept: an aborted job is + # a transient concurrency outcome, not a lock error. + return translated(OperationalError, transient=True) + return translated(IntegrityError) + + if status in (429, 503, 504): + return translated(OperationalError, transient=True) + + if status is not None and status >= 500: + # BigQuery (and the emulator) reports query-analysis failures as + # jobInternalError 500s; refine those by message before declaring + # the reply transient. + category = _message_category(message) + if category == "integrity": + return translated(IntegrityError) + if category == "query": + return translated(QueryError) + return translated(OperationalError, transient=True) + + if status is not None and 400 <= status < 500: + if _message_category(message) == "integrity": + return translated(IntegrityError) + return translated(QueryError) + + if gexc is not None and isinstance(error, gexc.GoogleAPIError): + # Driver errors without an HTTP status (emulator REST replies, + # wrapped job errors) still get the message fallback. + category = _message_category(message) + if category == "integrity": + return translated(IntegrityError) + if category == "query": + return translated(QueryError) + return translated(DatabaseError) + + +def _translate_job_errors(errors: Any) -> DatabaseError: + """Build the taxonomy error for a job that reported ``errors``. + + The google client normally surfaces job failures through + ``job.exception()``, but some implementations (notably + goccy/bigquery-emulator) only populate ``job.errors``. The list carries + the driver's own verdict in its message, so it is fed through the same + message refinement the translator applies to status-less + ``GoogleAPIError`` replies: a query-analysis failure becomes a + ``QueryError`` instead of a bare ``DatabaseError``. + """ + message = str(errors) + try: + from google.api_core.exceptions import GoogleAPIError + except ImportError: # driver import already succeeded in practice + return attach_diagnostics( + DatabaseError(message), operation="execute", backend="bigquery" + ) + return _translate_bigquery_error(GoogleAPIError(message)) + + +def _translate_connect_error(error: Exception) -> ConnectionError: + """Translate a client-construction failure into a ``ConnectionError``. + + Connection acquisition has one contract type: authentication or permission + refusals are ``ConnectionError`` with ``transient=False``, while lost + sockets and timeouts are ``ConnectionError`` with ``transient=True``. The + execute translator supplies the category; this wrapper only fixes the type + and the phase label. + """ + translated = _translate_bigquery_error(error) + transient = bool(getattr(translated, "transient", False)) + return attach_diagnostics( + ConnectionError(f"Failed to connect to BigQuery: {error}"), + operation="connect", + driver_code=_driver_status(error), + backend="bigquery", + transient=transient, + ) + + +def _classify_bigquery_error(error: Exception) -> str: + """Return the legacy category string for a driver exception. + + The taxonomy translation in :func:`_translate_bigquery_error` is + authoritative; this keeps the historical ``_classify_error`` contract + (``connection`` / ``integrity`` / ``query`` / ``generic``) available. + """ + translated = _translate_bigquery_error(error) + if isinstance(translated, IntegrityError): + return "integrity" + if isinstance(translated, QueryError): + return "query" + if isinstance(translated, (ConnectionError, OperationalError)): + return "connection" + return "generic" + + class BigQueryBackend( IntrospectorBackendMixin, BigQueryQualifiedNameMixin, StorageBackend ): @@ -113,7 +351,6 @@ def connect(self) -> None: client_options=client_options, ) elif getattr(config, 'credentials_json', None): - import json self._client = bigquery.Client.from_service_account_info( config.credentials_json, client_options=client_options, @@ -127,7 +364,9 @@ def connect(self) -> None: self._connected = True self.log(logging.DEBUG, "Connected to BigQuery") except Exception as e: - raise ConnectionError(f"Failed to connect to BigQuery: {e}") from e + # The connection phase always surfaces ``ConnectionError``; the + # translator only decides whether the failure is transient. + raise _translate_connect_error(e) from e def disconnect(self) -> None: if hasattr(self, '_client') and self._client: @@ -147,7 +386,6 @@ def ping(self, reconnect: bool = True) -> bool: return True return False try: - from google.cloud import bigquery dataset_ref = self._client.dataset(getattr(self.config, 'dataset', None) or 'default') list(self._client.list_dataset_refs(dataset_ref)) return True @@ -162,59 +400,34 @@ def ping(self, reconnect: bool = True) -> bool: return False def _handle_error(self, error: Exception) -> None: - category = self._classify_error(error) - if category == 'connection': - raise ConnectionError(str(error)) from error - elif category == 'integrity': - raise IntegrityError(str(error)) from error - elif category == 'query': - raise QueryError(str(error)) from error - else: - raise DatabaseError(str(error)) from error + """Translate a driver exception into a taxonomy error and raise it. + + The status/class-based mapping lives in + :func:`_translate_bigquery_error`; the original driver exception is + kept as the cause of the raised error. Framework errors are re-raised + unchanged, preserving their phase-specific type. + """ + if isinstance(error, DatabaseError): + raise error + raise _translate_bigquery_error(error) from error + + def _handle_auto_commit(self) -> None: + """No-op: a BigQuery query job commits its own statement. + + There is no connection-level commit or autocommit hook, so the + auto-commit phase has no failure mode and never raises + ``TransactionError``. Explicit transactions are driven with + ``BEGIN TRANSACTION`` / ``COMMIT TRANSACTION`` statements by + :class:`~..transaction.BigQueryTransactionManager`. + """ def _classify_error(self, error: Exception) -> str: - """Classify a google-cloud / google-api-core exception into an - error category used by :meth:`_handle_error`. + """Return the legacy category name for *error*. - Returns one of: 'connection', 'integrity', 'query', 'generic'. + Kept for callers that used this method before the taxonomy split; + :func:`_translate_bigquery_error` is authoritative. """ - try: - from google.api_core import exceptions as gexc - except ImportError: # google-cloud-bigquery not installed - return 'generic' - - if isinstance(error, gexc.InternalServerError): - # BigQuery (and the emulator) often reports query analysis / - # execution failures as jobInternalError 500s; only treat a bare - # internal error as a connection problem. - msg = str(error).lower() - integrity_hints = ('constraint failed', 'duplicate', 'not null', - 'primary key', 'foreign key', 'unique') - query_hints = ('failed to analyze', 'failed to parse', 'failed to exec', - 'not found', 'syntax error', 'unrecognized', 'invalid', - 'type mismatch', 'no such table', 'already exists') - if any(h in msg for h in integrity_hints): - return 'integrity' - if any(h in msg for h in query_hints): - return 'query' - return 'connection' - if isinstance(error, ( - gexc.ServiceUnavailable, gexc.DeadlineExceeded, gexc.GatewayTimeout, - gexc.Aborted, gexc.Unauthenticated, - gexc.ResourceExhausted, gexc.TooManyRequests, - )): - return 'connection' - if isinstance(error, (gexc.AlreadyExists, gexc.FailedPrecondition, - gexc.PermissionDenied)): - return 'integrity' - if isinstance(error, (gexc.BadRequest, gexc.InvalidArgument, - gexc.NotFound)): - msg = str(error).lower() - if any(k in msg for k in ('duplicate', 'constraint', 'primary key', - 'foreign key', 'not null')): - return 'integrity' - return 'query' - return 'generic' + return _classify_bigquery_error(error) def get_server_version(self) -> Tuple[int, ...]: return self._version @@ -366,6 +579,18 @@ def execute( :mod:`rhosocial.activerecord.backend.base.execution_logging`; the backend-specific part is only the row count, which BigQuery reports on the job rather than on a cursor. + + Transaction semantics: every BigQuery query job commits its own + statement, so there is no connection-level commit/autocommit hook and + the auto-commit phase cannot raise ``TransactionError``. Explicit + transactions issue ``BEGIN TRANSACTION`` / ``COMMIT TRANSACTION`` + through this method (see + :class:`~..transaction.BigQueryTransactionManager`). + + Raises: + DatabaseError: the phase-specific taxonomy error: driver + failures are mapped by :func:`_translate_bigquery_error` and + framework errors pass through unchanged. """ if options is not None: column_adapters = column_adapters or getattr(options, 'column_adapters', None) @@ -404,7 +629,9 @@ def execute( exc = job.exception() if exc is not None: raise exc - raise DatabaseError(str(job.errors)) + # No driver exception to chain (emulator-style job errors): + # classify the message like a status-less driver reply. + raise _translate_job_errors(job.errors) if is_select: rows_iter = job.result() # The model layer expects dictionary rows keyed by column name @@ -430,6 +657,9 @@ def execute( except Exception as e: log_execution_error(self, e, options) self._handle_error(e) + # _handle_error() is required to raise. The fallback keeps a + # backend that forgot from turning a failure into a None result. + raise DatabaseError(str(e)) from e @property def transaction_manager(self): @@ -602,15 +832,25 @@ def _execute_via_rest(self, sql: str, params: Optional[Tuple], is_select: bool, @staticmethod def _raise_http_error(status: int, message: str): """Map an emulator HTTP error response onto the google exception types - that :meth:`_classify_error` understands.""" + that :func:`_translate_bigquery_error` understands.""" from google.api_core import exceptions as gexc if status == 400: raise gexc.BadRequest(message) + if status == 401: + raise gexc.Unauthorized(message) + if status == 403: + raise gexc.Forbidden(message) if status == 404: raise gexc.NotFound(message) if status == 409: raise gexc.AlreadyExists(message) + if status == 429: + raise gexc.TooManyRequests(message) + if status == 503: + raise gexc.ServiceUnavailable(message) + if status == 504: + raise gexc.GatewayTimeout(message) if status >= 500: raise gexc.InternalServerError(message) raise gexc.GoogleAPIError(f"HTTP {status}: {message}") diff --git a/tests/rhosocial/activerecord_bigquery_test/feature/backend/errors/test_error_taxonomy_contract.py b/tests/rhosocial/activerecord_bigquery_test/feature/backend/errors/test_error_taxonomy_contract.py new file mode 100644 index 0000000..fccc9dd --- /dev/null +++ b/tests/rhosocial/activerecord_bigquery_test/feature/backend/errors/test_error_taxonomy_contract.py @@ -0,0 +1,9 @@ +# tests/rhosocial/activerecord_bigquery_test/feature/backend/errors/test_error_taxonomy_contract.py +"""Run the testsuite's semantic error taxonomy contract for BigQuery. + +The contract only imports the core taxonomy; it needs no database and no +BigQuery driver. Importing it here gives this backend's suite the same +hierarchy/retry/diagnostics assertions every backend must satisfy. +""" + +from rhosocial.activerecord.testsuite.feature.backend.errors.test_error_taxonomy import * # noqa: F401,F403 diff --git a/tests/rhosocial/activerecord_bigquery_test/feature/backend/test_driver_error_mapping.py b/tests/rhosocial/activerecord_bigquery_test/feature/backend/test_driver_error_mapping.py new file mode 100644 index 0000000..eb98acf --- /dev/null +++ b/tests/rhosocial/activerecord_bigquery_test/feature/backend/test_driver_error_mapping.py @@ -0,0 +1,251 @@ +# tests/rhosocial/activerecord_bigquery_test/feature/backend/test_driver_error_mapping.py +"""BigQuery driver exceptions map onto the semantic error taxonomy. + +The tests synthesize ``google.api_core.exceptions`` instances and call the +backend's error translator directly, so no BigQuery service or emulator is +needed. Both the synchronous and the asynchronous backend are exercised +because they share the translation in ``backend.py``. +""" + +import builtins + +import pytest + +gexc = pytest.importorskip("google.api_core.exceptions") + +from rhosocial.activerecord.backend.errors import ( # noqa: E402 + ConnectionError, + DataError, + DatabaseError, + DeadlockError, + InfrastructureError, + IntegrityError, + LockError, + OperationalError, + QueryError, +) +from rhosocial.activerecord.backend.impl.bigquery.backend import ( # noqa: E402 + AsyncBigQueryBackend, + BigQueryBackend, +) + + +@pytest.fixture(params=[BigQueryBackend, AsyncBigQueryBackend], ids=["sync", "async"]) +def backend(request): + """An unconnected backend of either flavour.""" + return request.param(project="test", dataset="test_dataset") + + +def _translated(backend, error): + with pytest.raises(DatabaseError) as info: + backend._handle_error(error) + return info.value + + +def _assert_diagnostics(error, status): + assert error.operation == "execute" + assert error.backend == "bigquery" + if status is not None: + assert error.driver_code == status + assert type(error.driver_code) is int + + +def test_bad_request_already_exists_maps_to_integrity(backend): + translated = _translated( + backend, gexc.BadRequest("Already Exists: Table test.test_dataset.orders") + ) + assert isinstance(translated, IntegrityError) + assert isinstance(translated, DataError) + assert translated.retryable is False + _assert_diagnostics(translated, 400) + + +def test_conflict_already_exists_maps_to_integrity(backend): + translated = _translated( + backend, gexc.AlreadyExists("Table test.test_dataset.orders already exists") + ) + assert isinstance(translated, IntegrityError) + assert isinstance(translated, DataError) + _assert_diagnostics(translated, 409) + + +def test_bad_request_duplicate_key_maps_to_integrity(backend): + translated = _translated( + backend, gexc.BadRequest("Duplicate primary key for row 3") + ) + assert isinstance(translated, IntegrityError) + _assert_diagnostics(translated, 400) + + +def test_not_found_maps_to_query_error(backend): + translated = _translated( + backend, gexc.NotFound("Not found: Table test.test_dataset.orders") + ) + assert isinstance(translated, QueryError) + assert translated.retryable is False + _assert_diagnostics(translated, 404) + + +def test_forbidden_maps_to_database_error_without_retry(backend): + translated = _translated(backend, gexc.Forbidden("Access Denied: Project test")) + assert isinstance(translated, DatabaseError) + assert not isinstance(translated, InfrastructureError) + assert translated.transient is False + assert translated.retryable is False + _assert_diagnostics(translated, 403) + + +def test_unauthorized_maps_to_database_error_without_retry(backend): + translated = _translated(backend, gexc.Unauthorized("Invalid credentials")) + assert isinstance(translated, DatabaseError) + assert not isinstance(translated, InfrastructureError) + assert translated.retryable is False + _assert_diagnostics(translated, 401) + + +def test_rate_limit_is_a_transient_operational_error(backend): + translated = _translated(backend, gexc.TooManyRequests("Exceeded rate limits")) + assert isinstance(translated, OperationalError) + assert isinstance(translated, InfrastructureError) + assert translated.transient is True + assert translated.retryable is True + _assert_diagnostics(translated, 429) + + +def test_resource_exhausted_is_a_transient_operational_error(backend): + translated = _translated(backend, gexc.ResourceExhausted("Quota exceeded")) + assert isinstance(translated, OperationalError) + assert translated.transient is True + assert translated.retryable is True + _assert_diagnostics(translated, 429) + + +def test_service_unavailable_is_a_transient_operational_error(backend): + translated = _translated(backend, gexc.ServiceUnavailable("Backend unavailable")) + assert isinstance(translated, OperationalError) + assert isinstance(translated, InfrastructureError) + assert translated.transient is True + assert translated.retryable is True + _assert_diagnostics(translated, 503) + + +def test_deadline_exceeded_is_a_transient_operational_error(backend): + translated = _translated(backend, gexc.DeadlineExceeded("Query timed out")) + assert isinstance(translated, OperationalError) + assert translated.transient is True + assert translated.retryable is True + _assert_diagnostics(translated, 504) + + +def test_gateway_timeout_is_a_transient_operational_error(backend): + translated = _translated(backend, gexc.GatewayTimeout("Gateway timeout")) + assert isinstance(translated, OperationalError) + assert translated.transient is True + _assert_diagnostics(translated, 504) + + +def test_internal_server_error_is_a_transient_operational_error(backend): + translated = _translated(backend, gexc.InternalServerError("jobInternalError")) + assert isinstance(translated, OperationalError) + assert isinstance(translated, InfrastructureError) + assert translated.transient is True + assert translated.retryable is True + _assert_diagnostics(translated, 500) + + +def test_internal_server_error_analysis_failure_stays_a_query_error(backend): + translated = _translated( + backend, gexc.InternalServerError("Failed to analyze query: syntax error") + ) + assert isinstance(translated, QueryError) + assert translated.retryable is False + _assert_diagnostics(translated, 500) + + +def test_invalid_argument_maps_to_query_error(backend): + translated = _translated( + backend, gexc.InvalidArgument("Invalid table name: `bad name`") + ) + assert isinstance(translated, QueryError) + assert translated.retryable is False + _assert_diagnostics(translated, 400) + + +def test_failed_precondition_constraint_maps_to_integrity(backend): + translated = _translated( + backend, + gexc.FailedPrecondition("Cannot add row: unique constraint violated"), + ) + assert isinstance(translated, IntegrityError) + _assert_diagnostics(translated, 400) + + +def test_aborted_is_operational_and_never_a_lock_error(backend): + translated = _translated(backend, gexc.Aborted("Transaction aborted due to conflict")) + assert isinstance(translated, OperationalError) + assert translated.transient is True + assert not isinstance(translated, LockError) + assert not isinstance(translated, DeadlockError) + _assert_diagnostics(translated, 409) + + +def test_connection_failure_maps_to_connection_error(backend): + translated = _translated(backend, builtins.ConnectionError("Connection reset by peer")) + assert isinstance(translated, ConnectionError) + assert isinstance(translated, InfrastructureError) + assert translated.transient is True + assert translated.retryable is True + _assert_diagnostics(translated, None) + + +def test_generic_exception_maps_to_database_error(backend): + translated = _translated(backend, ValueError("boom")) + assert isinstance(translated, DatabaseError) + assert not isinstance( + translated, + (DataError, InfrastructureError, IntegrityError, QueryError), + ) + assert translated.retryable is False + _assert_diagnostics(translated, None) + + +def test_status_wins_over_message_text(backend): + # A 404 reply whose text mentions a duplicate stays a query error: the + # class/status decides, the message only refines an unplaced reply. + translated = _translated( + backend, gexc.NotFound("Duplicate table name is not the issue") + ) + assert isinstance(translated, QueryError) + _assert_diagnostics(translated, 404) + + +@pytest.mark.parametrize( + "status,expected_class", + [ + (401, gexc.Unauthorized), + (403, gexc.Forbidden), + (429, gexc.TooManyRequests), + (503, gexc.ServiceUnavailable), + (504, gexc.GatewayTimeout), + ], +) +def test_emulator_http_error_keeps_status_specific_class(status, expected_class): + # The emulator REST fast path has to raise a class whose ``code`` is the + # real HTTP status, otherwise 429/503/504 lose their transient mapping. + with pytest.raises(expected_class): + BigQueryBackend._raise_http_error(status, "emulator error") + + +@pytest.mark.parametrize( + "error,expected", + [ + (gexc.BadRequest("Duplicate key value violates unique constraint"), "integrity"), + (gexc.BadRequest("Syntax error: unexpected end of input"), "query"), + (builtins.ConnectionError("Connection reset by peer"), "connection"), + (RuntimeError("unexpected storage failure"), "generic"), + ], + ids=["integrity", "query", "connection", "generic"], +) +def test_classify_error_keeps_legacy_category_contract(backend, error, expected): + """``_classify_error`` still answers the historical category strings.""" + assert backend._classify_error(error) == expected diff --git a/tests/rhosocial/activerecord_bigquery_test/feature/backend/test_execute_error_contract.py b/tests/rhosocial/activerecord_bigquery_test/feature/backend/test_execute_error_contract.py new file mode 100644 index 0000000..e8858b0 --- /dev/null +++ b/tests/rhosocial/activerecord_bigquery_test/feature/backend/test_execute_error_contract.py @@ -0,0 +1,345 @@ +# tests/rhosocial/activerecord_bigquery_test/feature/backend/test_execute_error_contract.py +"""BigQuery execute phases follow the framework error contract. + +These tests never reach BigQuery: the client, the query job and the service +replies are in-process fakes, so no project, dataset or emulator is needed. +They pin the phases that the execute pipeline must distinguish: + +* framework errors entering ``_handle_error`` keep their type; +* connection acquisition always surfaces ``ConnectionError`` (transient only + for transport failures); +* driver 429/503/504 replies stay ``OperationalError(transient=True)`` and + authentication refusals stay non-retryable; +* ``job.errors`` without a driver exception is classified by its message; +* ``execute`` never returns ``None`` when the handler fails to raise. + +Both the synchronous and the asynchronous backend are covered because they +share the translation in ``backend/backend.py``. +""" + +import pytest + +gexc = pytest.importorskip("google.api_core.exceptions") + +from rhosocial.activerecord.backend.errors import ( # noqa: E402 + ConnectionError, + DatabaseError, + IntegrityError, + OperationalError, + QueryError, + TypeConversionError, +) +from rhosocial.activerecord.backend.impl.bigquery.backend import ( # noqa: E402 + AsyncBigQueryBackend, + BigQueryBackend, +) + + +class _FakeJob: + """Minimal query job covering the fields ``execute`` reads.""" + + def __init__(self, errors=None, exception=None): + self.errors = errors or [] + self._exception = exception + self.num_dml_affected_rows = 0 + self.started = None + self.ended = None + + def exception(self): + return self._exception + + +class _FakeClient: + """Client stand-in: either raises on ``query`` or returns a fake job.""" + + def __init__(self, job=None, query_error=None): + self._job = job + self._query_error = query_error + + def query(self, sql, job_config=None): + if self._query_error is not None: + raise self._query_error + return self._job + + +@pytest.fixture(params=[BigQueryBackend], ids=["sync"]) +def sync_backend(request): + return request.param(project="test", dataset="test_dataset") + + +@pytest.fixture(params=[AsyncBigQueryBackend], ids=["async"]) +def async_backend(request): + return request.param(project="test", dataset="test_dataset") + + +def _attach(backend, job=None, query_error=None): + """Install a fake client and mark the backend connected.""" + backend._client = _FakeClient(job=job, query_error=query_error) + backend._connected = True + return backend + + +def _raise_from(error): + def _factory(*args, **kwargs): + raise error + + return _factory + + +def _assert_diagnostics(error, transient): + assert error.operation == "execute" + assert error.backend == "bigquery" + assert error.transient is transient + + +class TestExecuteErrorContract: + """Synchronous phase assertions.""" + + def test_framework_connection_error_is_not_flattened(self, sync_backend): + error = ConnectionError("connection lost") + with pytest.raises(ConnectionError) as info: + sync_backend._handle_error(error) + assert info.value is error + + def test_framework_type_conversion_error_is_not_flattened(self, sync_backend): + error = TypeConversionError("cannot adapt value") + with pytest.raises(TypeConversionError) as info: + sync_backend._handle_error(error) + assert info.value is error + + @pytest.mark.parametrize( + "status,driver_error", + [ + (429, gexc.TooManyRequests), + (503, gexc.ServiceUnavailable), + (504, gexc.GatewayTimeout), + ], + ) + def test_transient_service_reply_stays_operational_error( + self, sync_backend, status, driver_error + ): + backend = _attach( + sync_backend, query_error=driver_error("service unavailable") + ) + with pytest.raises(OperationalError) as info: + backend.execute("INSERT INTO test_dataset.orders VALUES (1)") + assert info.value.transient is True + assert info.value.retryable is True + assert info.value.driver_code == status + + def test_forbidden_is_non_retryable_database_error(self, sync_backend): + backend = _attach( + sync_backend, query_error=gexc.Forbidden("Access Denied") + ) + with pytest.raises(DatabaseError) as info: + backend.execute("INSERT INTO test_dataset.orders VALUES (1)") + assert not isinstance(info.value, OperationalError) + assert info.value.transient is False + assert info.value.retryable is False + assert info.value.driver_code == 403 + + def test_job_errors_query_failure_raises_query_error(self, sync_backend): + job = _FakeJob( + errors=[{ + "reason": "invalidQuery", + "message": "Failed to analyze query: syntax error", + }] + ) + backend = _attach(sync_backend, job=job) + with pytest.raises(QueryError): + backend.execute("INSERT INTO test_dataset.orders VALUES (1)") + + def test_job_errors_integrity_failure_raises_integrity_error(self, sync_backend): + job = _FakeJob( + errors=[{"message": "Duplicate primary key for row 1"}] + ) + backend = _attach(sync_backend, job=job) + with pytest.raises(IntegrityError): + backend.execute("INSERT INTO test_dataset.orders VALUES (1)") + + def test_job_errors_without_hint_keeps_diagnostics(self, sync_backend): + job = _FakeJob(errors=[{"message": "Something went wrong"}]) + backend = _attach(sync_backend, job=job) + with pytest.raises(DatabaseError) as info: + backend.execute("INSERT INTO test_dataset.orders VALUES (1)") + assert not isinstance(info.value, (QueryError, IntegrityError)) + _assert_diagnostics(info.value, transient=None) + + def test_job_errors_stays_typed_through_handle_error(self, sync_backend): + # A framework error raised by the job path must not be re-translated. + error = TypeConversionError("job result decode failed") + with pytest.raises(TypeConversionError) as info: + sync_backend._handle_error(error) + assert info.value is error + + def test_execute_raises_when_handler_returns_none(self, sync_backend, monkeypatch): + backend = _attach(sync_backend, query_error=RuntimeError("boom")) + monkeypatch.setattr(backend, "_handle_error", lambda error: None) + with pytest.raises(DatabaseError): + backend.execute("INSERT INTO test_dataset.orders VALUES (1)") + + def test_auto_commit_hook_has_no_transaction_phase(self, sync_backend): + # BigQuery commits each job; the base hook stays a no-op and cannot + # raise TransactionError. + sync_backend._handle_auto_commit_if_needed() + + def test_connect_auth_failure_is_connection_error(self, monkeypatch): + bigquery = pytest.importorskip("google.cloud.bigquery") + monkeypatch.setattr( + bigquery, "Client", _raise_from(gexc.Forbidden("Access Denied")) + ) + backend = BigQueryBackend(project="test") + with pytest.raises(ConnectionError) as info: + backend.connect() + assert not isinstance(info.value, OperationalError) + assert info.value.operation == "connect" + assert info.value.transient is False + assert info.value.retryable is False + + def test_connect_transport_failure_is_transient_connection_error(self, monkeypatch): + bigquery = pytest.importorskip("google.cloud.bigquery") + from google.auth.exceptions import TransportError + + monkeypatch.setattr( + bigquery, "Client", _raise_from(TransportError("network down")) + ) + backend = BigQueryBackend(project="test") + with pytest.raises(ConnectionError) as info: + backend.connect() + assert info.value.operation == "connect" + assert info.value.transient is True + assert info.value.retryable is True + + +class TestAsyncExecuteErrorContract: + """Asynchronous phase assertions (same names as the sync class).""" + + @pytest.mark.asyncio + async def test_framework_connection_error_is_not_flattened(self, async_backend): + error = ConnectionError("connection lost") + with pytest.raises(ConnectionError) as info: + async_backend._handle_error(error) + assert info.value is error + + @pytest.mark.asyncio + async def test_framework_type_conversion_error_is_not_flattened(self, async_backend): + error = TypeConversionError("cannot adapt value") + with pytest.raises(TypeConversionError) as info: + async_backend._handle_error(error) + assert info.value is error + + @pytest.mark.asyncio + @pytest.mark.parametrize( + "status,driver_error", + [ + (429, gexc.TooManyRequests), + (503, gexc.ServiceUnavailable), + (504, gexc.GatewayTimeout), + ], + ) + async def test_transient_service_reply_stays_operational_error( + self, async_backend, status, driver_error + ): + backend = _attach( + async_backend, query_error=driver_error("service unavailable") + ) + with pytest.raises(OperationalError) as info: + await backend.execute("INSERT INTO test_dataset.orders VALUES (1)") + assert info.value.transient is True + assert info.value.retryable is True + assert info.value.driver_code == status + + @pytest.mark.asyncio + async def test_forbidden_is_non_retryable_database_error(self, async_backend): + backend = _attach( + async_backend, query_error=gexc.Forbidden("Access Denied") + ) + with pytest.raises(DatabaseError) as info: + await backend.execute("INSERT INTO test_dataset.orders VALUES (1)") + assert not isinstance(info.value, OperationalError) + assert info.value.transient is False + assert info.value.retryable is False + assert info.value.driver_code == 403 + + @pytest.mark.asyncio + async def test_job_errors_query_failure_raises_query_error(self, async_backend): + job = _FakeJob( + errors=[{ + "reason": "invalidQuery", + "message": "Failed to analyze query: syntax error", + }] + ) + backend = _attach(async_backend, job=job) + with pytest.raises(QueryError): + await backend.execute("INSERT INTO test_dataset.orders VALUES (1)") + + @pytest.mark.asyncio + async def test_job_errors_integrity_failure_raises_integrity_error( + self, async_backend + ): + job = _FakeJob( + errors=[{"message": "Duplicate primary key for row 1"}] + ) + backend = _attach(async_backend, job=job) + with pytest.raises(IntegrityError): + await backend.execute("INSERT INTO test_dataset.orders VALUES (1)") + + @pytest.mark.asyncio + async def test_job_errors_without_hint_keeps_diagnostics(self, async_backend): + job = _FakeJob(errors=[{"message": "Something went wrong"}]) + backend = _attach(async_backend, job=job) + with pytest.raises(DatabaseError) as info: + await backend.execute("INSERT INTO test_dataset.orders VALUES (1)") + assert not isinstance(info.value, (QueryError, IntegrityError)) + _assert_diagnostics(info.value, transient=None) + + @pytest.mark.asyncio + async def test_job_errors_stays_typed_through_handle_error(self, async_backend): + error = TypeConversionError("job result decode failed") + with pytest.raises(TypeConversionError) as info: + async_backend._handle_error(error) + assert info.value is error + + @pytest.mark.asyncio + async def test_execute_raises_when_handler_returns_none( + self, async_backend, monkeypatch + ): + backend = _attach(async_backend, query_error=RuntimeError("boom")) + monkeypatch.setattr(backend, "_handle_error", lambda error: None) + with pytest.raises(DatabaseError): + await backend.execute("INSERT INTO test_dataset.orders VALUES (1)") + + @pytest.mark.asyncio + async def test_auto_commit_hook_has_no_transaction_phase(self, async_backend): + await async_backend._handle_auto_commit_if_needed() + + @pytest.mark.asyncio + async def test_connect_auth_failure_is_connection_error(self, monkeypatch): + bigquery = pytest.importorskip("google.cloud.bigquery") + monkeypatch.setattr( + bigquery, "Client", _raise_from(gexc.Forbidden("Access Denied")) + ) + backend = AsyncBigQueryBackend(project="test") + with pytest.raises(ConnectionError) as info: + await backend.connect() + assert not isinstance(info.value, OperationalError) + assert info.value.operation == "connect" + assert info.value.transient is False + assert info.value.retryable is False + + @pytest.mark.asyncio + async def test_connect_transport_failure_is_transient_connection_error( + self, monkeypatch + ): + bigquery = pytest.importorskip("google.cloud.bigquery") + from google.auth.exceptions import TransportError + + monkeypatch.setattr( + bigquery, "Client", _raise_from(TransportError("network down")) + ) + backend = AsyncBigQueryBackend(project="test") + with pytest.raises(ConnectionError) as info: + await backend.connect() + assert info.value.operation == "connect" + assert info.value.transient is True + assert info.value.retryable is True