airbyte.exceptions

  1# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
  2
  3"""All exceptions used in the PyAirbyte.
  4
  5This design is modeled after structlog's exceptions, in that we bias towards auto-generated
  6property prints rather than sentence-like string concatenation.
  7
  8E.g. Instead of this:
  9
 10> `Subprocess failed with exit code '1'`
 11
 12We do this:
 13
 14> `Subprocess failed. (exit_code=1)`
 15
 16The benefit of this approach is that we can easily support structured logging, and we can
 17easily add new properties to exceptions without having to update all the places where they
 18are raised. We can also support any arbitrary number of properties in exceptions, without spending
 19time on building sentence-like string constructions with optional inputs.
 20
 21
 22In addition, the following principles are applied for exception class design:
 23
 24- All exceptions inherit from a common base class.
 25- All exceptions have a message attribute.
 26- The first line of the docstring is used as the default message.
 27- The default message can be overridden by explicitly setting the message attribute.
 28- Exceptions may optionally have a guidance attribute.
 29- Exceptions may optionally have a help_url attribute.
 30- Rendering is automatically handled by the base class.
 31- Any helpful context not defined by the exception class can be passed in the `context` dict arg.
 32- Within reason, avoid sending PII to the exception constructor.
 33- Exceptions are dataclasses, so they can be instantiated with keyword arguments.
 34- Use the 'from' syntax to chain exceptions when it is helpful to do so.
 35  E.g. `raise AirbyteConnectorNotFoundError(...) from FileNotFoundError(connector_path)`
 36- Any exception that adds a new property should also be decorated as `@dataclass`.
 37"""
 38
 39from __future__ import annotations
 40
 41import logging
 42from dataclasses import dataclass
 43from pathlib import Path
 44from textwrap import indent
 45from typing import TYPE_CHECKING, Any, Protocol
 46
 47from airbyte.constants import (
 48    AIRBYTE_PRINT_FULL_ERROR_LOGS,
 49    CLOUD_BEARER_TOKEN_ENV_VAR,
 50    CLOUD_CLIENT_ID_ENV_VAR,
 51    CLOUD_CLIENT_SECRET_ENV_VAR,
 52    CLOUD_WORKSPACE_ID_ENV_VAR,
 53    MCP_BEARER_TOKEN_HEADER,
 54    MCP_CLIENT_ID_HEADER,
 55    MCP_CLIENT_SECRET_HEADER,
 56    MCP_WORKSPACE_ID_HEADER,
 57    is_hosted_mcp_mode,
 58)
 59
 60
 61if TYPE_CHECKING:
 62    from airbyte._util.api_duck_types import AirbyteApiResponseDuckType
 63
 64
 65NEW_ISSUE_URL = "https://github.com/airbytehq/airbyte/issues/new/choose"
 66DOCS_URL_BASE = "https://airbytehq.github.io/PyAirbyte"
 67DOCS_URL = f"{DOCS_URL_BASE}/airbyte.html"
 68
 69VERTICAL_SEPARATOR = "\n" + "-" * 60
 70
 71
 72# Base error class
 73
 74
 75@dataclass
 76class PyAirbyteError(Exception):
 77    """Base class for exceptions in Airbyte."""
 78
 79    guidance: str | None = None
 80    help_url: str | None = None
 81    log_text: str | list[str] | None = None
 82    log_file: Path | None = None
 83    print_full_log: bool = AIRBYTE_PRINT_FULL_ERROR_LOGS
 84    context: dict[str, Any] | None = None
 85    message: str | None = None
 86    original_exception: Exception | None = None
 87
 88    def get_message(self) -> str:
 89        """Return the best description for the exception.
 90
 91        We resolve the following in order:
 92        1. The message sent to the exception constructor (if provided).
 93        2. The first line of the class's docstring.
 94        """
 95        if self.message:
 96            return self.message
 97
 98        return self.__doc__.split("\n")[0] if self.__doc__ else ""
 99
100    def __str__(self) -> str:
101        """Return a string representation of the exception."""
102        special_properties = [
103            "message",
104            "guidance",
105            "help_url",
106            "log_text",
107            "context",
108            "log_file",
109            "print_full_log",
110            "original_exception",
111        ]
112        display_properties = {
113            k: v
114            for k, v in self.__dict__.items()
115            if k not in special_properties and not k.startswith("_") and v is not None
116        }
117        display_properties.update(self.context or {})
118        context_str = "\n    ".join(
119            f"{str(k).replace('_', ' ').title()}: {v!r}" for k, v in display_properties.items()
120        )
121        exception_str = (
122            f"{self.get_message()} ({self.__class__.__name__})"
123            + VERTICAL_SEPARATOR
124            + f"\n{self.__class__.__name__}: {self.get_message()}"
125        )
126
127        if self.guidance:
128            exception_str += f"\n    {self.guidance}"
129
130        if self.help_url:
131            exception_str += f"\n    More info: {self.help_url}"
132
133        if context_str:
134            exception_str += "\n    " + context_str
135
136        if self.log_text:
137            if isinstance(self.log_text, list):
138                self.log_text = "\n".join(self.log_text)
139
140            exception_str += f"\n    Log output: \n    {indent(self.log_text, '    ')}"
141
142        if self.original_exception:
143            exception_str += VERTICAL_SEPARATOR + f"\nCaused by: {self.original_exception!s}"
144
145        if self.log_file:
146            if self.print_full_log:
147                if not self.log_file.is_file():
148                    exception_str += f"\n    No log file found at: {self.log_file.absolute()!s}"
149
150                else:
151                    try:
152                        full_log_file_text = self.log_file.read_text()
153                    except Exception as ex:
154                        full_log_file_text = (
155                            f"[ERROR] Log file could not be read from: {self.log_file.absolute()!s}"
156                            f"\nRead error: {ex!s}"
157                        )
158
159                    exception_str += (
160                        f"\n    Full log file text from {self.log_file.absolute()!s}:"
161                        + VERTICAL_SEPARATOR
162                        + full_log_file_text
163                        + VERTICAL_SEPARATOR
164                    )
165            else:
166                exception_str += f"\n    Log file: {self.log_file.absolute()!s}"
167        return exception_str
168
169    def __repr__(self) -> str:
170        """Return a string representation of the exception."""
171        class_name = self.__class__.__name__
172        properties_str = ", ".join(
173            f"{k}={v!r}" for k, v in self.__dict__.items() if not k.startswith("_")
174        )
175        return f"{class_name}({properties_str})"
176
177    def safe_logging_dict(self) -> dict[str, Any]:
178        """Return a dictionary of the exception's properties which is safe for logging.
179
180        We avoid any properties which could potentially contain PII.
181        """
182        result = {
183            # The class name is safe to log:
184            "class": self.__class__.__name__,
185            # We discourage interpolated strings in 'message' so that this should never contain PII:
186            "message": self.get_message(),
187        }
188        safe_attrs = ["connector_name", "stream_name", "violation", "exit_code"]
189        for attr in safe_attrs:
190            if hasattr(self, attr):
191                result[attr] = getattr(self, attr)
192
193        return result
194
195
196# PyAirbyte Internal Errors (these are probably bugs)
197
198
199@dataclass
200class PyAirbyteInternalError(PyAirbyteError):
201    """An internal error occurred in PyAirbyte."""
202
203    guidance = "Please consider reporting this error to the Airbyte team."
204    help_url = NEW_ISSUE_URL
205
206
207# PyAirbyte Input Errors (replaces ValueError for user input)
208
209
210@dataclass
211class PyAirbyteInputError(PyAirbyteError, ValueError):
212    """The input provided to PyAirbyte did not match expected validation rules.
213
214    This inherits from ValueError so that it can be used as a drop-in replacement for
215    ValueError in the PyAirbyte API.
216    """
217
218    guidance = "Please check the provided value and try again."
219    help_url = DOCS_URL
220    input_value: str | None = None
221
222
223@dataclass
224class PyAirbyteNoStreamsSelectedError(PyAirbyteInputError):
225    """No streams were selected for the source."""
226
227    guidance = (
228        "Please call `select_streams()` to select at least one stream from the list provided. "
229        "You can also call `select_all_streams()` to select all available streams for this source."
230    )
231    connector_name: str | None = None
232    available_streams: list[str] | None = None
233
234
235@dataclass
236class AirbyteNoCloudCredentialsError(PyAirbyteInputError):
237    """No Airbyte credentials found."""
238
239    guidance: str | None = None
240    _allow_bearer: bool = True
241    _env_vars: bool = True
242
243    def __post_init__(self) -> None:
244        """Set guidance for the current execution mode."""
245        if self.guidance is not None:
246            return
247        if is_hosted_mcp_mode():
248            if self._allow_bearer:
249                self.guidance = (
250                    f"Provide a bearer token via the `{MCP_BEARER_TOKEN_HEADER}` header, "
251                    f"or client credentials via the `{MCP_CLIENT_ID_HEADER}` and "
252                    f"`{MCP_CLIENT_SECRET_HEADER}` headers."
253                )
254            else:
255                self.guidance = (
256                    f"Provide client credentials via the `{MCP_CLIENT_ID_HEADER}` and "
257                    f"`{MCP_CLIENT_SECRET_HEADER}` headers."
258                )
259        elif self._allow_bearer and self._env_vars:
260            self.guidance = (
261                f"Provide `bearer_token`, or both `client_id` and `client_secret`, as "
262                f"arguments or via the `{CLOUD_BEARER_TOKEN_ENV_VAR}`, "
263                f"`{CLOUD_CLIENT_ID_ENV_VAR}`, and `{CLOUD_CLIENT_SECRET_ENV_VAR}` "
264                "environment variables."
265            )
266        elif self._allow_bearer:
267            self.guidance = "Provide `bearer_token`, or both `client_id` and `client_secret`."
268        elif self._env_vars:
269            self.guidance = (
270                f"Provide both `client_id` and `client_secret`, as arguments or via the "
271                f"`{CLOUD_CLIENT_ID_ENV_VAR}` and `{CLOUD_CLIENT_SECRET_ENV_VAR}` "
272                "environment variables."
273            )
274        else:
275            self.guidance = "Provide both `client_id` and `client_secret`."
276
277
278@dataclass
279class AirbyteMissingWorkspaceContextError(PyAirbyteInputError):
280    """Workspace ID is required but not provided."""
281
282    guidance: str | None = None
283
284    def __post_init__(self) -> None:
285        """Set guidance for the current execution mode."""
286        if self.guidance is not None:
287            return
288        if is_hosted_mcp_mode():
289            self.guidance = (
290                f"Provide the workspace ID via the `{MCP_WORKSPACE_ID_HEADER}` header, "
291                "or pass the `workspace_id` parameter."
292            )
293        else:
294            self.guidance = (
295                f"Set the `{CLOUD_WORKSPACE_ID_ENV_VAR}` environment variable, or pass "
296                "the `workspace_id` parameter."
297            )
298
299
300# MCP Server Errors
301
302
303@dataclass
304class AirbyteMCPError(PyAirbyteError):
305    """An error occurred in the Airbyte MCP server."""
306
307
308@dataclass
309class AirbyteTrustedExecutionRequiredError(AirbyteMCPError):
310    """A trusted-execution-only capability was invoked while trusted execution is disabled.
311
312    Trusted execution grants the MCP server its trusted-machine capabilities: local
313    filesystem access, local connector installation/execution, and server-side secret
314    resolution. It defaults to *off* on every transport and is permanently unavailable
315    over the HTTP transport, so a backend helper that exposes one of those capabilities
316    hard-fails when the gate is disabled -- independently of whether the corresponding
317    tool was hidden from the tool listing.
318    """
319
320    guidance = (
321        "Set `AIRBYTE_MCP_TRUSTED_EXECUTION=1` on the MCP server process and restart it. "
322        "Trusted execution is only available on the stdio transport; it can never be "
323        "enabled for an HTTP/hosted deployment."
324    )
325    feature: str | None = None
326
327
328# Normalization Errors
329
330
331@dataclass
332class PyAirbyteNameNormalizationError(PyAirbyteError, ValueError):
333    """Error occurred while normalizing a table or column name."""
334
335    guidance = (
336        "Please consider renaming the source object if possible, or "
337        "raise an issue in GitHub if not."
338    )
339    help_url = NEW_ISSUE_URL
340
341    raw_name: str | None = None
342    normalization_result: str | None = None
343
344
345# PyAirbyte Cache Errors
346
347
348class PyAirbyteCacheError(PyAirbyteError):
349    """Error occurred while accessing the cache."""
350
351
352@dataclass
353class PyAirbyteCacheTableValidationError(PyAirbyteCacheError):
354    """Cache table validation failed."""
355
356    violation: str | None = None
357
358
359@dataclass
360class AirbyteConnectorConfigurationMissingError(PyAirbyteCacheError):
361    """Connector is missing configuration."""
362
363    connector_name: str | None = None
364
365
366# Subprocess Errors
367
368
369@dataclass
370class AirbyteSubprocessError(PyAirbyteError):
371    """Error when running subprocess."""
372
373    run_args: list[str] | None = None
374
375
376@dataclass
377class AirbyteSubprocessFailedError(AirbyteSubprocessError):
378    """Subprocess failed."""
379
380    exit_code: int | None = None
381
382
383# Connector Registry Errors
384
385
386class AirbyteConnectorRegistryError(PyAirbyteError):
387    """Error when accessing the connector registry."""
388
389
390@dataclass
391class AirbyteConnectorNotRegisteredError(AirbyteConnectorRegistryError):
392    """Connector not found in registry."""
393
394    connector_name: str | None = None
395    guidance = (
396        "Please double check the connector name. "
397        "Alternatively, you can provide an explicit connector install method to `get_source()`: "
398        "`pip_url`, `local_executable`, `docker_image`, or `source_manifest`."
399    )
400    help_url = DOCS_URL_BASE + "/airbyte/sources/util.html#get_source"
401
402
403@dataclass
404class AirbyteConnectorNotPyPiPublishedError(AirbyteConnectorRegistryError):
405    """Connector found, but not published to PyPI."""
406
407    connector_name: str | None = None
408    guidance = "This likely means that the connector is not ready for use with PyAirbyte."
409
410
411# Connector Errors
412
413
414@dataclass
415class AirbyteConnectorError(PyAirbyteError):
416    """Error when running the connector."""
417
418    connector_name: str | None = None
419
420    def __post_init__(self) -> None:
421        """Set the log file path for the connector."""
422        self.log_file = self._get_log_file()
423        if not self.guidance and self.log_file:
424            self.guidance = "Please review the log file for more information."
425
426    def _get_log_file(self) -> Path | None:
427        """Return the log file path for the connector."""
428        if self.connector_name:
429            logger = logging.getLogger(f"airbyte.{self.connector_name}")
430
431            log_paths: list[Path] = [
432                Path(handler.baseFilename).absolute()
433                for handler in logger.handlers
434                if isinstance(handler, logging.FileHandler)
435            ]
436
437            if log_paths:
438                return log_paths[0]
439
440        return None
441
442
443class AirbyteConnectorExecutableNotFoundError(AirbyteConnectorError):
444    """Connector executable not found."""
445
446
447class AirbyteConnectorInstallationError(AirbyteConnectorError):
448    """Error when installing the connector."""
449
450
451class AirbyteConnectorReadError(AirbyteConnectorError):
452    """Error when reading from the connector."""
453
454
455class AirbyteConnectorWriteError(AirbyteConnectorError):
456    """Error when writing to the connector."""
457
458
459class AirbyteConnectorSpecFailedError(AirbyteConnectorError):
460    """Error when getting spec from the connector."""
461
462
463class AirbyteConnectorDiscoverFailedError(AirbyteConnectorError):
464    """Error when running discovery on the connector."""
465
466
467class AirbyteNoDataFromConnectorError(AirbyteConnectorError):
468    """No data was provided from the connector."""
469
470
471class AirbyteConnectorMissingCatalogError(AirbyteConnectorError):
472    """Connector did not return a catalog."""
473
474
475class AirbyteConnectorMissingSpecError(AirbyteConnectorError):
476    """Connector did not return a spec."""
477
478
479class AirbyteConnectorValidationFailedError(AirbyteConnectorError):
480    """Connector config validation failed."""
481
482    guidance = (
483        "Please double-check your config and review the validation errors for more information."
484    )
485
486
487class AirbyteConnectorCheckFailedError(AirbyteConnectorError):
488    """Connector check failed."""
489
490    guidance = (
491        "Please double-check your config or review the connector's logs for more information."
492    )
493
494
495@dataclass
496class AirbyteConnectorFailedError(AirbyteConnectorError):
497    """Connector failed."""
498
499    exit_code: int | None = None
500
501
502@dataclass
503class AirbyteStreamNotFoundError(AirbyteConnectorError):
504    """Connector stream not found."""
505
506    stream_name: str | None = None
507    available_streams: list[str] | None = None
508
509
510@dataclass
511class AirbyteStateNotFoundError(AirbyteConnectorError, KeyError):
512    """State entry not found."""
513
514    stream_name: str | None = None
515    available_streams: list[str] | None = None
516
517
518@dataclass
519class PyAirbyteSecretNotFoundError(PyAirbyteError):
520    """Secret not found."""
521
522    guidance = "Please ensure that the secret is set."
523    help_url = (
524        "https://docs.airbyte.com/using-airbyte/airbyte-lib/getting-started#secrets-management"
525    )
526
527    secret_name: str | None = None
528    sources: list[str] | None = None
529
530
531# Airbyte API Errors
532
533
534class _WorkspaceWithUrl(Protocol):
535    """Structural type for a workspace that exposes a `workspace_url`.
536
537    Declared locally so `exceptions` does not need to import `airbyte.cloud`, which
538    would create an import cycle. Any object with a `workspace_url` attribute (e.g.
539    `CloudWorkspace`) satisfies this via structural (duck) typing.
540    """
541
542    @property
543    def workspace_url(self) -> str | None:
544        """The web URL of the workspace."""
545
546
547@dataclass
548class AirbyteError(PyAirbyteError):
549    """An error occurred while communicating with the hosted Airbyte instance."""
550
551    response: AirbyteApiResponseDuckType | None = None
552    """The API response from the failed request."""
553
554    workspace: _WorkspaceWithUrl | None = None
555    """The workspace where the error occurred."""
556
557    @property
558    def workspace_url(self) -> str | None:
559        """The URL to the workspace where the error occurred."""
560        if self.workspace:
561            return self.workspace.workspace_url
562
563        return None
564
565
566@dataclass
567class AirbyteConnectionError(AirbyteError):
568    """An connection error occurred while communicating with the hosted Airbyte instance."""
569
570    connection_id: str | None = None
571    """The connection ID where the error occurred."""
572
573    job_id: int | None = None
574    """The job ID where the error occurred (if applicable)."""
575
576    job_status: str | None = None
577    """The latest status of the job where the error occurred (if applicable)."""
578
579    @property
580    def connection_url(self) -> str | None:
581        """The web URL to the connection where the error occurred."""
582        if self.workspace_url and self.connection_id:
583            return f"{self.workspace_url}/connections/{self.connection_id}"
584
585        return None
586
587    @property
588    def job_history_url(self) -> str | None:
589        """The URL to the job history where the error occurred."""
590        if self.connection_url:
591            return f"{self.connection_url}/timeline"
592
593        return None
594
595    @property
596    def job_url(self) -> str | None:
597        """The URL to the job where the error occurred."""
598        if self.job_history_url and self.job_id:
599            return f"{self.job_history_url}#{self.job_id}::0"
600
601        return None
602
603
604@dataclass
605class AirbyteConnectionSyncError(AirbyteConnectionError):
606    """An error occurred while executing the remote Airbyte job."""
607
608
609@dataclass
610class AirbyteConnectionSyncActiveError(AirbyteConnectionError):
611    """State update rejected because a sync is currently running (HTTP 423)."""
612
613
614@dataclass
615class AirbyteWorkspaceMismatchError(AirbyteError):
616    """Resource does not belong to the expected workspace.
617
618    This error is raised when a resource (connection, source, or destination) is fetched
619    from the API and the workspace ID in the response does not match the expected workspace.
620    """
621
622    resource_type: str | None = None
623    """The type of resource (e.g., 'connection', 'source', 'destination')."""
624
625    resource_id: str | None = None
626    """The ID of the resource that was fetched."""
627
628    expected_workspace_id: str | None = None
629    """The workspace ID that was expected."""
630
631    actual_workspace_id: str | None = None
632    """The workspace ID returned by the API."""
633
634
635@dataclass
636class AirbyteWorkspaceNotEmptyError(AirbyteError):
637    """Workspace cannot be deleted because it contains connections."""
638
639    workspace_id: str | None = None
640    """The workspace ID that was expected to be empty."""
641
642    connection_ids: list[str] | None = None
643    """The connection IDs found in the workspace."""
644
645
646@dataclass
647class AirbyteConnectionSyncTimeoutError(AirbyteConnectionSyncError):
648    """An timeout occurred while waiting for the remote Airbyte job to complete."""
649
650    timeout: int | None = None
651    """The timeout in seconds that was reached."""
652
653
654# Airbyte Resource Errors (General)
655
656
657@dataclass
658class AirbyteMissingResourceError(AirbyteError):
659    """Remote Airbyte resources does not exist."""
660
661    resource_type: str | None = None
662    resource_name_or_id: str | None = None
663
664
665@dataclass
666class AirbyteDuplicateResourcesError(AirbyteError):
667    """Process failed because resource name was not unique."""
668
669    resource_type: str | None = None
670    resource_name: str | None = None
671
672
673# Custom Warnings
674@dataclass
675class AirbyteMultipleResourcesError(AirbyteError):
676    """Could not locate the resource because multiple matching resources were found."""
677
678    resource_type: str | None = None
679    resource_name_or_id: str | None = None
680
681
682# Custom Warnings
683
684
685class AirbyteExperimentalFeatureWarning(FutureWarning):
686    """Warning whenever using experimental features in PyAirbyte."""
687
688
689# PyAirbyte Warnings
690
691
692class PyAirbyteWarning(Warning):
693    """General warnings from PyAirbyte."""
694
695
696class PyAirbyteDataLossWarning(PyAirbyteWarning):
697    """Warning for potential data loss.
698
699    Users can ignore this warning by running:
700    > warnings.filterwarnings("ignore", category="airbyte.exceptions.PyAirbyteDataLossWarning")
701    """