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