Skip to content

Commit

Permalink
Add class methods for creating a Schema from a file or resource
Browse files Browse the repository at this point in the history
Add method for creating a Schema from a string path, Path object, or
existing Python file object.

Add method for creating a Schema from an lsst-resources ResourcePath
object or a resource string from which the ResourcePath will be
created automatically.
  • Loading branch information
JeremyMcCormick committed Sep 3, 2024
1 parent eadb4ce commit bd35258
Showing 1 changed file with 57 additions and 2 deletions.
59 changes: 57 additions & 2 deletions python/felis/datamodel.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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)

0 comments on commit bd35258

Please sign in to comment.