feat(config): refactor configuration classes to use pydantic-settings for better validation and management

nexus-mcp/lib/config.py

@@ -1,73 +1,177 @@
-"""Centralised config — loaded from environment / .env file."""
+"""Centralised config — loaded from environment / .env file using pydantic-settings."""
 
-import os
 from pathlib import Path
-from dotenv import load_dotenv
+from typing import Optional
-
+from pydantic import Field, field_validator
-# Load .env from the project root (nexus-mcp/)
+from pydantic_settings import BaseSettings, SettingsConfigDict
-load_dotenv(Path(__file__).parent.parent / ".env")
 
 
-class ADConfig:
+class ADConfig(BaseSettings):
-    server: str = os.getenv("AD_SERVER", "")
+    """Active Directory / LDAP configuration."""
-    port: int = int(os.getenv("AD_PORT", "389"))
+    model_config = SettingsConfigDict(
-    base_dn: str = os.getenv("AD_BASE_DN", "")
+        env_prefix="AD_",
-    user: str = os.getenv("AD_USER", "")
+        env_file=".env",
-    password: str = os.getenv("AD_PASSWORD", "")
+        env_file_encoding="utf-8",
-    use_ssl: bool = os.getenv("AD_USE_SSL", "false").lower() == "true"
+        extra="ignore"
+    )
+    
+    server: str = ""
+    port: int = 389
+    base_dn: str = ""
+    user: str = ""
+    password: str = ""
+    use_ssl: bool = False
 
 
-class EntraConfig:
+class EntraConfig(BaseSettings):
-    tenant_id: str = os.getenv("ENTRA_TENANT_ID", "")
+    """Microsoft Entra ID (Azure AD) configuration."""
-    client_id: str = os.getenv("ENTRA_CLIENT_ID", "")
+    model_config = SettingsConfigDict(
-    client_secret: str = os.getenv("ENTRA_CLIENT_SECRET", "")
+        env_prefix="ENTRA_",
+        env_file=".env",
+        env_file_encoding="utf-8",
+        extra="ignore"
+    )
+    
+    tenant_id: str = ""
+    client_id: str = ""
+    client_secret: str = ""
 
 
-class IntuneConfig:
+class IntuneConfig(BaseSettings):
-    tenant_id: str = os.getenv("INTUNE_TENANT_ID") or os.getenv("ENTRA_TENANT_ID", "")
+    """Microsoft Intune configuration (falls back to Entra credentials)."""
-    client_id: str = os.getenv("INTUNE_CLIENT_ID") or os.getenv("ENTRA_CLIENT_ID", "")
+    model_config = SettingsConfigDict(
-    client_secret: str = os.getenv("INTUNE_CLIENT_SECRET") or os.getenv("ENTRA_CLIENT_SECRET", "")
+        env_file=".env",
+        env_file_encoding="utf-8",
+        extra="ignore"
+    )
+    
+    intune_tenant_id: Optional[str] = Field(default=None, alias="INTUNE_TENANT_ID")
+    intune_client_id: Optional[str] = Field(default=None, alias="INTUNE_CLIENT_ID")
+    intune_client_secret: Optional[str] = Field(default=None, alias="INTUNE_CLIENT_SECRET")
+    
+    # Fallback to Entra credentials
+    entra_tenant_id: Optional[str] = Field(default=None, alias="ENTRA_TENANT_ID")
+    entra_client_id: Optional[str] = Field(default=None, alias="ENTRA_CLIENT_ID")
+    entra_client_secret: Optional[str] = Field(default=None, alias="ENTRA_CLIENT_SECRET")
+    
+    @property
+    def tenant_id(self) -> str:
+        return self.intune_tenant_id or self.entra_tenant_id or ""
+    
+    @property
+    def client_id(self) -> str:
+        return self.intune_client_id or self.entra_client_id or ""
+    
+    @property
+    def client_secret(self) -> str:
+        return self.intune_client_secret or self.entra_client_secret or ""
 
 
-class WorkdayConfig:
+class WorkdayConfig(BaseSettings):
-    base_url: str = os.getenv("WORKDAY_BASE_URL", "")
+    """Workday HCM API configuration."""
-    tenant: str = os.getenv("WORKDAY_TENANT", "")
+    model_config = SettingsConfigDict(
-    client_id: str = os.getenv("WORKDAY_CLIENT_ID", "")
+        env_prefix="WORKDAY_",
-    client_secret: str = os.getenv("WORKDAY_CLIENT_SECRET", "")
+        env_file=".env",
-    refresh_token: str = os.getenv("WORKDAY_REFRESH_TOKEN", "")
+        env_file_encoding="utf-8",
+        extra="ignore"
+    )
+    
+    base_url: str = ""
+    tenant: str = ""
+    client_id: str = ""
+    client_secret: str = ""
+    refresh_token: str = ""
 
 
-class HelixConfig:
+class HelixConfig(BaseSettings):
-    base_url: str = os.getenv("HELIX_BASE_URL", "")
+    """BMC Helix ITSM configuration."""
-    username: str = os.getenv("HELIX_USERNAME", "")
+    model_config = SettingsConfigDict(
-    password: str = os.getenv("HELIX_PASSWORD", "")
+        env_prefix="HELIX_",
+        env_file=".env",
+        env_file_encoding="utf-8",
+        extra="ignore"
+    )
+    
+    base_url: str = ""
+    username: str = ""
+    password: str = ""
 
 
-class LansweeperConfig:
+class LansweeperConfig(BaseSettings):
-    api_url: str = os.getenv("LANSWEEPER_API_URL", "https://api.lansweeper.com/api/v2/graphql")
+    """Lansweeper asset management API configuration."""
-    application_id: str = os.getenv("LANSWEEPER_APPLICATION_ID", "")
+    model_config = SettingsConfigDict(
-    application_secret: str = os.getenv("LANSWEEPER_APPLICATION_SECRET", "")
+        env_prefix="LANSWEEPER_",
-    site_id: str = os.getenv("LANSWEEPER_SITE_ID", "")
+        env_file=".env",
+        env_file_encoding="utf-8",
+        extra="ignore"
+    )
+    
+    api_url: str = "https://api.lansweeper.com/api/v2/graphql"
+    application_id: str = ""
+    application_secret: str = ""
+    site_id: str = ""
 
 
-class FedExConfig:
+class FedExConfig(BaseSettings):
-    api_url: str = os.getenv("FEDEX_API_URL", "https://apis.fedex.com")
+    """FedEx shipping API configuration."""
-    api_key: str = os.getenv("FEDEX_API_KEY", "")
+    model_config = SettingsConfigDict(
-    api_secret: str = os.getenv("FEDEX_API_SECRET", "")
+        env_prefix="FEDEX_",
-    account_number: str = os.getenv("FEDEX_ACCOUNT_NUMBER", "")
+        env_file=".env",
+        env_file_encoding="utf-8",
+        extra="ignore"
+    )
+    
+    api_url: str = "https://apis.fedex.com"
+    api_key: str = ""
+    api_secret: str = ""
+    account_number: str = ""
 
 
-class ReportConfig:
+class ReportConfig(BaseSettings):
-    output_dir: Path = Path(os.getenv("REPORT_OUTPUT_DIR", "./reports"))
+    """Report generation configuration."""
+    model_config = SettingsConfigDict(
+        env_prefix="REPORT_",
+        env_file=".env",
+        env_file_encoding="utf-8",
+        extra="ignore"
+    )
+    
+    output_dir: Path = Path("./reports")
+    
+    @field_validator("output_dir", mode="before")
+    @classmethod
+    def parse_path(cls, v):
+        if isinstance(v, str):
+            return Path(v)
+        return v
 
 
-class AuditConfig:
+class AuditConfig(BaseSettings):
     """SOC 2 audit log configuration.
 
     Controls:
       CC7.2  — System Monitoring: log_file is the append-only audit trail.
       CC6.1  — Logical Access: log_to_stderr enables SIEM/syslog forwarding.
     """
-    log_file: Path = Path(os.getenv("AUDIT_LOG_FILE", "./logs/nexus_audit.jsonl"))
+    model_config = SettingsConfigDict(
-    log_to_stderr: bool = os.getenv("AUDIT_LOG_STDERR", "true").lower() == "true"
+        env_prefix="AUDIT_",
-    enabled: bool = os.getenv("AUDIT_LOGGING_ENABLED", "true").lower() == "true"
+        env_file=".env",
+        env_file_encoding="utf-8",
+        extra="ignore"
+    )
+    
+    log_file: Path = Field(default=Path("./logs/nexus_audit.jsonl"), alias="AUDIT_LOG_FILE")
+    log_to_stderr: bool = Field(default=True, alias="AUDIT_LOG_STDERR")
+    logging_enabled: bool = Field(default=True, alias="AUDIT_LOGGING_ENABLED")
+    
+    @field_validator("log_file", mode="before")
+    @classmethod
+    def parse_path(cls, v):
+        if isinstance(v, str):
+            return Path(v)
+        return v
+    
+    # Backwards compatibility alias
+    @property
+    def enabled(self) -> bool:
+        return self.logging_enabled

nexus-mcp/pyproject.toml

@@ -12,6 +12,7 @@ dependencies = [
     "httpx>=0.27.0",
     "python-dotenv>=1.0.0",
     "pydantic>=2.0.0",
+    "pydantic-settings>=2.0.0",
     "ldap3>=2.9.1",
     "msal>=1.28.0",
     "schedule>=1.2.0",