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