Add Pydantic models for configuration validation

.ci_support/environment-docs.yml

@@ -13,5 +13,6 @@ dependencies:
 - tqdm =4.67.3
 - jupyter-book =1.0.0
 - python =3.12
+- pydantic =2.12.5
 - hatchling =1.29.0
 - hatch-vcs =0.5.0

.ci_support/environment-old.yml

@@ -7,5 +7,6 @@ dependencies:
 - jinja2 =2.11.3
 - paramiko =2.7.1
 - tqdm =4.66.1
+- pydantic =2.5.3
 - hatchling =1.27.0
 - hatch-vcs =0.4.0

.ci_support/environment.yml

@@ -10,3 +10,4 @@ dependencies:
 - tqdm =4.67.3
 - hatchling =1.29.0
 - hatch-vcs =0.5.0
+- pydantic =2.12.5

pyproject.toml

@@ -50,6 +50,7 @@ remote = [
     "tqdm==4.67.3",
 ]
 twofactor = ["pyauthenticator==0.3.0"]
+pydantic = ["pydantic==2.12.5"]
 
 [project.scripts]
 pysqa = "pysqa.cmd:command_line"

src/pysqa/base/config.py

@@ -9,6 +9,13 @@
 from pysqa.base.core import QueueAdapterCore, execute_command
 from pysqa.base.validate import check_queue_parameters, value_error_if_none
 
+try:
+    from pysqa.base.models import validate_config
+except ImportError:
+
+    def validate_config(config: dict) -> dict:
+        return config
+
 
 class Queues:
     """
@@ -73,10 +80,10 @@ def __init__(
         directory: str = "~/.queues",
         execute_command: Callable = execute_command,
     ):
+        self._config = validate_config(config)
         super().__init__(
-            queue_type=config["queue_type"], execute_command=execute_command
+            queue_type=self._config["queue_type"], execute_command=execute_command
         )
-        self._config = config
         self._fill_queue_dict(queue_lst_dict=self._config["queues"])
         self._load_templates(queue_lst_dict=self._config["queues"], directory=directory)
         self._queues = Queues(self.queue_list)
@@ -309,7 +316,7 @@ def _load_templates(queue_lst_dict: dict, directory: str = ".") -> None:
             directory (str, optional): The directory where the queue template files are located. Defaults to ".".
         """
         for queue_dict in queue_lst_dict.values():
-            if "script" in queue_dict:
+            if "script" in queue_dict and queue_dict["script"] is not None:
                 with open(os.path.join(directory, queue_dict["script"])) as f:
                     try:
                         queue_dict["template"] = Template(f.read())

src/pysqa/base/models.py

@@ -0,0 +1,59 @@
+from typing import Optional, Union
+
+from pydantic import BaseModel, ConfigDict
+
+
+class QueueModel(BaseModel):
+    """
+    Pydantic model for a single queue configuration.
+    """
+
+    model_config = ConfigDict(extra="allow")
+
+    script: Optional[str] = None
+    cores_min: Optional[int] = None
+    cores_max: Optional[int] = None
+    run_time_max: Optional[int] = None
+    memory_max: Optional[Union[int, str]] = None
+
+
+class ConfigModel(BaseModel):
+    """
+    Pydantic model for the overall configuration.
+    """
+
+    model_config = ConfigDict(extra="allow")
+
+    queue_type: str
+    queue_primary: Optional[str] = None
+    ssh_host: Optional[str] = None
+    ssh_username: Optional[str] = None
+    known_hosts: Optional[str] = None
+    ssh_key: Optional[str] = None
+    ssh_password: Optional[str] = None
+    ssh_ask_for_password: Optional[str] = None
+    ssh_key_passphrase: Optional[str] = None
+    ssh_two_factor_authentication: bool = False
+    ssh_authenticator_service: Optional[str] = None
+    ssh_proxy_host: Optional[str] = None
+    ssh_remote_config_dir: Optional[str] = None
+    ssh_remote_path: Optional[str] = None
+    ssh_local_path: Optional[str] = None
+    ssh_port: Optional[int] = None
+    ssh_continous_connection: bool = False
+    ssh_delete_file_on_remote: bool = True
+    python_executable: Optional[str] = None
+    queues: dict[str, QueueModel]
+
+
+def validate_config(config: dict) -> dict:
+    """
+    Validate the configuration dictionary against the ConfigModel.
+
+    Args:
+        config (dict): The configuration dictionary to validate.
+
+    Returns:
+        dict: The validated configuration dictionary.
+    """
+    return ConfigModel(**config).model_dump(exclude_none=True, exclude_defaults=True)

tests/unit/base/test_config.py

@@ -1,8 +1,70 @@
 import unittest
-from pysqa.base.config import QueueAdapterWithConfig
+import sys
+from unittest.mock import patch
+
+try:
+    import pydantic
+
+    skip_pydantic_test = False
+except ImportError:
+    skip_pydantic_test = True
 
 
 class TestConfig(unittest.TestCase):
     def test_bad_queue_type(self):
+        from pysqa.base.config import QueueAdapterWithConfig
+
         with self.assertRaises(ValueError):
             QueueAdapterWithConfig(config={"queue_type": "error", "queues": {}})
+
+    @unittest.skipIf(skip_pydantic_test, "pydantic not installed")
+    def test_pydantic_validation_missing_queues(self):
+        from pysqa.base.config import QueueAdapterWithConfig
+
+        with self.assertRaises(ValueError):
+            QueueAdapterWithConfig(config={"queue_type": "SLURM"})
+
+    @unittest.skipIf(skip_pydantic_test, "pydantic not installed")
+    def test_pydantic_validation_wrong_type(self):
+        from pysqa.base.config import QueueAdapterWithConfig
+
+        with self.assertRaises(ValueError):
+            QueueAdapterWithConfig(
+                config={
+                    "queue_type": "SLURM",
+                    "queues": {"sq": {"script": "slurm.sh", "cores_min": "not_an_int"}},
+                }
+            )
+
+    def test_pydantic_validation_extra_fields(self):
+        from pysqa.base.config import QueueAdapterWithConfig
+
+        config = {
+            "queue_type": "SLURM",
+            "queue_primary": "sq",
+            "queues": {"sq": {"extra_field": "allowed"}},
+            "extra_top_level": "also_allowed",
+        }
+        qa = QueueAdapterWithConfig(config=config)
+        self.assertEqual(qa.config["queues"]["sq"]["extra_field"], "allowed")
+        self.assertEqual(qa.config["extra_top_level"], "also_allowed")
+
+    def test_no_pydantic_validation_extra_fields(self):
+        with patch.dict('sys.modules', {'pydantic': None}):
+            if 'pysqa.base.models' in sys.modules:
+                del sys.modules['pysqa.base.models']
+
+            if 'pysqa.base.config' in sys.modules:
+                del sys.modules['pysqa.base.config']
+
+            from pysqa.base.config import QueueAdapterWithConfig
+
+            config = {
+                "queue_type": "SLURM",
+                "queue_primary": "sq",
+                "queues": {"sq": {"extra_field": "allowed"}},
+                "extra_top_level": "also_allowed",
+            }
+            qa = QueueAdapterWithConfig(config=config)
+            self.assertEqual(qa.config["queues"]["sq"]["extra_field"], "allowed")
+            self.assertEqual(qa.config["extra_top_level"], "also_allowed")