diff --git a/python/felis/datamodel.py b/python/felis/datamodel.py index 25a19b03..f0e2100e 100644 --- a/python/felis/datamodel.py +++ b/python/felis/datamodel.py @@ -24,12 +24,16 @@ from __future__ import annotations import logging +import os from collections.abc import Sequence from enum import StrEnum, auto -from typing import Annotated, Any, Literal, TypeAlias, TypeVar, Union +from pathlib import Path +from typing import IO, Annotated, Any, Literal, TypeAlias, TypeVar, Union +import yaml from astropy import units as units # type: ignore from astropy.io.votable import ucd # type: ignore +from lsst.resources import ResourcePath from pydantic import BaseModel, ConfigDict, Field, ValidationInfo, field_validator, model_validator from .db.dialects import get_supported_dialects @@ -253,7 +257,7 @@ def check_units(self) -> Column: Raises ------ ValueError - Raised If both FITS and IVOA units are provided, or if the unit is + Raised if both FITS and IVOA units are provided, or if the unit is invalid. """ fits_unit = self.fits_tunit @@ -1000,3 +1004,54 @@ def get_table_by_column(self, column: Column) -> Table: if column in table.columns: return table raise ValueError(f"Column '{column.name}' not found in any table") + + @classmethod + def from_resource( + cls, resource_path: str | ResourcePath, context: dict[str, Any] = {}, validate: bool = True + ) -> Schema: + """Load a `Schema` from a string representing a ``ResourcePath``. + + Parameters + ---------- + resource_path + The ``ResourcePath`` pointing to a YAML file. + + Returns + ------- + `str` + The ID of the object. + """ + logger.debug(f"Loading schema from '{resource_path}'") + if isinstance(resource_path, str): + resource_path = ResourcePath(resource_path) + data = resource_path.read() + try: + yaml_data = yaml.safe_load(data) + except yaml.YAMLError as e: + raise ValueError(f"Error loading YAML file from '{resource_path}': {e}") + return Schema.model_validate(yaml_data, context=context) + + @classmethod + def from_file(cls, source: str | Path | IO[str], context: dict[str, Any] = {}) -> Schema: + """Load a `Schema` from a file, a ``Path``, or a string representing a + file system path. + + Parameters + ---------- + source + The file, path, or file object to load the schema from. + + Returns + ------- + `Schema` + The schema loaded from the source. + """ + logger.debug("Loading schema from '%s'", source) + if isinstance(source, str) or isinstance(source, Path): + if not os.path.exists(source): + raise FileNotFoundError(f"File '{source}' not found") + with open(source) as file: + yaml_data = yaml.safe_load(file) + else: + yaml_data = yaml.safe_load(source) + return Schema.model_validate(yaml_data, context=context)