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