Initial implementation of HTTP connector

Author: edgarrmondragon

noxfile.py

@@ -34,8 +34,9 @@
 test_dependencies = [
     "coverage[toml]",
     "pytest",
-    "pytest-snapshot",
     "pytest-durations",
+    "pytest-httpserver",
+    "pytest-snapshot",
     "freezegun",
     "pandas",
     "pyarrow",

poetry.lock

@@ -113,6 +113,7 @@ types-simplejson = "^3.18.0"
 types-PyYAML = "^6.0.12"
 coverage = {extras = ["toml"], version = "^7.2"}
 pyarrow = "^11.0.0"
+pytest-httpserver = "^1.0.6"
 pytest-snapshot = "^0.9.0"
 
 # Cookiecutter tests

pyproject.toml

@@ -13,6 +13,7 @@
 import requests
 from cryptography.hazmat.backends import default_backend
 from cryptography.hazmat.primitives import serialization
+from requests.auth import AuthBase
 
 from singer_sdk.helpers._util import utc_now
 
@@ -589,3 +590,52 @@ def oauth_request_payload(self) -> dict:
                 "RS256",
             ),
         }
+
+
+class NoopAuth(AuthBase):
+    """No-op authenticator."""
+
+    def __call__(self, r: requests.PreparedRequest) -> requests.PreparedRequest:
+        """Do nothing.
+
+        Args:
+            r: The prepared request.
+
+        Returns:
+            The unmodified prepared request.
+        """
+        return r
+
+
+class HeaderAuth(AuthBase):
+    """Header-based authenticator."""
+
+    def __init__(
+        self,
+        keyword: str,
+        value: str,
+        header: str = "Authorization",
+    ) -> None:
+        """Initialize the authenticator.
+
+        Args:
+            keyword: The keyword to use in the header, e.g. "Bearer".
+            value: The value to use in the header, e.g. "my-token".
+            header: The header to add the keyword and value to, defaults to
+                ``"Authorization"``.
+        """
+        self.keyword = keyword
+        self.value = value
+        self.header = header
+
+    def __call__(self, r: requests.PreparedRequest) -> requests.PreparedRequest:
+        """Add the header to the request.
+
+        Args:
+            r: The prepared request.
+
+        Returns:
+            The prepared request with the header added.
+        """
+        r.headers[self.header] = f"{self.keyword} {self.value}"
+        return r

singer_sdk/authenticators.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+from ._http import HTTPConnector
 from .sql import SQLConnector
 
-__all__ = ["SQLConnector"]
+__all__ = ["HTTPConnector", "SQLConnector"]

singer_sdk/connectors/__init__.py

@@ -0,0 +1,112 @@
+"""HTTP-based tap class for Singer SDK."""
+
+from __future__ import annotations
+
+import typing as t
+
+import requests
+
+from singer_sdk.authenticators import NoopAuth
+from singer_sdk.connectors.base import BaseConnector
+
+if t.TYPE_CHECKING:
+    import sys
+
+    from requests.adapters import BaseAdapter
+
+    if sys.version_info >= (3, 10):
+        from typing import TypeAlias  # noqa: ICN003
+    else:
+        from typing_extensions import TypeAlias
+
+_Auth: TypeAlias = t.Callable[[requests.PreparedRequest], requests.PreparedRequest]
+
+
+class HTTPConnector(BaseConnector[requests.Session]):
+    """Base class for all HTTP-based connectors."""
+
+    def __init__(self, config: t.Mapping[str, t.Any] | None) -> None:
+        """Initialize the HTTP connector.
+
+        Args:
+            config: Connector configuration parameters.
+        """
+        super().__init__(config)
+        self._session = self.get_session()
+        self.refresh_auth()
+
+    def get_connection(self, *, authenticate: bool = True) -> requests.Session:
+        """Return a new HTTP session object.
+
+        Adds adapters and optionally authenticates the session.
+
+        Args:
+            authenticate: Whether to authenticate the request.
+
+        Returns:
+            A new HTTP session object.
+        """
+        for prefix, adapter in self.adapters.items():
+            self._session.mount(prefix, adapter)
+
+        self._session.auth = self._auth if authenticate else None
+
+        return self._session
+
+    def get_session(self) -> requests.Session:
+        """Return a new HTTP session object.
+
+        Returns:
+            A new HTTP session object.
+        """
+        return requests.Session()
+
+    def get_authenticator(self) -> _Auth:
+        """Authenticate the HTTP session.
+
+        Returns:
+            An auth callable.
+        """
+        return NoopAuth()
+
+    def refresh_auth(self) -> None:
+        """Refresh the HTTP session authentication."""
+        self._auth = self.get_authenticator()
+
+    @property
+    def adapters(self) -> dict[str, BaseAdapter]:
+        """Return a mapping of URL prefixes to adapter objects.
+
+        Returns:
+            A mapping of URL prefixes to adapter objects.
+        """
+        return {}
+
+    @property
+    def default_request_kwargs(self) -> dict[str, t.Any]:
+        """Return default kwargs for HTTP requests.
+
+        Returns:
+            A mapping of default kwargs for HTTP requests.
+        """
+        return {}
+
+    def request(
+        self,
+        *args: t.Any,
+        authenticate: bool = True,
+        **kwargs: t.Any,
+    ) -> requests.Response:
+        """Make an HTTP request.
+
+        Args:
+            *args: Positional arguments to pass to the request method.
+            authenticate: Whether to authenticate the request.
+            **kwargs: Keyword arguments to pass to the request method.
+
+        Returns:
+            The HTTP response object.
+        """
+        with self._connect(authenticate=authenticate) as session:
+            kwargs = {**self.default_request_kwargs, **kwargs}
+            return session.request(*args, **kwargs)

singer_sdk/connectors/_http.py

@@ -0,0 +1,69 @@
+"""Base class for all connectors."""
+
+from __future__ import annotations
+
+import abc
+import typing as t
+from contextlib import contextmanager
+
+from singer_sdk.helpers._compat import Protocol
+
+_T = t.TypeVar("_T", covariant=True)
+
+
+class ContextManagerProtocol(Protocol[_T]):
+    """Protocol for context manager enter/exit."""
+
+    def __enter__(self) -> _T:  # noqa: D105
+        ...  # pragma: no cover
+
+    def __exit__(self, *args: t.Any) -> None:  # noqa: D105
+        ...  # pragma: no cover
+
+
+_C = t.TypeVar("_C", bound=ContextManagerProtocol)
+
+
+class BaseConnector(abc.ABC, t.Generic[_C]):
+    """Base class for all connectors."""
+
+    def __init__(self, config: t.Mapping[str, t.Any] | None) -> None:
+        """Initialize the connector.
+
+        Args:
+            config: Plugin configuration parameters.
+        """
+        self._config = config or {}
+
+    @property
+    def config(self) -> t.Mapping:
+        """Return the connector configuration.
+
+        Returns:
+            A mapping of configuration parameters.
+        """
+        return self._config
+
+    @contextmanager
+    def _connect(self, *args: t.Any, **kwargs: t.Any) -> t.Generator[_C, None, None]:
+        """Connect to the destination.
+
+        Args:
+            args: Positional arguments to pass to the connection method.
+            kwargs: Keyword arguments to pass to the connection method.
+
+        Yields:
+            A connection object.
+        """
+        with self.get_connection(*args, **kwargs) as connection:
+            yield connection
+
+    @abc.abstractmethod
+    def get_connection(self, *args: t.Any, **kwargs: t.Any) -> _C:
+        """Connect to the destination.
+
+        Args:
+            args: Positional arguments to pass to the connection method.
+            kwargs: Keyword arguments to pass to the connection method.
+        """
+        ...