Create base class for client factories

.pre-commit-config.yaml

@@ -46,7 +46,7 @@ repos:
     additional_dependencies: [flake8-bugbear, flake8-pyproject]
 
 - repo: https://github.com/pre-commit/mirrors-mypy
-  rev: 'v0.991'
+  rev: 'v1.0.1'
   hooks:
   - id: mypy
     additional_dependencies: [pydantic~=1.10]

setup.cfg

@@ -85,7 +85,7 @@ dev =
     black~=22.0
     flake8~=5.0
     isort~=5.0
-    mypy~=0.9
+    mypy~=1.0
     flake8-pyproject~=1.0
     sphinx-autodoc-typehints~=1.21
     interrogate~=1.5

src/orca/errors.py

@@ -9,8 +9,8 @@ class ClientRequestError(OrcaError):
     """Client request failed."""
 
 
-class ClientArgsError(OrcaError):
-    """Client arguments are missing or invalid."""
+class ClientAttrError(OrcaError):
+    """Client attributes are missing or invalid."""
 
 
 class OptionalAttrRequiredError(OrcaError):

src/orca/services/base/__init__.py

@@ -0,0 +1,6 @@
+"""Submodule for base classes containing shared functionality."""
+
+from orca.services.base.client_factory import BaseClientFactory
+from orca.services.base.config import BaseServiceConfig
+
+__all__ = ["BaseServiceConfig", "BaseClientFactory"]

src/orca/services/base/client_factory.py

@@ -0,0 +1,119 @@
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from functools import cached_property
+from typing import Any, Generic, TypeVar
+
+from pydantic.dataclasses import dataclass
+from typing_extensions import Self
+
+from orca.services.base.config import BaseServiceConfig
+
+ClientClass = TypeVar("ClientClass", bound=Any)
+
+ServiceConfig = TypeVar("ServiceConfig", bound=BaseServiceConfig)
+
+
+@dataclass(kw_only=False)
+class BaseClientFactory(ABC, Generic[ClientClass, ServiceConfig]):
+    """Base factory for constructing clients."""
+
+    # Using `__post_init_post_parse__()` to perform steps after validation
+    def __post_init_post_parse__(self) -> None:
+        """Resolve any attributes using the available methods."""
+        self.resolve()
+
+    @property
+    @abstractmethod
+    def config_class(self) -> ServiceConfig:
+        """Service configuration class."""
+
+    @property
+    @abstractmethod
+    def client_class(self) -> ClientClass:
+        """Service client class."""
+
+    @abstractmethod
+    def update_with_config(self, config: ServiceConfig):
+        """Update instance attributes based on client configuration.
+
+        Args:
+            config: Arguments relevant to this service.
+        """
+
+    @abstractmethod
+    def validate(self) -> None:
+        """Validate the currently available attributes.
+
+        Raises:
+            ClientAttrError: If one of the attributes is invalid.
+        """
+
+    @abstractmethod
+    def prepare_client_kwargs(self) -> dict[str, Any]:
+        """Prepare client keyword arguments.
+
+        Returns:
+            Dictionary of keyword arguments.
+        """
+
+    @staticmethod
+    @abstractmethod
+    def test_client(client: ClientClass) -> None:
+        """Test the client with an authenticated request.
+
+        Raises:
+            ClientRequestError: If an error occured while making a request.
+        """
+
+    @classmethod
+    def from_config(cls, config: ServiceConfig) -> Self:
+        """Construct client factory from configuration.
+
+        Args:
+            config: Arguments relevant to this service.
+
+        Returns:
+            An instantiated client factory.
+        """
+        factory = cls()
+        factory.update_with_config(config)
+        return factory
+
+    def resolve(self) -> None:
+        """Resolve credentials based on priority.
+
+        This method will update the attribute values (if applicable).
+        """
+        config = self.config_class.from_env()
+        self.update_with_config(config)
+
+    def create_client(self) -> ClientClass:
+        """Create authenticated client using the available attributes.
+
+        Returns:
+            An authenticated client for this service.
+        """
+        self.validate()
+        kwargs = self.prepare_client_kwargs()
+        client = self.client_class(**kwargs)
+        return client
+
+    @cached_property
+    def _client(self) -> ClientClass:
+        """An authenticated client."""
+        return self.create_client()
+
+    def get_client(self, test=False) -> ClientClass:
+        """Retrieve (and optionally, test) an authenticated client.
+
+        Args:
+            test: Whether to test the client before returning it.
+
+        Returns:
+            An authenticated client.
+        """
+        client = self._client
+        if test:
+            self.test_client(client)
+        return client

src/orca/services/base/config.py

@@ -0,0 +1,71 @@
+from __future__ import annotations
+
+import os
+from abc import ABC, abstractmethod
+from typing import TYPE_CHECKING, ClassVar
+
+from pydantic.dataclasses import dataclass
+from typing_extensions import Self
+
+if TYPE_CHECKING:
+    from airflow.models.connection import Connection
+
+
+@dataclass(kw_only=False)
+class BaseServiceConfig(ABC):
+    """Simple container class for service-related configuration."""
+
+    connection_env_var: ClassVar[str]
+
+    @classmethod
+    @abstractmethod
+    def from_connection(cls, connection: Connection) -> Self:
+        """Parse Airflow connection as a service configuration.
+
+        Args:
+            connection: An Airflow connection object.
+
+        Returns:
+            Configuration relevant to this service.
+        """
+
+    @classmethod
+    def from_env(cls) -> Self:
+        """Parse environment as a service configuration.
+
+        Args:
+            connection: An Airflow connection object.
+
+        Returns:
+            Configuration relevant to this service.
+        """
+        # Short-circuit method if absent because Connection is slow-ish
+        if cls.is_env_available():
+            connection = cls.get_connection_from_env()
+            config = cls.from_connection(connection)
+        else:
+            config = cls()
+        return config
+
+    @classmethod
+    def get_connection_from_env(cls) -> Connection:
+        """Generate Airflow connection from environment variable.
+
+        Returns:
+            An Airflow connection
+        """
+        # Following Airflow's lead on this non-standard practice
+        # because this import does introduce a bit of overhead
+        from airflow.models.connection import Connection
+
+        env_connection_uri = os.environ.get(cls.connection_env_var)
+        return Connection(uri=env_connection_uri)
+
+    @classmethod
+    def is_env_available(cls) -> bool:
+        """Check if the connection environment variable is available.
+
+        Returns:
+            Whether the connection environment variable is available.
+        """
+        return os.environ.get(cls.connection_env_var) is not None

src/orca/services/sevenbridges/__init__.py

@@ -1,7 +1,13 @@
 """Submodule for SevenBridges platforms (like Cavatica and CGC)."""
 
 from orca.services.sevenbridges.client_factory import SevenBridgesClientFactory
+from orca.services.sevenbridges.config import SevenBridgesConfig
 from orca.services.sevenbridges.hook import SevenBridgesHook
 from orca.services.sevenbridges.ops import SevenBridgesOps
 
-__all__ = ["SevenBridgesClientFactory", "SevenBridgesOps", "SevenBridgesHook"]
+__all__ = [
+    "SevenBridgesConfig",
+    "SevenBridgesClientFactory",
+    "SevenBridgesOps",
+    "SevenBridgesHook",
+]