Refactor of authentication client

config/config.py

@@ -1,52 +1,87 @@
-"""Module for managing configuration settings for the Datacosmos SDK.
+"""Configuration module for the Datacosmos SDK.
 
-Supports loading from YAML files and environment variables.
+Handles configuration management using Pydantic and Pydantic Settings.
+It loads default values, allows overrides via YAML configuration files,
+and supports environment variable-based overrides.
 """
 
 import os
-from dataclasses import dataclass
+from typing import Literal
 
 import yaml
+from pydantic import model_validator
+from pydantic_settings import BaseSettings, SettingsConfigDict
 
+from config.models.m2m_authentication_config import M2MAuthenticationConfig
+from config.models.url import URL
 
-@dataclass
-class Config:
-    """Configuration for the Datacosmos SDK.
 
-    Contains authentication details such as client ID, secret, token
-    URL, and audience.
-    """
+class Config(BaseSettings):
+    """Centralized configuration for the Datacosmos SDK."""
 
-    client_id: str
-    client_secret: str
-    token_url: str
-    audience: str
+    model_config = SettingsConfigDict(
+        env_nested_delimiter="__",
+        nested_model_default_partial_update=True,
+    )
 
-    @staticmethod
-    def from_yaml(file_path: str = "config/config.yaml") -> "Config":
-        """Load configuration from a YAML file.
+    # General configurations
+    environment: Literal["local", "test", "prod"] = "test"
+    log_format: Literal["json", "text"] = "text"
+    log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR"] = "INFO"
 
-        Defaults to 'config/config.yaml' unless otherwise specified.
+    # Authentication configuration
+    authentication: M2MAuthenticationConfig = M2MAuthenticationConfig(
+        type="m2m",
+        client_id="zCeZWJamwnb8ZIQEK35rhx0hSAjsZI4D",
+        token_url="https://login.open-cosmos.com/oauth/token",
+        audience="https://test.beeapp.open-cosmos.com",
+        client_secret="tAeaSgLds7g535ofGq79Zm2DSbWMCOsuRyY5lbyObJe9eAeSN_fxoy-5kaXnVSYa",
+    )
+
+    # STAC API configuration
+    stac: URL = URL(
+        protocol="https",
+        host="test.app.open-cosmos.com",
+        port=443,
+        path="/api/data/v0/stac",
+    )
+
+    @classmethod
+    def from_yaml(cls, file_path: str = "config/config.yaml") -> "Config":
+        """Load configuration from a YAML file and override defaults.
+
+        Args:
+            file_path (str): The path to the YAML configuration file.
+
+        Returns:
+            Config: An instance of the Config class with loaded settings.
         """
-        with open(file_path, "r") as f:
-            data = yaml.safe_load(f)
-        auth = data.get("auth", {})
-        return Config(
-            client_id=auth["client-id"],
-            client_secret=auth["client-secret"],
-            token_url=auth["token-url"],
-            audience=auth["audience"],
-        )
-
-    @staticmethod
-    def from_env() -> "Config":
-        """Load configuration from environment variables.
-
-        Raises an exception if any required variable is missing.
+        config_data = {}
+        if os.path.exists(file_path):
+            with open(file_path, "r") as f:
+                yaml_data = yaml.safe_load(f) or {}
+                # Remove empty values from YAML to avoid overwriting with `None`
+                config_data = {
+                    k: v for k, v in yaml_data.items() if v not in [None, ""]
+                }
+        return cls(**config_data)
+
+    @model_validator(mode="before")
+    @classmethod
+    def merge_with_env(cls, values):
+        """Override settings with environment variables if set.
+
+        This method checks if any environment variables corresponding to the
+        config fields are set and updates their values accordingly.
+
+        Args:
+            values (dict): The configuration values before validation.
+
+        Returns:
+            dict: The updated configuration values with environment variable overrides.
         """
-        return Config(
-            client_id=os.getenv("OC_AUTH_CLIENT_ID"),
-            client_secret=os.getenv("OC_AUTH_CLIENT_SECRET"),
-            token_url=os.getenv("OC_AUTH_TOKEN_URL"),
-            audience=os.getenv("OC_AUTH_AUDIENCE"),
-        )
+        for field in cls.model_fields:
+            env_value = os.getenv(f"OC_{field.upper()}")
+            if env_value:
+                values[field] = env_value
+        return values

config/models/m2m_authentication_config.py

@@ -0,0 +1,24 @@
+"""Module for configuring machine-to-machine (M2M) authentication.
+
+Used when running scripts in the cluster that require automated authentication
+without user interaction.
+"""
+
+from typing import Literal
+
+from pydantic import BaseModel
+
+
+class M2MAuthenticationConfig(BaseModel):
+    """Configuration for machine-to-machine authentication.
+
+    This is used when running scripts in the cluster that require authentication
+    with client credentials.
+    """
+
+    type: Literal["m2m"]
+    client_id: str
+    token_url: str
+    audience: str
+    # Some infrastructure deployments do not require a client secret.
+    client_secret: str = ""

config/models/url.py

@@ -0,0 +1,34 @@
+"""Module defining a structured URL configuration model.
+
+Ensures that URLs contain required components such as protocol, host,
+port, and path.
+"""
+
+from common.domain.url import URL as DomainURL
+from pydantic import BaseModel
+
+
+class URL(BaseModel):
+    """Generic configuration model for a URL.
+
+    This class provides attributes to store URL components and a method
+    to convert them into a `DomainURL` instance.
+    """
+
+    protocol: str
+    host: str
+    port: int
+    path: str
+
+    def as_domain_url(self) -> DomainURL:
+        """Convert the URL instance to a `DomainURL` object.
+
+        Returns:
+            DomainURL: A domain-specific URL object.
+        """
+        return DomainURL(
+            protocol=self.protocol,
+            host=self.host,
+            port=self.port,
+            base=self.path,
+        )

datacosmos/client.py

@@ -5,7 +5,6 @@
 """
 
 import logging
-import os
 from datetime import datetime, timedelta, timezone
 from typing import Any, Optional
 
@@ -27,51 +26,35 @@ class DatacosmosClient:
     def __init__(
         self,
         config: Optional[Config] = None,
-        config_file: str = "config/config.yaml",
     ):
         """Initialize the DatacosmosClient.
 
-        If no configuration is provided, it will load from the specified
-        YAML file or fall back to environment variables.
+        Load configuration from the specified YAML file, environment variables,
+        or fallback to the default values provided in the `Config` class.
         """
         self.logger = logging.getLogger(__name__)
         self.logger.setLevel(logging.INFO)
 
-        self.config = config or self._load_config(config_file)
+        self.config = config or Config.from_yaml()
         self.token = None
         self.token_expiry = None
         self._http_client = self._authenticate_and_initialize_client()
 
-    def _load_config(self, config_file: str) -> Config:
-        """Load configuration from the YAML file.
-
-        Fall back to environment variables if the file is missing.
-        """
-        try:
-            if os.path.exists(config_file):
-                self.logger.info(f"Loading configuration from {config_file}")
-                return Config.from_yaml(config_file)
-            self.logger.info(
-                "Loading configuration from environment variables"
-            )
-            return Config.from_env()
-        except Exception as e:
-            self.logger.error(f"Failed to load configuration: {e}")
-            raise
-
     def _authenticate_and_initialize_client(self) -> requests.Session:
         """Authenticate and initialize the HTTP client with a valid token."""
         try:
             self.logger.info("Authenticating with the token endpoint")
-            client = BackendApplicationClient(client_id=self.config.client_id)
+            client = BackendApplicationClient(
+                client_id=self.config.authentication.client_id
+            )
             oauth_session = OAuth2Session(client=client)
 
             # Fetch the token using client credentials
             token_response = oauth_session.fetch_token(
-                token_url=self.config.token_url,
-                client_id=self.config.client_id,
-                client_secret=self.config.client_secret,
-                audience=self.config.audience,
+                token_url=self.config.authentication.token_url,
+                client_id=self.config.authentication.client_id,
+                client_secret=self.config.authentication.client_secret,
+                audience=self.config.authentication.audience,
             )
 
             self.token = token_response["access_token"]
@@ -82,9 +65,7 @@ def _authenticate_and_initialize_client(self) -> requests.Session:
 
             # Initialize the HTTP session with the Authorization header
             http_client = requests.Session()
-            http_client.headers.update(
-                {"Authorization": f"Bearer {self.token}"}
-            )
+            http_client.headers.update({"Authorization": f"Bearer {self.token}"})
             return http_client
         except RequestException as e:
             self.logger.error(f"Request failed during authentication: {e}")
@@ -141,4 +122,4 @@ def put(self, url: str, *args: Any, **kwargs: Any) -> requests.Response:
 
     def delete(self, url: str, *args: Any, **kwargs: Any) -> requests.Response:
         """Send a DELETE request using the authenticated session."""
-        return self.request("DELETE", url, *args, **kwargs)
+        return self.request("DELETE", url, *args, **kwargs)

tests/unit/datacosmos/client/test_client_authentication.py

@@ -1,6 +1,7 @@
 from unittest.mock import patch
 
 from config.config import Config
+from config.models.m2m_authentication_config import M2MAuthenticationConfig
 from datacosmos.client import DatacosmosClient
 
 
@@ -21,10 +22,10 @@ def test_client_authentication(mock_auth_client, mock_fetch_token):
     def mock_authenticate_and_initialize_client(self):
         # Call the real fetch_token (simulated by the mock)
         token_response = mock_fetch_token(
-            token_url=self.config.token_url,
-            client_id=self.config.client_id,
-            client_secret=self.config.client_secret,
-            audience=self.config.audience,
+            token_url=self.config.authentication.token_url,
+            client_id=self.config.authentication.client_id,
+            client_secret=self.config.authentication.client_secret,
+            audience=self.config.authentication.audience,
         )
         self.token = token_response["access_token"]
         self.token_expiry = "mock-expiry"
@@ -34,10 +35,13 @@ def mock_authenticate_and_initialize_client(self):
 
     # Create a mock configuration
     config = Config(
-        client_id="test-client-id",
-        client_secret="test-client-secret",
-        token_url="https://mock.token.url/oauth/token",
-        audience="https://mock.audience",
+        authentication=M2MAuthenticationConfig(
+            type="m2m",
+            client_id="test-client-id",
+            client_secret="test-client-secret",
+            token_url="https://mock.token.url/oauth/token",
+            audience="https://mock.audience",
+        )
     )
 
     # Initialize the client
@@ -52,4 +56,4 @@ def mock_authenticate_and_initialize_client(self):
         client_secret="test-client-secret",
         audience="https://mock.audience",
     )
-    mock_auth_client.assert_called_once_with(client)
+    mock_auth_client.assert_called_once_with(client)

tests/unit/datacosmos/client/test_client_delete_request.py

@@ -1,12 +1,11 @@
 from unittest.mock import MagicMock, patch
 
 from config.config import Config
+from config.models.m2m_authentication_config import M2MAuthenticationConfig
 from datacosmos.client import DatacosmosClient
 
 
-@patch(
-    "datacosmos.client.DatacosmosClient._authenticate_and_initialize_client"
-)
+@patch("datacosmos.client.DatacosmosClient._authenticate_and_initialize_client")
 def test_delete_request(mock_auth_client):
     """Test that the client performs a DELETE request correctly."""
     # Mock the HTTP client
@@ -17,11 +16,15 @@ def test_delete_request(mock_auth_client):
     mock_auth_client.return_value = mock_http_client
 
     config = Config(
-        client_id="test-client-id",
-        client_secret="test-client-secret",
-        token_url="https://mock.token.url/oauth/token",
-        audience="https://mock.audience",
+        authentication=M2MAuthenticationConfig(
+            type="m2m",
+            client_id="test-client-id",
+            client_secret="test-client-secret",
+            token_url="https://mock.token.url/oauth/token",
+            audience="https://mock.audience",
+        )
     )
+
     client = DatacosmosClient(config=config)
     response = client.delete("https://mock.api/some-endpoint")
 
@@ -30,4 +33,4 @@ def test_delete_request(mock_auth_client):
     mock_http_client.request.assert_called_once_with(
         "DELETE", "https://mock.api/some-endpoint"
     )
-    mock_auth_client.call_count == 2
+    mock_auth_client.call_count == 2