feat: base classes for the configuration system (#5975)

components/src/dynamo/common/configuration/__init__.py

+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""
+ArgGroup-based configuration system for Dynamo.
+
+This module provides a modular, domain-driven configuration architecture where:
+- Each ArgGroup owns a specific domain of configuration parameters
+- Components declare which ArgGroups they need
+- Unrecognized arguments are captured for backend engines (passthrough)
+"""
+
+from .arg_group import ArgGroup
+from .config_base import ConfigBase
+from .utils import add_argument, add_negatable_bool_argument, env_or_default
+
+__all__ = [
+    # Base classes
+    "ArgGroup",
+    "ConfigBase",
+    # Utilities
+    "add_argument",
+    "env_or_default",
+    "add_negatable_bool_argument",
+]

components/src/dynamo/common/configuration/arg_group.py

+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Base ArgGroup interface."""
+from abc import ABC, abstractmethod
+
+
+class ArgGroup(ABC):
+    """
+    Base interface for configuration groups.
+
+    Each ArgGroup represents a domain of configuration parameters with clear ownership.
+    """
+
+    @abstractmethod
+    def add_arguments(self, parser) -> None:
+        """
+        Register CLI arguments owned by this group.
+
+        This method must be side-effect free beyond parser mutation.
+        It must not depend on runtime state or other groups.
+
+        Args:
+            parser: argparse.ArgumentParser or argument group
+        """
+        ...

components/src/dynamo/common/configuration/config_base.py

+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+import argparse
+
+
+class ConfigBase:
+    """Base configuration class that allows properties with and without defaults in arbitrary order."""
+
+    @classmethod
+    def from_cli_args(cls, args: argparse.Namespace):
+        obj = cls.__new__(cls)
+
+        # 1) Set everything provided by argparse
+        for k, v in vars(args).items():
+            setattr(obj, k, v)
+
+        # 2) Populate annotated defaults from the class (and base classes)
+        #    only if not already set by argparse.
+        for base in reversed(cls.__mro__):
+            anns = getattr(base, "__annotations__", {})
+            for name in anns:
+                if name.startswith("_"):
+                    continue
+
+                # IMPORTANT: only skip if it's already set on the INSTANCE
+                if name in obj.__dict__:
+                    continue
+
+                # If the class defines a default, materialize it onto the instance
+                if name in getattr(base, "__dict__", {}):
+                    setattr(obj, name, getattr(base, name))
+
+        return obj
+
+    def __repr__(self) -> str:
+        items = ", ".join(f"{k}={v!r}" for k, v in sorted(self.__dict__.items()))
+        return f"{self.__class__.__name__}({items})"

components/src/dynamo/common/configuration/utils.py

+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Utility functions for ArgGroup configuration."""
+
+import argparse
+import os
+from typing import Any, Optional, TypeVar
+
+T = TypeVar("T")
+
+
+def env_or_default(env_var: str, default: T) -> T:
+    """
+    Get value from environment variable or return default.
+
+    Performs type conversion based on the default value's type.
+
+    Args:
+        env_var: Environment variable name (e.g., "DYN_NAMESPACE")
+        default: Default value if env var not set
+
+    Returns:
+        Environment variable value (type-converted) or default
+
+    Examples:
+        >>> env_or_default("DYN_NAMESPACE", "test")
+        "test"  # if DYN_NAMESPACE not set
+        >>> env_or_default("DYN_MIGRATION_LIMIT", 0)
+        5  # if DYN_MIGRATION_LIMIT="5"
+    """
+    value = os.environ.get(env_var)
+    if value is None:
+        return default
+
+    # Type conversion based on default type
+    if isinstance(default, bool):
+        return value.lower() in ("true", "1", "yes", "on")  # type: ignore
+    elif isinstance(default, int):
+        return int(value)  # type: ignore
+    elif isinstance(default, float):
+        return float(value)  # type: ignore
+    else:
+        return value  # type: ignore
+
+
+def add_argument(
+    parser,
+    *,
+    flag_name: str,
+    env_var: str,
+    default: Any,
+    help: str,
+    obsolete_flag: Optional[str] = None,
+    arg_type: Optional[type] = str,
+    **kwargs: Any,
+) -> None:
+    """
+    Add a CLI argument with env var default, optional alias and dest, and help message construction.
+
+    Args:
+        parser: ArgumentParser or argument group
+        flag_name: Primary flag (must start with '--', e.g., "--foo")
+        env_var: Environment variable name (e.g., "DYN_FOO")
+        default: Default value
+        help: Help text
+        alias: Optional alias for the flag (must start with '--')
+        obsolete_flag: Optional obsolete/legacy flag (for help msg only, must start with '--')
+        dest: Optional destination name (defaults to flag_name with dashes replaced by underscores)
+        choices: Optional list of valid values for the argument.
+        arg_type: Type for the argument (default: str)
+    """
+    arg_dest = _get_dest_name(flag_name, kwargs.get("dest"))
+    default_with_env = env_or_default(env_var, default)
+
+    names = [flag_name]
+
+    env_help = _build_help_message(help, env_var, default_with_env, obsolete_flag)
+
+    add_arg_opts = {
+        "dest": arg_dest,
+        "default": default_with_env,
+        "help": env_help,
+        "type": arg_type,
+    }
+    kwargs.update(add_arg_opts)
+
+    parser.add_argument(*names, **kwargs)
+
+
+def add_negatable_bool_argument(
+    parser,
+    *,
+    flag_name: str,
+    env_var: str,
+    default: bool,
+    help: str,
+    dest: Optional[str] = None,
+) -> None:
+    """
+    Add negatable boolean flag (--foo / --no-foo).
+
+    Args:
+        parser: ArgumentParser or argument group
+        flag_name: Primary flag (must start with '--', e.g. "--enable-feature")
+        env_var: Environment variable name (e.g., "DYN_ENABLE_FEATURE")
+        default: Default value
+        help: Help text
+    """
+    arg_dest = _get_dest_name(flag_name, dest)
+    default_with_env = env_or_default(env_var, default)
+
+    parser.add_argument(
+        flag_name,
+        dest=arg_dest,
+        action=argparse.BooleanOptionalAction,
+        default=default_with_env,
+        help=_build_help_message(help, env_var, default),
+    )
+
+
+def _build_help_message(
+    help_text: str, env_var: str, default: Any, obsolete_flag: Optional[str] = None
+) -> str:
+    """
+    Build help message with env var and default value.
+    """
+    if obsolete_flag:
+        return f"{help_text}\nenv var: {env_var} | default: {default}\nobsolete flag: {obsolete_flag}"
+    return f"{help_text}\nenv var: {env_var} | default: {default}"
+
+
+def _get_dest_name(flag_name: str, dest: Optional[str] = None) -> str:
+    """
+    Get the destination name for the flag.
+    """
+    return dest if dest else flag_name.lstrip("-").replace("-", "_")

components/src/dynamo/common/tests/configuration/test_utils.py

+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Tests for configuration utility functions."""
+import argparse
+
+import pytest
+
+from dynamo.common.configuration.utils import (
+    add_negatable_bool_argument,
+    env_or_default,
+)
+
+pytestmark = [
+    pytest.mark.unit,
+    pytest.mark.pre_merge,
+]
+
+
+class TestEnvOrDefault:
+    """Test env_or_default function."""
+
+    def test_returns_default_when_env_not_set(self, monkeypatch):
+        """Test returns default value when env var not set."""
+        monkeypatch.delenv("TEST_VAR", raising=False)
+
+        result = env_or_default("TEST_VAR", "default_value")
+        assert result == "default_value"
+
+    def test_returns_env_when_set(self, monkeypatch):
+        """Test returns env value when set."""
+        monkeypatch.setenv("TEST_VAR", "env_value")
+
+        result = env_or_default("TEST_VAR", "default_value")
+        assert result == "env_value"
+
+    def test_bool_conversion_true(self, monkeypatch):
+        """Test bool conversion for true values."""
+        test_cases = ["true", "True", "1", "yes", "YES", "on", "ON"]
+
+        for value in test_cases:
+            monkeypatch.setenv("TEST_BOOL", value)
+            result = env_or_default("TEST_BOOL", False)
+            assert result is True, f"Failed for value: {value}"
+
+    def test_bool_conversion_false(self, monkeypatch):
+        """Test bool conversion for false values."""
+        test_cases = ["false", "False", "0", "no", "NO", "off", "OFF"]
+
+        for value in test_cases:
+            monkeypatch.setenv("TEST_BOOL", value)
+            result = env_or_default("TEST_BOOL", True)
+            assert result is False, f"Failed for value: {value}"
+
+    def test_int_conversion(self, monkeypatch):
+        """Test int conversion."""
+        monkeypatch.setenv("TEST_INT", "42")
+
+        result = env_or_default("TEST_INT", 0)
+        assert result == 42
+        assert isinstance(result, int)
+
+    def test_float_conversion(self, monkeypatch):
+        """Test float conversion."""
+        monkeypatch.setenv("TEST_FLOAT", "3.14")
+
+        result = env_or_default("TEST_FLOAT", 0.0)
+        assert result == 3.14
+        assert isinstance(result, float)
+
+    def test_string_passthrough(self, monkeypatch):
+        """Test string values are passed through."""
+        monkeypatch.setenv("TEST_STR", "hello world")
+
+        result = env_or_default("TEST_STR", "default")
+        assert result == "hello world"
+
+    def test_preserves_default_type(self, monkeypatch):
+        """Test that default type is preserved when env not set."""
+        monkeypatch.delenv("TEST_VAR", raising=False)
+
+        # String
+        assert isinstance(env_or_default("TEST_VAR", "str"), str)
+        # Int
+        assert isinstance(env_or_default("TEST_VAR", 42), int)
+        # Float
+        assert isinstance(env_or_default("TEST_VAR", 3.14), float)
+        # Bool
+        assert isinstance(env_or_default("TEST_VAR", True), bool)
+
+
+class TestAddNegatableBool:
+    """Test add_negatable_bool function."""
+
+    def test_positive_flag(self):
+        """Test that --flag added."""
+        parser = argparse.ArgumentParser()
+        add_negatable_bool_argument(
+            parser,
+            flag_name="--enable-feature",
+            env_var="TEST_ENABLE",
+            default=True,
+            help="Enable feature",
+        )
+
+        # Test positive flag
+        args = parser.parse_args(["--enable-feature"])
+        assert args.enable_feature is True
+
+    def test_negative_flag(self):
+        """Test that --no-flag are added."""
+        # Test negative flag
+        parser = argparse.ArgumentParser()
+        add_negatable_bool_argument(
+            parser,
+            flag_name="--enable-feature",
+            env_var="TEST_ENABLE",
+            default=True,
+            help="Enable feature",
+        )
+        args = parser.parse_args(["--no-enable-feature"])
+        assert args.enable_feature is False
+
+    def test_uses_default_when_no_flag(self, monkeypatch):
+        """Test uses default value when no flag provided."""
+        monkeypatch.delenv("TEST_ENABLE", raising=False)
+
+        parser = argparse.ArgumentParser()
+        add_negatable_bool_argument(
+            parser,
+            flag_name="--enable-feature",
+            env_var="TEST_ENABLE",
+            default=True,
+            help="Enable feature",
+        )
+
+        args = parser.parse_args([])
+        assert args.enable_feature is True
+
+    def test_uses_env_var_when_set(self, monkeypatch):
+        """Test uses environment variable when set."""
+        monkeypatch.setenv("TEST_ENABLE", "false")
+
+        parser = argparse.ArgumentParser()
+        add_negatable_bool_argument(
+            parser,
+            flag_name="--enable-feature",
+            env_var="TEST_ENABLE",
+            default=True,
+            help="Enable feature",
+        )
+
+        args = parser.parse_args([])
+        assert args.enable_feature is False
+
+    def test_converts_hyphens_to_underscores(self):
+        """Test that flag name with hyphens converts to underscores in dest."""
+        parser = argparse.ArgumentParser()
+        add_negatable_bool_argument(
+            parser,
+            flag_name="--my-cool-feature",
+            env_var="TEST_FEATURE",
+            default=False,
+            help="Cool feature",
+        )
+
+        args = parser.parse_args([])
+        # Should have my_cool_feature attribute
+        assert hasattr(args, "my_cool_feature")
+
+    def test_help_includes_env_var(self):
+        """Test that help text includes environment variable name."""
+        parser = argparse.ArgumentParser()
+        add_negatable_bool_argument(
+            parser,
+            flag_name="--feature",
+            env_var="MY_ENV_VAR",
+            default=True,
+            help="Test feature",
+        )
+
+        help_text = parser.format_help()
+        assert "MY_ENV_VAR" in help_text
+        assert "Test feature" in help_text
+
+    def test_help_shows_current_default(self, monkeypatch):
+        """Test that help shows the default value."""
+        monkeypatch.setenv("TEST_VAR", "true")
+
+        parser = argparse.ArgumentParser()
+        add_negatable_bool_argument(
+            parser,
+            flag_name="--feature",
+            env_var="TEST_VAR",
+            default=False,
+            help="Test",
+        )
+
+        help_text = parser.format_help()
+        assert "False" in help_text or "false" in help_text