Skip to content

Commit

Permalink
Source stripe: add documentation url
Browse files Browse the repository at this point in the history
  • Loading branch information
aldogonzalez8 committed Aug 5, 2024
1 parent 6fa27d2 commit 8e7d682
Showing 1 changed file with 3 additions and 33 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,6 @@

import requests
from airbyte_cdk.models import FailureType
from airbyte_cdk.sources.concurrent_source.concurrent_source_adapter import ConcurrentSourceAdapter
from airbyte_cdk.sources.streams.http import HttpStream
from airbyte_cdk.sources.streams.http.error_handlers import HttpStatusErrorHandler
from airbyte_cdk.sources.streams.http.error_handlers.response_models import ErrorResolution, ResponseAction
Expand All @@ -20,32 +19,8 @@
"oauth_not_supported": "Please use a different authentication method.",
}


def _visit_docs_message(logger: logging.Logger, source: ConcurrentSourceAdapter) -> str:
"""
Creates a message indicating where to look in the documentation for
more information on a given source by checking the spec of that source
(if provided) for a 'documentationUrl'.
:param logger: source logger
:param source: optional (source)
:return: A message telling the user where to go to learn more about the source.
"""

try:
connector_spec = source.spec(logger)
docs_url = connector_spec.documentationUrl
if docs_url:
return f"Please visit {docs_url} to learn more. "
else:
return "Please visit the connector's documentation to learn more. "

except FileNotFoundError: # If we are unit testing without implementing spec() method in source
if source:
docs_url = f"https://docs.airbyte.com/integrations/sources/{source.name}"
else:
docs_url = "https://docs.airbyte.com/integrations/sources/test"

return f"Please visit {docs_url} to learn more."
DOCS_URL = f"https://docs.airbyte.com/integrations/sources/stripe"
DOCUMENTATION_MESSAGE = f"Please visit {DOCS_URL} to learn more. "


class StripeErrorHandler(HttpStatusErrorHandler):
Expand All @@ -56,12 +31,7 @@ def interpret_response(self, response_or_exception: Optional[Union[requests.Resp
error_code = parsed_error.get("error", {}).get("code")
error_message = STRIPE_ERROR_CODES.get(error_code, parsed_error.get("error", {}).get("message"))
if error_message:
# todo can I just ifnore the doc message for now?
# doc_ref = self._visit_docs_message(self._logger, source)
reason = (
f"The endpoint {response_or_exception.url} returned {status_code}: {response_or_exception.reason}. {error_message}."
)
# reason = f"The endpoint {response_or_exception.url} returned {status_code}: {response_or_exception.reason}. {error_message}. {doc_ref} "
reason = f"The endpoint {response_or_exception.url} returned {status_code}: {response_or_exception.reason}. {error_message}. {DOCUMENTATION_MESSAGE} "
response_error_message = HttpStream.parse_response_error_message(response_or_exception)
if response_error_message:
reason += response_error_message
Expand Down

0 comments on commit 8e7d682

Please sign in to comment.