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