source-pendo: validate API key

Previously, users could successfully create tasks with an invalid Pendo
API key. Pendo returns a `403` code when we send a requests with an
invalid API key, and we now key off that to prevent tasks with invalid
API keys from being created.

source-pendo/source_pendo/__init__.py

@@ -14,7 +14,7 @@
 )
 from estuary_cdk.http import HTTPMixin
 
-from .resources import all_resources
+from .resources import all_resources, validate_api_key
 from .models import (
     ConnectorState,
     EndpointConfig,
@@ -48,6 +48,7 @@ async def validate(
         log: Logger,
         validate: request.Validate[EndpointConfig, ResourceConfig],
     ) -> response.Validated:
+        await validate_api_key(log, self, validate.config)
         resources = await all_resources(log, self, validate.config)
         resolved = common.resolve_bindings(validate.bindings, resources)
         return common.validated(resolved)

source-pendo/source_pendo/resources.py

@@ -2,9 +2,9 @@
 import functools
 from logging import Logger
 
-from estuary_cdk.flow import CaptureBinding
+from estuary_cdk.flow import CaptureBinding, ValidationError
 from estuary_cdk.capture import common, Task
-from estuary_cdk.http import HTTPMixin, TokenSource
+from estuary_cdk.http import HTTPMixin, TokenSource, HTTPError
 
 
 from .models import (
@@ -23,9 +23,29 @@
     fetch_events,
     fetch_aggregated_events,
     fetch_metadata,
+    API,
 )
 
 
+AUTHORIZATION_HEADER = "x-pendo-integration-key"
+
+
+async def validate_api_key(
+        log: Logger, http: HTTPMixin, config: EndpointConfig
+):
+    http.token_source = TokenSource(oauth_spec=None, credentials=config.credentials, authorization_header=AUTHORIZATION_HEADER)
+    url = f"{API}/metadata/schema/AccountMetadata"
+
+    try:
+        await http.request(log, url)
+    except HTTPError as err:
+        if err.code == 403:
+            msg = "Invalid API key. Please confirm the provided API key is correct."
+            raise ValidationError([msg])
+        else:
+            raise err
+
+
 def resources(
         log: Logger, http: HTTPMixin, config: EndpointConfig
 ) -> list[common.Resource]:
@@ -205,7 +225,7 @@ def open(
 async def all_resources(
     log: Logger, http: HTTPMixin, config: EndpointConfig
 ) -> list[common.Resource]:
-    http.token_source = TokenSource(oauth_spec=None, credentials=config.credentials, authorization_header="x-pendo-integration-key")
+    http.token_source = TokenSource(oauth_spec=None, credentials=config.credentials, authorization_header=AUTHORIZATION_HEADER)
 
     return [
         *resources(log, http, config),