Initial implementation of HTTP connector

Author: edgarrmondragon

docs/classes/singer_sdk.connectors.BaseConnector.rst

@@ -0,0 +1,8 @@
+singer_sdk.connectors.BaseConnector
+===================================
+
+.. currentmodule:: singer_sdk.connectors
+
+.. autoclass:: BaseConnector
+    :members:
+    :special-members: __init__, __call__

docs/guides/custom-connector.md

@@ -0,0 +1,32 @@
+# Using a custom connector class
+
+The Singer SDK has a few built-in connector classes that are designed to work with a variety of sources:
+
+* [`SQLConnector`](../../classes/singer_sdk.SQLConnector) for SQL databases
+
+If you need to connect to a source that is not supported by one of these built-in connectors, you can create your own connector class. This guide will walk you through the process of creating a custom connector class.
+
+## Subclass `BaseConnector`
+
+The first step is to create a subclass of [`BaseConnector`](../../classes/singer_sdk.connectors.BaseConnector). This class is responsible for creating streams and handling the connection to the source.
+
+```python
+from singer_sdk.connectors import BaseConnector
+
+
+class MyConnector(BaseConnector):
+    pass
+```
+
+## Implement `get_connection`
+
+The [`get_connection`](http://127.0.0.1:5500/build/classes/singer_sdk.connectors.BaseConnector.html#singer_sdk.connectors.BaseConnector.get_connection) method is responsible for creating a connection to the source. It should return an object that implements the [context manager protocol](https://docs.python.org/3/reference/datamodel.html#with-statement-context-managers), e.g. it has `__enter__` and `__exit__` methods.
+
+```python
+from singer_sdk.connectors import BaseConnector
+
+
+class MyConnector(BaseConnector):
+    def get_connection(self):
+        return MyConnection()
+```

docs/guides/index.md

@@ -7,4 +7,5 @@ The following pages contain useful information for developers building on top of
 
 porting
 pagination-classes
+custom-connector
 ```

docs/reference.rst

@@ -130,3 +130,12 @@ Pagination
     pagination.BaseOffsetPaginator
     pagination.LegacyPaginatedStreamProtocol
     pagination.LegacyStreamPaginator
+
+Abstract Connector Classes
+--------------------------
+
+.. autosummary::
+    :toctree: classes
+    :template: class.rst
+
+    connectors.BaseConnector

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,<13"
+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
 
@@ -590,3 +591,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,8 @@
 
 from __future__ import annotations
 
+from ._http import HTTPConnector
+from .base import BaseConnector
 from .sql import SQLConnector
 
-__all__ = ["SQLConnector"]
+__all__ = ["BaseConnector", "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.
+        """
+        ...