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# Normalization Errors
226
227
228@dataclass
229class PyAirbyteNameNormalizationError(PyAirbyteError, ValueError):
230    """Error occurred while normalizing a table or column name."""
231
232    guidance = (
233        "Please consider renaming the source object if possible, or "
234        "raise an issue in GitHub if not."
235    )
236    help_url = NEW_ISSUE_URL
237
238    raw_name: str | None = None
239    normalization_result: str | None = None
240
241
242# PyAirbyte Cache Errors
243
244
245class PyAirbyteCacheError(PyAirbyteError):
246    """Error occurred while accessing the cache."""
247
248
249@dataclass
250class PyAirbyteCacheTableValidationError(PyAirbyteCacheError):
251    """Cache table validation failed."""
252
253    violation: str | None = None
254
255
256@dataclass
257class AirbyteConnectorConfigurationMissingError(PyAirbyteCacheError):
258    """Connector is missing configuration."""
259
260    connector_name: str | None = None
261
262
263# Subprocess Errors
264
265
266@dataclass
267class AirbyteSubprocessError(PyAirbyteError):
268    """Error when running subprocess."""
269
270    run_args: list[str] | None = None
271
272
273@dataclass
274class AirbyteSubprocessFailedError(AirbyteSubprocessError):
275    """Subprocess failed."""
276
277    exit_code: int | None = None
278
279
280# Connector Registry Errors
281
282
283class AirbyteConnectorRegistryError(PyAirbyteError):
284    """Error when accessing the connector registry."""
285
286
287@dataclass
288class AirbyteConnectorNotRegisteredError(AirbyteConnectorRegistryError):
289    """Connector not found in registry."""
290
291    connector_name: str | None = None
292    guidance = (
293        "Please double check the connector name. "
294        "Alternatively, you can provide an explicit connector install method to `get_source()`: "
295        "`pip_url`, `local_executable`, `docker_image`, or `source_manifest`."
296    )
297    help_url = DOCS_URL_BASE + "/airbyte/sources/util.html#get_source"
298
299
300@dataclass
301class AirbyteConnectorNotPyPiPublishedError(AirbyteConnectorRegistryError):
302    """Connector found, but not published to PyPI."""
303
304    connector_name: str | None = None
305    guidance = "This likely means that the connector is not ready for use with PyAirbyte."
306
307
308# Connector Errors
309
310
311@dataclass
312class AirbyteConnectorError(PyAirbyteError):
313    """Error when running the connector."""
314
315    connector_name: str | None = None
316
317    def __post_init__(self) -> None:
318        """Set the log file path for the connector."""
319        self.log_file = self._get_log_file()
320        if not self.guidance and self.log_file:
321            self.guidance = "Please review the log file for more information."
322
323    def _get_log_file(self) -> Path | None:
324        """Return the log file path for the connector."""
325        if self.connector_name:
326            logger = logging.getLogger(f"airbyte.{self.connector_name}")
327
328            log_paths: list[Path] = [
329                Path(handler.baseFilename).absolute()
330                for handler in logger.handlers
331                if isinstance(handler, logging.FileHandler)
332            ]
333
334            if log_paths:
335                return log_paths[0]
336
337        return None
338
339
340class AirbyteConnectorExecutableNotFoundError(AirbyteConnectorError):
341    """Connector executable not found."""
342
343
344class AirbyteConnectorInstallationError(AirbyteConnectorError):
345    """Error when installing the connector."""
346
347
348class AirbyteConnectorReadError(AirbyteConnectorError):
349    """Error when reading from the connector."""
350
351
352class AirbyteConnectorWriteError(AirbyteConnectorError):
353    """Error when writing to the connector."""
354
355
356class AirbyteConnectorSpecFailedError(AirbyteConnectorError):
357    """Error when getting spec from the connector."""
358
359
360class AirbyteConnectorDiscoverFailedError(AirbyteConnectorError):
361    """Error when running discovery on the connector."""
362
363
364class AirbyteNoDataFromConnectorError(AirbyteConnectorError):
365    """No data was provided from the connector."""
366
367
368class AirbyteConnectorMissingCatalogError(AirbyteConnectorError):
369    """Connector did not return a catalog."""
370
371
372class AirbyteConnectorMissingSpecError(AirbyteConnectorError):
373    """Connector did not return a spec."""
374
375
376class AirbyteConnectorValidationFailedError(AirbyteConnectorError):
377    """Connector config validation failed."""
378
379    guidance = (
380        "Please double-check your config and review the validation errors for more information."
381    )
382
383
384class AirbyteConnectorCheckFailedError(AirbyteConnectorError):
385    """Connector check failed."""
386
387    guidance = (
388        "Please double-check your config or review the connector's logs for more information."
389    )
390
391
392@dataclass
393class AirbyteConnectorFailedError(AirbyteConnectorError):
394    """Connector failed."""
395
396    exit_code: int | None = None
397
398
399@dataclass
400class AirbyteStreamNotFoundError(AirbyteConnectorError):
401    """Connector stream not found."""
402
403    stream_name: str | None = None
404    available_streams: list[str] | None = None
405
406
407@dataclass
408class AirbyteStateNotFoundError(AirbyteConnectorError, KeyError):
409    """State entry not found."""
410
411    stream_name: str | None = None
412    available_streams: list[str] | None = None
413
414
415@dataclass
416class PyAirbyteSecretNotFoundError(PyAirbyteError):
417    """Secret not found."""
418
419    guidance = "Please ensure that the secret is set."
420    help_url = (
421        "https://docs.airbyte.com/using-airbyte/airbyte-lib/getting-started#secrets-management"
422    )
423
424    secret_name: str | None = None
425    sources: list[str] | None = None
426
427
428# Airbyte API Errors
429
430
431@dataclass
432class AirbyteError(PyAirbyteError):
433    """An error occurred while communicating with the hosted Airbyte instance."""
434
435    response: AirbyteApiResponseDuckType | None = None
436    """The API response from the failed request."""
437
438    workspace: CloudWorkspace | None = None
439    """The workspace where the error occurred."""
440
441    @property
442    def workspace_url(self) -> str | None:
443        """The URL to the workspace where the error occurred."""
444        if self.workspace:
445            return self.workspace.workspace_url
446
447        return None
448
449
450@dataclass
451class AirbyteConnectionError(AirbyteError):
452    """An connection error occurred while communicating with the hosted Airbyte instance."""
453
454    connection_id: str | None = None
455    """The connection ID where the error occurred."""
456
457    job_id: int | None = None
458    """The job ID where the error occurred (if applicable)."""
459
460    job_status: str | None = None
461    """The latest status of the job where the error occurred (if applicable)."""
462
463    @property
464    def connection_url(self) -> str | None:
465        """The URL to the connection where the error occurred."""
466        if self.workspace_url and self.connection_id:
467            return f"{self.workspace_url}/connections/{self.connection_id}"
468
469        return None
470
471    @property
472    def job_history_url(self) -> str | None:
473        """The URL to the job history where the error occurred."""
474        if self.connection_url:
475            return f"{self.connection_url}/job-history"
476
477        return None
478
479    @property
480    def job_url(self) -> str | None:
481        """The URL to the job where the error occurred."""
482        if self.job_history_url and self.job_id:
483            return f"{self.job_history_url}#{self.job_id}::0"
484
485        return None
486
487
488@dataclass
489class AirbyteConnectionSyncError(AirbyteConnectionError):
490    """An error occurred while executing the remote Airbyte job."""
491
492
493@dataclass
494class AirbyteConnectionSyncTimeoutError(AirbyteConnectionSyncError):
495    """An timeout occurred while waiting for the remote Airbyte job to complete."""
496
497    timeout: int | None = None
498    """The timeout in seconds that was reached."""
499
500
501# Airbyte Resource Errors (General)
502
503
504@dataclass
505class AirbyteMissingResourceError(AirbyteError):
506    """Remote Airbyte resources does not exist."""
507
508    resource_type: str | None = None
509    resource_name_or_id: str | None = None
510
511
512@dataclass
513class AirbyteDuplicateResourcesError(AirbyteError):
514    """Process failed because resource name was not unique."""
515
516    resource_type: str | None = None
517    resource_name: str | None = None
518
519
520# Custom Warnings
521@dataclass
522class AirbyteMultipleResourcesError(AirbyteError):
523    """Could not locate the resource because multiple matching resources were found."""
524
525    resource_type: str | None = None
526    resource_name_or_id: str | None = None
527
528
529# Custom Warnings
530
531
532class AirbyteExperimentalFeatureWarning(FutureWarning):
533    """Warning whenever using experimental features in PyAirbyte."""
534
535
536# PyAirbyte Warnings
537
538
539class PyAirbyteWarning(Warning):
540    """General warnings from PyAirbyte."""
541
542
543class PyAirbyteDataLossWarning(PyAirbyteWarning):
544    """Warning for potential data loss.
545
546    Users can ignore this warning by running:
547    > warnings.filterwarnings("ignore", category="airbyte.exceptions.PyAirbyteDataLossWarning")
548    """