Pydantic request validation

pyproject.toml

@@ -42,6 +42,7 @@ dependencies = [
     "email-validator>=1.1",
     "imas-python",
     "numpy>=1.14",
+    "pydantic>=2.10.6",
     "python-dateutil>=2.6",
     "pyuda>=2.9.2",
     "pyyaml>=3.13",

src/simdb/cli/simdb.py

@@ -1,6 +1,6 @@
 import copy
 import sys
-from typing import IO
+from typing import TextIO
 
 import click
 
@@ -53,7 +53,7 @@ def list_commands(self, ctx):
 @click.option("-v", "--verbose", is_flag=True, help="Run with verbose output.")
 @click.option("-c", "--config-file", type=click.File("r"), help="Config file to load.")
 @click.pass_context
-def cli(ctx, debug: bool, verbose: bool, config_file: IO):
+def cli(ctx, debug: bool, verbose: bool, config_file: TextIO):
     if not ctx.obj:
         ctx.obj = Config()
         ctx.obj.load(config_file)

src/simdb/database/database.py

@@ -16,6 +16,7 @@
 
 from simdb.config import Config
 from simdb.query import QueryType, query_compare
+from simdb.remote.models import SimulationReference
 
 from .models import Base
 from .models.file import File
@@ -226,8 +227,8 @@ def _find_simulation(self, sim_ref: str) -> "Simulation":
                 )
             except SQLAlchemyError:
                 simulation = None
-            if not simulation:
-                raise DatabaseError(f"Simulation {sim_ref} not found.") from None
+        if not simulation:
+            raise DatabaseError(f"Simulation {sim_ref} not found.") from None
         return simulation
 
     def remove(self):
@@ -571,6 +572,24 @@ def get_simulation_parents(self, simulation: "Simulation") -> List[dict]:
         )
         return [{"uuid": r.uuid, "alias": r.alias} for r in query.all()]
 
+    def get_simulation_parents_ref(
+        self, simulation: "Simulation"
+    ) -> List[SimulationReference]:
+        subquery = (
+            self.session.query(File.checksum)
+            .filter(File.checksum != "")
+            .filter(File.input_for.contains(simulation))
+            .subquery()
+        )
+        query = (
+            self.session.query(Simulation.uuid, Simulation.alias)
+            .join(Simulation.outputs)
+            .filter(File.checksum.in_(subquery))
+            .filter(Simulation.alias != simulation.alias)
+            .distinct()
+        )
+        return [SimulationReference(uuid=r.uuid, alias=r.alias) for r in query.all()]
+
     def get_simulation_children(self, simulation: "Simulation") -> List[dict]:
         subquery = (
             self.session.query(File.checksum)
@@ -587,6 +606,24 @@ def get_simulation_children(self, simulation: "Simulation") -> List[dict]:
         )
         return [{"uuid": r.uuid, "alias": r.alias} for r in query.all()]
 
+    def get_simulation_children_ref(
+        self, simulation: "Simulation"
+    ) -> List[SimulationReference]:
+        subquery = (
+            self.session.query(File.checksum)
+            .filter(File.checksum != "")
+            .filter(File.output_of.contains(simulation))
+            .subquery()
+        )
+        query = (
+            self.session.query(Simulation.uuid, Simulation.alias)
+            .join(Simulation.inputs)
+            .filter(File.checksum.in_(subquery))
+            .filter(Simulation.alias != simulation.alias)
+            .distinct()
+        )
+        return [SimulationReference(uuid=r.uuid, alias=r.alias) for r in query.all()]
+
     def get_file(self, file_uuid_str: str) -> "File":
         """
         Get the specified file from the database.

src/simdb/database/models/file.py

@@ -14,6 +14,7 @@
 from simdb.docstrings import inherit_docstrings
 from simdb.imas.checksum import checksum as imas_checksum
 from simdb.imas.utils import imas_timestamp
+from simdb.remote.models import FileData
 from simdb.uda.checksum import checksum as uda_checksum
 
 from .base import Base
@@ -125,6 +126,23 @@ def from_data(cls, data: Dict) -> "File":
         file.datetime = date_parser.parse(checked_get(data, "datetime", str))
         return file
 
+    @classmethod
+    def from_data_model(cls, data: FileData) -> "File":
+        data_type = data.type
+        uri = data.uri
+        file = File(
+            DataObject.Type[data_type], urilib.URI(uri), perform_integrity_check=False
+        )
+        file.uuid = data.uuid
+        file.usage = data.usage
+        file.checksum = data.checksum
+        file.purpose = data.purpose
+        file.sensitivity = data.sensitivity
+        file.access = data.access
+        file.embargo = data.embargo
+        file.datetime = data.datetime
+        return file
+
     def data(self, recurse: bool = False) -> Dict[str, str]:
         data = {
             "uuid": self.uuid,
@@ -139,3 +157,17 @@ def data(self, recurse: bool = False) -> Dict[str, str]:
             "datetime": self.datetime.isoformat(),
         }
         return data
+
+    def to_model(self) -> FileData:
+        return FileData(
+            type=self.type.name,
+            uri=str(self.uri),
+            uuid=self.uuid,
+            checksum=self.checksum,
+            datetime=self.datetime,
+            usage=self.usage,
+            purpose=self.purpose,
+            sensitivity=self.sensitivity,
+            access=self.access,
+            embargo=self.embargo,
+        )

src/simdb/database/models/metadata.py

@@ -4,6 +4,7 @@
 from sqlalchemy import types as sql_types
 
 from simdb.docstrings import inherit_docstrings
+from simdb.remote.models import MetadataData
 
 from .base import Base
 
@@ -32,12 +33,20 @@ def from_data(cls, data: Dict) -> "MetaData":
         meta = MetaData(data["element"], data["value"])
         return meta
 
+    @classmethod
+    def from_data_model(cls, data: MetadataData) -> "MetaData":
+        meta = MetaData(data.element, data.value)
+        return meta
+
     def data(self, recurse: bool = False) -> Dict[str, str]:
         data = {
             "element": self.element,
             "value": self.value,
         }
         return data
 
+    def to_model(self) -> MetadataData:
+        return MetadataData(element=self.element, value=self.value)
+
 
 Index("metadata_index", MetaData.sim_id, MetaData.element, unique=True)

src/simdb/database/models/simulation.py

@@ -9,6 +9,8 @@
 from pathlib import Path
 from typing import Any, Dict, List, Optional, Set, Union
 
+from simdb.remote.models import FileDataList, MetadataDataList, SimulationData
+
 if sys.version_info < (3, 11):
     from backports.datetime_fromisoformat import MonkeyPatch
 
@@ -336,6 +338,17 @@ def from_data(cls, data: Dict[str, Union[str, Dict, List]]) -> "Simulation":
                 simulation.meta.append(MetaData.from_data(el))
         return simulation
 
+    @classmethod
+    def from_data_model(cls, data: SimulationData) -> "Simulation":
+        simulation = Simulation(None)
+        simulation.uuid = data.uuid
+        simulation.alias = data.alias
+        simulation.datetime = data.datetime
+        simulation.inputs = [File.from_data_model(el) for el in data.inputs.root]
+        simulation.outputs = [File.from_data_model(el) for el in data.outputs.root]
+        simulation.meta = [MetaData.from_data_model(el) for el in data.metadata.root]
+        return simulation
+
     def data(
         self, recurse: bool = False, meta_keys: Optional[List[str]] = None
     ) -> Dict[str, Union[str, List]]:
@@ -354,6 +367,29 @@ def data(
             ]
         return data
 
+    def to_model(
+        self, recurse: bool = False, meta_keys: Optional[List[str]] = None
+    ) -> SimulationData:
+        inputs = FileDataList()
+        outputs = FileDataList()
+        metadata = MetadataDataList()
+        if recurse:
+            inputs = FileDataList([f.to_model() for f in self.inputs])
+            outputs = FileDataList([f.to_model() for f in self.outputs])
+            metadata = MetadataDataList([m.to_model() for m in self.meta])
+        elif meta_keys:
+            metadata = MetadataDataList(
+                [m.to_model() for m in self.meta if m.element in meta_keys]
+            )
+        return SimulationData(
+            uuid=self.uuid,
+            alias=self.alias,
+            datetime=self.datetime,
+            inputs=inputs,
+            outputs=outputs,
+            metadata=metadata,
+        )
+
     def meta_dict(self) -> Dict[str, Union[Dict, Any]]:
         meta = {m.element: m.value for m in self.meta}
         return unflatten_dict(meta)