Refactor configuration file loading and validation

python/lsst/consdb/efd_transform/config_model.py

@@ -0,0 +1,66 @@
+from typing import Dict, List, Optional
+
+from pydantic import BaseModel
+
+
+class Field(BaseModel):
+    """
+    Represents a field in the Topic model.
+
+    Attributes:
+      name (str): The name of the field.
+      is_array (bool, optional): Indicates if the field is an array. Defaults to False.
+    """
+
+    name: str
+    is_array: Optional[bool] = False
+
+
+class Topic(BaseModel):
+    """
+    Represents a topic in the Column model.
+
+    Attributes:
+      name (str): The name of the topic.
+      fields (List[Field]): A list of fields associated with the topic.
+    """
+
+    name: str
+    fields: List[Field]
+
+
+class Column(BaseModel):
+    """
+    Represents a column in a database table.
+
+    Attributes:
+      name (str): The name of the column.
+      function (str): The function to be applied to the column.
+      function_args (Optional[Dict]): Optional arguments for the function.
+      type (str): The data type of the column.
+      unit (str): The unit of measurement for the column.
+      description (str): A description of the column.
+      tables (Optional[List[str]]): Optional list of tables where the column is
+      present.
+      topics (List[Topic]): List of topics associated with the column.
+    """
+
+    name: str
+    function: str
+    function_args: Optional[Dict] = None
+    type: str
+    unit: str
+    description: str
+    tables: Optional[List[str]] = ["ExposureEFD", "VisitEFD"]
+    topics: List[Topic]
+
+
+class ConfigModel(BaseModel):
+    """
+    Represents a configuration model.
+
+    Attributes:
+      columns (List[Column]): A list of columns.
+    """
+
+    columns: List[Column]

python/lsst/consdb/efd_transform/transform.py

@@ -129,26 +129,27 @@ async def process_interval(
             # topics = [{'name': topic name, series: pandas.DataFrame}]
             topics = await self.topics_by_column(column, topic_interval)
 
-            for exposure in exposures:
-                column_value = self.proccess_column_value(
-                    start_time=exposure["timespan"].begin,
-                    end_time=exposure["timespan"].end,
-                    topics=topics,
-                    transform_function=column["function"],
-                )
-
-                result_exp[exposure["id"]][column["name"]] = column_value
-
-            for visit in visits:
-
-                column_value = self.proccess_column_value(
-                    start_time=visit["timespan"].begin,
-                    end_time=visit["timespan"].end,
-                    topics=topics,
-                    transform_function=column["function"],
-                )
-
-                result_vis[visit["id"]][column["name"]] = column_value
+            if "ExposureEFD" in column["tables"]:
+                for exposure in exposures:
+                    column_value = self.proccess_column_value(
+                        start_time=exposure["timespan"].begin,
+                        end_time=exposure["timespan"].end,
+                        topics=topics,
+                        transform_function=column["function"],
+                    )
+
+                    result_exp[exposure["id"]][column["name"]] = column_value
+
+            if "VisitEFD" in column["tables"]:
+                for visit in visits:
+                    column_value = self.proccess_column_value(
+                        start_time=visit["timespan"].begin,
+                        end_time=visit["timespan"].end,
+                        topics=topics,
+                        transform_function=column["function"],
+                    )
+
+                    result_vis[visit["id"]][column["name"]] = column_value
 
         results = []
         for result_row in result_exp:

python/lsst/consdb/transform_efd.py

@@ -8,8 +8,10 @@
 import astropy.time
 import lsst_efd_client
 import yaml
+from efd_transform.config_model import ConfigModel
 from efd_transform.transform import Transform
 from lsst.daf.butler import Butler
+from pydantic import ValidationError
 
 # from sqlalchemy import create_engine
 
@@ -117,16 +119,27 @@ def build_argparser() -> argparse.ArgumentParser:
 
 def read_config(config_name: str) -> Dict[str, Any]:
     """
-    Reads a configuration file and returns its contents as a dictionary.
+    Reads a configuration file and returns the configuration as a dictionary.
 
     Args:
         config_name (str): The name of the configuration file.
 
     Returns:
-        dict[str, Any]: The contents of the configuration file as a dictionary.
+        dict: The configuration as a dictionary.
+
+    Raises:
+        ValidationError: If the configuration file is invalid.
+
     """
-    with open(config_name, "r") as f:
-        return yaml.safe_load(f)
+    try:
+        with open(config_name, "r") as file:
+            data = yaml.safe_load(file)
+            config = ConfigModel(**data)
+
+            return config.model_dump()
+
+    except ValidationError as e:
+        raise e
 
 
 async def main() -> None: