Coverage for python / felis / datamodel.py: 31%
712 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-30 01:39 -0800
1"""Define Pydantic data models for Felis."""
3# This file is part of felis.
4#
5# Developed for the LSST Data Management System.
6# This product includes software developed by the LSST Project
7# (https://www.lsst.org).
8# See the COPYRIGHT file at the top-level directory of this distribution
9# for details of code ownership.
10#
11# This program is free software: you can redistribute it and/or modify
12# it under the terms of the GNU General Public License as published by
13# the Free Software Foundation, either version 3 of the License, or
14# (at your option) any later version.
15#
16# This program is distributed in the hope that it will be useful,
17# but WITHOUT ANY WARRANTY; without even the implied warranty of
18# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
19# GNU General Public License for more details.
20#
21# You should have received a copy of the GNU General Public License
22# along with this program. If not, see <https://www.gnu.org/licenses/>.
24from __future__ import annotations
26import json
27import logging
28import sys
29from collections.abc import Sequence
30from enum import StrEnum, auto
31from typing import IO, Annotated, Any, Generic, Literal, TypeAlias, TypeVar
33import yaml
34from astropy import units as units # type: ignore
35from astropy.io.votable import ucd # type: ignore
36from lsst.resources import ResourcePath, ResourcePathExpression
37from pydantic import (
38 BaseModel,
39 ConfigDict,
40 Field,
41 PrivateAttr,
42 ValidationError,
43 ValidationInfo,
44 field_serializer,
45 field_validator,
46 model_validator,
47)
48from pydantic_core import InitErrorDetails
50from .db._dialects import get_supported_dialects, string_to_typeengine
51from .db._sqltypes import get_type_func
52from .types import Boolean, Byte, Char, Double, FelisType, Float, Int, Long, Short, String, Text, Unicode
54logger = logging.getLogger(__name__)
56__all__ = (
57 "BaseObject",
58 "CheckConstraint",
59 "Column",
60 "ColumnOverrides",
61 "ColumnResourceRef",
62 "Constraint",
63 "DataType",
64 "ForeignKeyConstraint",
65 "Index",
66 "Resource",
67 "Schema",
68 "SchemaVersion",
69 "Table",
70 "UniqueConstraint",
71)
73CONFIG = ConfigDict(
74 populate_by_name=True, # Populate attributes by name.
75 extra="forbid", # Do not allow extra fields.
76 str_strip_whitespace=True, # Strip whitespace from string fields.
77 use_enum_values=False, # Do not use enum values during serialization.
78)
79"""Pydantic model configuration as described in:
80https://docs.pydantic.dev/2.0/api/config/#pydantic.config.ConfigDict
81"""
83DESCR_MIN_LENGTH = 3
84"""Minimum length for a description field."""
86DescriptionStr: TypeAlias = Annotated[str, Field(min_length=DESCR_MIN_LENGTH)]
87"""Type for a description, which must be three or more characters long."""
90class BaseObject(BaseModel):
91 """Base model.
93 All classes representing objects in the Felis data model should inherit
94 from this class.
95 """
97 model_config = CONFIG
98 """Pydantic model configuration."""
100 name: str
101 """Name of the database object."""
103 id: str = Field(alias="@id")
104 """Unique identifier of the database object."""
106 description: DescriptionStr | None = None
107 """Description of the database object."""
109 votable_utype: str | None = Field(None, alias="votable:utype")
110 """VOTable utype (usage-specific or unique type) of the object."""
112 @model_validator(mode="after")
113 def check_description(self, info: ValidationInfo) -> BaseObject:
114 """Check that the description is present if required.
116 Parameters
117 ----------
118 info
119 Validation context used to determine if the check is enabled.
121 Returns
122 -------
123 `BaseObject`
124 The object being validated.
125 """
126 context = info.context
127 if not context or not context.get("check_description", False):
128 return self
129 if self.description is None or self.description == "":
130 raise ValueError("Description is required and must be non-empty")
131 if len(self.description) < DESCR_MIN_LENGTH:
132 raise ValueError(f"Description must be at least {DESCR_MIN_LENGTH} characters long")
133 return self
136class DataType(StrEnum):
137 """``Enum`` representing the data types supported by Felis."""
139 boolean = auto()
140 byte = auto()
141 short = auto()
142 int = auto()
143 long = auto()
144 float = auto()
145 double = auto()
146 char = auto()
147 string = auto()
148 unicode = auto()
149 text = auto()
150 binary = auto()
151 timestamp = auto()
154def validate_ivoa_ucd(ivoa_ucd: str) -> str:
155 """Validate IVOA UCD values.
157 Parameters
158 ----------
159 ivoa_ucd
160 IVOA UCD value to check.
162 Returns
163 -------
164 `str`
165 The IVOA UCD value if it is valid.
167 Raises
168 ------
169 ValueError
170 If the IVOA UCD value is invalid.
171 """
172 if ivoa_ucd is not None:
173 try:
174 ucd.parse_ucd(ivoa_ucd, check_controlled_vocabulary=True, has_colon=";" in ivoa_ucd)
175 except ValueError as e:
176 raise ValueError(f"Invalid IVOA UCD: {e}")
177 return ivoa_ucd
180class Column(BaseObject):
181 """Column model."""
183 datatype: DataType
184 """Datatype of the column."""
186 length: int | None = Field(None, gt=0)
187 """Length of the column."""
189 precision: int | None = Field(None, ge=0)
190 """The numerical precision of the column.
192 For timestamps, this is the number of fractional digits retained in the
193 seconds field.
194 """
196 nullable: bool = True
197 """Whether the column can be ``NULL``."""
199 value: str | int | float | bool | None = None
200 """Default value of the column."""
202 autoincrement: bool | None = None
203 """Whether the column is autoincremented."""
205 ivoa_ucd: str | None = Field(None, alias="ivoa:ucd")
206 """IVOA UCD of the column."""
208 fits_tunit: str | None = Field(None, alias="fits:tunit")
209 """FITS TUNIT of the column."""
211 ivoa_unit: str | None = Field(None, alias="ivoa:unit")
212 """IVOA unit of the column."""
214 tap_column_index: int | None = Field(None, alias="tap:column_index")
215 """TAP_SCHEMA column index of the column."""
217 tap_principal: int | None = Field(0, alias="tap:principal", ge=0, le=1)
218 """Whether this is a TAP_SCHEMA principal column."""
220 votable_arraysize: int | str | None = Field(None, alias="votable:arraysize")
221 """VOTable arraysize of the column."""
223 tap_std: int | None = Field(0, alias="tap:std", ge=0, le=1)
224 """TAP_SCHEMA indication that this column is defined by an IVOA standard.
225 """
227 votable_xtype: str | None = Field(None, alias="votable:xtype")
228 """VOTable xtype (extended type) of the column."""
230 votable_datatype: str | None = Field(None, alias="votable:datatype")
231 """VOTable datatype of the column."""
233 mysql_datatype: str | None = Field(None, alias="mysql:datatype")
234 """MySQL datatype override on the column."""
236 postgresql_datatype: str | None = Field(None, alias="postgresql:datatype")
237 """PostgreSQL datatype override on the column."""
239 _is_resource_ref: bool = PrivateAttr(False)
240 """Whether this column is a resource reference column."""
242 @model_validator(mode="after")
243 def check_value(self) -> Column:
244 """Check that the default value is valid.
246 Returns
247 -------
248 `Column`
249 The column being validated.
250 """
251 if (value := self.value) is not None:
252 if value is not None and self.autoincrement is True:
253 raise ValueError("Column cannot have both a default value and be autoincremented")
254 felis_type = FelisType.felis_type(self.datatype)
255 if felis_type.is_numeric:
256 if felis_type in (Byte, Short, Int, Long) and not isinstance(value, int):
257 raise ValueError("Default value must be an int for integer type columns")
258 elif felis_type in (Float, Double) and not isinstance(value, float):
259 raise ValueError("Default value must be a decimal number for float and double columns")
260 elif felis_type in (String, Char, Unicode, Text):
261 if not isinstance(value, str):
262 raise ValueError("Default value must be a string for string columns")
263 if not len(value):
264 raise ValueError("Default value must be a non-empty string for string columns")
265 elif felis_type is Boolean and not isinstance(value, bool):
266 raise ValueError("Default value must be a boolean for boolean columns")
267 return self
269 @field_validator("ivoa_ucd")
270 @classmethod
271 def check_ivoa_ucd(cls, ivoa_ucd: str) -> str:
272 """Check that IVOA UCD values are valid.
274 Parameters
275 ----------
276 ivoa_ucd
277 IVOA UCD value to check.
279 Returns
280 -------
281 `str`
282 The IVOA UCD value if it is valid.
283 """
284 return validate_ivoa_ucd(ivoa_ucd)
286 @model_validator(mode="after")
287 def check_units(self) -> Column:
288 """Check that the ``fits:tunit`` or ``ivoa:unit`` field has valid
289 units according to astropy. Only one may be provided.
291 Returns
292 -------
293 `Column`
294 The column being validated.
296 Raises
297 ------
298 ValueError
299 Raised if both FITS and IVOA units are provided, or if the unit is
300 invalid.
301 """
302 fits_unit = self.fits_tunit
303 ivoa_unit = self.ivoa_unit
305 if fits_unit and ivoa_unit:
306 raise ValueError("Column cannot have both FITS and IVOA units")
307 unit = fits_unit or ivoa_unit
309 if unit is not None:
310 try:
311 units.Unit(unit)
312 except ValueError as e:
313 raise ValueError(f"Invalid unit: {e}")
315 return self
317 @model_validator(mode="before")
318 @classmethod
319 def check_length(cls, values: dict[str, Any]) -> dict[str, Any]:
320 """Check that a valid length is provided for sized types.
322 Parameters
323 ----------
324 values
325 Values of the column.
327 Returns
328 -------
329 `dict` [ `str`, `Any` ]
330 The values of the column.
332 Raises
333 ------
334 ValueError
335 Raised if a length is not provided for a sized type.
336 """
337 datatype = values.get("datatype")
338 if datatype is None:
339 # Skip this validation if datatype is not provided
340 return values
341 length = values.get("length")
342 felis_type = FelisType.felis_type(datatype)
343 if felis_type.is_sized and length is None:
344 raise ValueError(
345 f"Length must be provided for type '{datatype}'"
346 + (f" in column '{values['@id']}'" if "@id" in values else "")
347 )
348 elif not felis_type.is_sized and length is not None:
349 logger.warning(
350 f"The datatype '{datatype}' does not support a specified length"
351 + (f" in column '{values['@id']}'" if "@id" in values else "")
352 )
353 return values
355 @model_validator(mode="after")
356 def check_redundant_datatypes(self, info: ValidationInfo) -> Column:
357 """Check for redundant datatypes on columns.
359 Parameters
360 ----------
361 info
362 Validation context used to determine if the check is enabled.
364 Returns
365 -------
366 `Column`
367 The column being validated.
369 Raises
370 ------
371 ValueError
372 Raised if a datatype override is redundant.
373 """
374 context = info.context
375 if not context or not context.get("check_redundant_datatypes", False):
376 return self
377 if all(
378 getattr(self, f"{dialect}:datatype", None) is not None
379 for dialect in get_supported_dialects().keys()
380 ):
381 return self
383 datatype = self.datatype
384 length: int | None = self.length or None
386 datatype_func = get_type_func(datatype)
387 felis_type = FelisType.felis_type(datatype)
388 if felis_type.is_sized:
389 datatype_obj = datatype_func(length)
390 else:
391 datatype_obj = datatype_func()
393 for dialect_name, dialect in get_supported_dialects().items():
394 db_annotation = f"{dialect_name}_datatype"
395 if datatype_string := self.model_dump().get(db_annotation):
396 db_datatype_obj = string_to_typeengine(datatype_string, dialect, length)
397 if datatype_obj.compile(dialect) == db_datatype_obj.compile(dialect):
398 raise ValueError(
399 "'{}: {}' is a redundant override of 'datatype: {}' in column '{}'{}".format(
400 db_annotation,
401 datatype_string,
402 self.datatype,
403 self.id,
404 "" if length is None else f" with length {length}",
405 )
406 )
407 else:
408 logger.debug(
409 f"Type override of 'datatype: {self.datatype}' "
410 f"with '{db_annotation}: {datatype_string}' in column '{self.id}' "
411 f"compiled to '{datatype_obj.compile(dialect)}' and "
412 f"'{db_datatype_obj.compile(dialect)}'"
413 )
414 return self
416 @model_validator(mode="after")
417 def check_precision(self) -> Column:
418 """Check that precision is only valid for timestamp columns.
420 Returns
421 -------
422 `Column`
423 The column being validated.
424 """
425 if self.precision is not None and self.datatype != "timestamp":
426 raise ValueError("Precision is only valid for timestamp columns")
427 return self
429 @model_validator(mode="before")
430 @classmethod
431 def check_votable_arraysize(cls, values: dict[str, Any], info: ValidationInfo) -> dict[str, Any]:
432 """Set the default value for the ``votable_arraysize`` field, which
433 corresponds to ``arraysize`` in the IVOA VOTable standard.
435 Parameters
436 ----------
437 values
438 Values of the column.
439 info
440 Validation context used to determine if the check is enabled.
442 Returns
443 -------
444 `dict` [ `str`, `Any` ]
445 The values of the column.
447 Notes
448 -----
449 Following the IVOA VOTable standard, an ``arraysize`` of 1 should not
450 be used.
451 """
452 if values.get("name", None) is None or values.get("datatype", None) is None:
453 # Skip bad column data that will not validate
454 return values
455 context = info.context if info.context else {}
456 arraysize = values.get("votable:arraysize", None)
457 if arraysize is None:
458 length = values.get("length", None)
459 datatype = values.get("datatype")
460 if length is not None and length > 1:
461 # Following the IVOA standard, arraysize of 1 is disallowed
462 if datatype == "char":
463 arraysize = str(length)
464 elif datatype in ("string", "unicode", "binary"):
465 if context.get("force_unbounded_arraysize", False):
466 arraysize = "*"
467 logger.debug(
468 f"Forced VOTable's 'arraysize' to '*' on column '{values['name']}' with datatype "
469 + f"'{values['datatype']}' and length '{length}'"
470 )
471 else:
472 arraysize = f"{length}*"
473 elif datatype in ("timestamp", "text"):
474 arraysize = "*"
475 if arraysize is not None:
476 values["votable:arraysize"] = arraysize
477 logger.debug(
478 f"Set default 'votable:arraysize' to '{arraysize}' on column '{values['name']}'"
479 + f" with datatype '{values['datatype']}' and length '{values.get('length', None)}'"
480 )
481 else:
482 logger.debug(f"Using existing 'votable:arraysize' of '{arraysize}' on column '{values['name']}'")
483 if isinstance(values["votable:arraysize"], int):
484 logger.warning(
485 f"Usage of an integer value for 'votable:arraysize' in column '{values['name']}' is "
486 + "deprecated"
487 )
488 values["votable:arraysize"] = str(arraysize)
489 return values
491 @field_serializer("datatype")
492 def serialize_datatype(self, value: DataType) -> str:
493 """Convert `DataType` to string when serializing to JSON/YAML.
495 Parameters
496 ----------
497 value
498 The `DataType` value to serialize.
500 Returns
501 -------
502 `str`
503 The serialized `DataType` value.
504 """
505 return str(value)
507 @field_validator("datatype", mode="before")
508 @classmethod
509 def deserialize_datatype(cls, value: str) -> DataType:
510 """Convert string back into `DataType` when loading from JSON/YAML.
512 Parameters
513 ----------
514 value
515 The string value to deserialize.
517 Returns
518 -------
519 `DataType`
520 The deserialized `DataType` value.
521 """
522 return DataType(value)
524 @model_validator(mode="after")
525 def check_votable_xtype(self) -> Column:
526 """Set the default value for the ``votable_xtype`` field, which
527 corresponds to an Extended Datatype or ``xtype`` in the IVOA VOTable
528 standard.
530 Returns
531 -------
532 `Column`
533 The column being validated.
535 Notes
536 -----
537 This is currently only set automatically for the Felis ``timestamp``
538 datatype.
539 """
540 if self.datatype == DataType.timestamp and self.votable_xtype is None:
541 self.votable_xtype = "timestamp"
542 return self
545class Constraint(BaseObject):
546 """Table constraint model."""
548 deferrable: bool = False
549 """Whether this constraint will be declared as deferrable."""
551 initially: Literal["IMMEDIATE", "DEFERRED"] | None = None
552 """Value for ``INITIALLY`` clause; only used if `deferrable` is
553 `True`."""
555 @model_validator(mode="after")
556 def check_deferrable(self) -> Constraint:
557 """Check that the ``INITIALLY`` clause is only used if `deferrable` is
558 `True`.
560 Returns
561 -------
562 `Constraint`
563 The constraint being validated.
564 """
565 if self.initially is not None and not self.deferrable:
566 raise ValueError("INITIALLY clause can only be used if deferrable is True")
567 return self
570class CheckConstraint(Constraint):
571 """Table check constraint model."""
573 type: Literal["Check"] = Field("Check", alias="@type")
574 """Type of the constraint."""
576 expression: str
577 """Expression for the check constraint."""
579 @field_serializer("type")
580 def serialize_type(self, value: str) -> str:
581 """Ensure '@type' is included in serialized output.
583 Parameters
584 ----------
585 value
586 The value to serialize.
588 Returns
589 -------
590 `str`
591 The serialized value.
592 """
593 return value
596class UniqueConstraint(Constraint):
597 """Table unique constraint model."""
599 type: Literal["Unique"] = Field("Unique", alias="@type")
600 """Type of the constraint."""
602 columns: list[str]
603 """Columns in the unique constraint."""
605 @field_serializer("type")
606 def serialize_type(self, value: str) -> str:
607 """Ensure '@type' is included in serialized output.
609 Parameters
610 ----------
611 value
612 The value to serialize.
614 Returns
615 -------
616 `str`
617 The serialized value.
618 """
619 return value
622class ForeignKeyConstraint(Constraint):
623 """Table foreign key constraint model.
625 This constraint is used to define a foreign key relationship between two
626 tables in the schema. There must be at least one column in the
627 `columns` list, and at least one column in the `referenced_columns` list
628 or a validation error will be raised.
630 Notes
631 -----
632 These relationships will be reflected in the TAP_SCHEMA ``keys`` and
633 ``key_columns`` data.
634 """
636 type: Literal["ForeignKey"] = Field("ForeignKey", alias="@type")
637 """Type of the constraint."""
639 columns: list[str] = Field(min_length=1)
640 """The columns comprising the foreign key."""
642 referenced_columns: list[str] = Field(alias="referencedColumns", min_length=1)
643 """The columns referenced by the foreign key."""
645 on_delete: Literal["CASCADE", "SET NULL", "SET DEFAULT", "RESTRICT", "NO ACTION"] | None = None
646 """Action to take when the referenced row is deleted."""
648 on_update: Literal["CASCADE", "SET NULL", "SET DEFAULT", "RESTRICT", "NO ACTION"] | None = None
649 """Action to take when the referenced row is updated."""
651 @field_serializer("type")
652 def serialize_type(self, value: str) -> str:
653 """Ensure '@type' is included in serialized output.
655 Parameters
656 ----------
657 value
658 The value to serialize.
660 Returns
661 -------
662 `str`
663 The serialized value.
664 """
665 return value
667 @model_validator(mode="after")
668 def check_column_lengths(self) -> ForeignKeyConstraint:
669 """Check that the `columns` and `referenced_columns` lists have the
670 same length.
672 Returns
673 -------
674 `ForeignKeyConstraint`
675 The foreign key constraint being validated.
677 Raises
678 ------
679 ValueError
680 Raised if the `columns` and `referenced_columns` lists do not have
681 the same length.
682 """
683 if len(self.columns) != len(self.referenced_columns):
684 raise ValueError(
685 "Columns and referencedColumns must have the same length for a ForeignKey constraint"
686 )
687 return self
690_ConstraintType = Annotated[
691 CheckConstraint | ForeignKeyConstraint | UniqueConstraint, Field(discriminator="type")
692]
693"""Type alias for a constraint type."""
696class Index(BaseObject):
697 """Table index model.
699 An index can be defined on either columns or expressions, but not both.
700 """
702 columns: list[str] | None = None
703 """Columns in the index."""
705 expressions: list[str] | None = None
706 """Expressions in the index."""
708 @model_validator(mode="before")
709 @classmethod
710 def check_columns_or_expressions(cls, values: dict[str, Any]) -> dict[str, Any]:
711 """Check that columns or expressions are specified, but not both.
713 Parameters
714 ----------
715 values
716 Values of the index.
718 Returns
719 -------
720 `dict` [ `str`, `Any` ]
721 The values of the index.
723 Raises
724 ------
725 ValueError
726 Raised if both columns and expressions are specified, or if neither
727 are specified.
728 """
729 if "columns" in values and "expressions" in values:
730 raise ValueError("Defining columns and expressions is not valid")
731 elif "columns" not in values and "expressions" not in values:
732 raise ValueError("Must define columns or expressions")
733 return values
736ColumnRef: TypeAlias = str
737"""Type alias for a column reference."""
740class ColumnGroup(BaseObject):
741 """Column group model."""
743 columns: list[ColumnRef | Column] = Field(..., min_length=1)
744 """Columns in the group."""
746 ivoa_ucd: str | None = Field(None, alias="ivoa:ucd")
747 """IVOA UCD of the column."""
749 table: Table | None = Field(None, exclude=True)
750 """Reference to the parent table."""
752 @field_validator("ivoa_ucd")
753 @classmethod
754 def check_ivoa_ucd(cls, ivoa_ucd: str) -> str:
755 """Check that IVOA UCD values are valid.
757 Parameters
758 ----------
759 ivoa_ucd
760 IVOA UCD value to check.
762 Returns
763 -------
764 `str`
765 The IVOA UCD value if it is valid.
766 """
767 return validate_ivoa_ucd(ivoa_ucd)
769 @model_validator(mode="after")
770 def check_unique_columns(self) -> ColumnGroup:
771 """Check that the columns list contains unique items.
773 Returns
774 -------
775 `ColumnGroup`
776 The column group being validated.
777 """
778 column_ids = [col if isinstance(col, str) else col.id for col in self.columns]
779 if len(column_ids) != len(set(column_ids)):
780 raise ValueError("Columns in the group must be unique")
781 return self
783 def _dereference_columns(self) -> None:
784 """Dereference ColumnRef to Column objects."""
785 if self.table is None:
786 raise ValueError("ColumnGroup must have a reference to its parent table")
788 dereferenced_columns: list[ColumnRef | Column] = []
789 for col in self.columns:
790 if isinstance(col, str):
791 # Dereference ColumnRef to Column object
792 try:
793 col_obj = self.table._find_column_by_id(col)
794 except KeyError as e:
795 raise ValueError(f"Column '{col}' not found in table '{self.table.name}'") from e
796 dereferenced_columns.append(col_obj)
797 else:
798 dereferenced_columns.append(col)
800 self.columns = dereferenced_columns
802 @field_serializer("columns")
803 def serialize_columns(self, columns: list[ColumnRef | Column]) -> list[str]:
804 """Serialize columns as their IDs.
806 Parameters
807 ----------
808 columns
809 The columns to serialize.
811 Returns
812 -------
813 `list` [ `str` ]
814 The serialized column IDs.
815 """
816 return [col if isinstance(col, str) else col.id for col in columns]
819class ColumnOverrides(BaseModel):
820 """Allowed overrides for a referenced column.
822 All fields are optional; missing values mean "inherit from the base
823 column".
824 """
826 datatype: DataType | None = None
827 """New datatype for the column."""
829 length: int | None = None
830 """New length for the column."""
832 description: str | None = None
833 """New description for the column."""
835 nullable: bool = True
836 """New nullable flag for the column."""
838 tap_principal: int | None = Field(default=None, alias="tap:principal")
839 """Override for the TAP_SCHEMA 'principal' flag."""
841 tap_column_index: int | None = Field(default=None, alias="tap:column_index")
842 """Override for the TAP_SCHEMA column index."""
844 @field_serializer("datatype")
845 def serialize_datatype(self, value: DataType | None) -> str | None:
846 """Convert `DataType` to string when serializing to JSON/YAML.
848 Parameters
849 ----------
850 value
851 The `DataType` value to serialize, or None.
853 Returns
854 -------
855 `str` | None
856 The serialized `DataType` value, or None if the input was None.
857 """
858 if value is None:
859 return None
860 return str(value)
862 @field_validator("datatype", mode="before")
863 @classmethod
864 def deserialize_datatype(cls, value: str | None) -> DataType | None:
865 """Convert string back into `DataType` when loading from JSON/YAML.
867 Parameters
868 ----------
869 value
870 The string value to deserialize, or None.
872 Returns
873 -------
874 `DataType` | None
875 The deserialized `DataType` value, or None if the input was None.
876 """
877 if value is None:
878 return None
879 return DataType(value)
882class ColumnResourceRef(BaseModel):
883 """A column which is dervived from an external resource."""
885 ref_name: str | None = None
886 """Name of the referenced column in the resource
887 (if different from the key)."""
889 overrides: ColumnOverrides | None = None
890 """Optional overrides of the referenced column's attributes."""
893# Type aliases for the nested mapping structure of referenced columns
894ResourceColumnMap: TypeAlias = dict[str, ColumnResourceRef | None]
895ResourceTableMap: TypeAlias = dict[str, ResourceColumnMap]
896ResourceMap: TypeAlias = dict[str, ResourceTableMap]
899class Table(BaseObject):
900 """Table model."""
902 primary_key: str | list[str] | None = Field(None, alias="primaryKey")
903 """Primary key of the table."""
905 tap_table_index: int | None = Field(None, alias="tap:table_index")
906 """IVOA TAP_SCHEMA table index of the table."""
908 mysql_engine: str | None = Field("MyISAM", alias="mysql:engine")
909 """MySQL engine to use for the table."""
911 mysql_charset: str | None = Field(None, alias="mysql:charset")
912 """MySQL charset to use for the table."""
914 columns: list[Column] = Field(default_factory=list)
915 """Columns in the table."""
917 column_refs: ResourceMap = Field(default_factory=dict, alias="columnRefs")
918 """Referenced columns from external resources."""
920 column_groups: list[ColumnGroup] = Field(default_factory=list, alias="columnGroups")
921 """Column groups in the table."""
923 constraints: list[_ConstraintType] = Field(default_factory=list)
924 """Constraints on the table."""
926 indexes: list[Index] = Field(default_factory=list)
927 """Indexes on the table."""
929 @field_validator("columns", mode="after")
930 @classmethod
931 def check_unique_column_names(cls, columns: list[Column]) -> list[Column]:
932 """Check that column names are unique.
934 Parameters
935 ----------
936 columns
937 The columns to check.
939 Returns
940 -------
941 `list` [ `Column` ]
942 The columns if they are unique.
944 Raises
945 ------
946 ValueError
947 Raised if column names are not unique.
948 """
949 if len(columns) != len(set(column.name for column in columns)):
950 raise ValueError("Column names must be unique")
951 return columns
953 @model_validator(mode="after")
954 def check_tap_table_index(self, info: ValidationInfo) -> Table:
955 """Check that the table has a TAP table index.
957 Parameters
958 ----------
959 info
960 Validation context used to determine if the check is enabled.
962 Returns
963 -------
964 `Table`
965 The table being validated.
967 Raises
968 ------
969 ValueError
970 Raised If the table is missing a TAP table index.
971 """
972 context = info.context
973 if not context or not context.get("check_tap_table_indexes", False):
974 return self
975 if self.tap_table_index is None:
976 raise ValueError("Table is missing a TAP table index")
977 return self
979 @model_validator(mode="after")
980 def check_tap_principal(self, info: ValidationInfo) -> Table:
981 """Check that at least one column is flagged as 'principal' for TAP
982 purposes.
984 Parameters
985 ----------
986 info
987 Validation context used to determine if the check is enabled.
989 Returns
990 -------
991 `Table`
992 The table being validated.
994 Raises
995 ------
996 ValueError
997 Raised if the table is missing a column flagged as 'principal'.
998 """
999 context = info.context
1000 if not context or not context.get("check_tap_principal", False):
1001 return self
1002 for col in self.columns:
1003 if col.tap_principal == 1:
1004 return self
1005 raise ValueError(f"Table '{self.name}' is missing at least one column designated as 'tap:principal'")
1007 def _find_column_by_id(self, id: str) -> Column:
1008 """Find a column by ID.
1010 Parameters
1011 ----------
1012 id
1013 The ID of the column to find.
1015 Returns
1016 -------
1017 `Column`
1018 The column with the given ID.
1020 Raises
1021 ------
1022 ValueError
1023 Raised if the column is not found.
1024 """
1025 for column in self.columns:
1026 if column.id == id:
1027 return column
1028 raise KeyError(f"Column '{id}' not found in table '{self.name}'")
1030 def _find_column_by_name(self, name: str) -> Column:
1031 for column in self.columns:
1032 if column.name == name:
1033 return column
1034 raise KeyError(f"Column '{name}' not found in table '{self.name}'")
1036 @model_validator(mode="after")
1037 def dereference_column_groups(self: Table) -> Table:
1038 """Dereference columns in column groups.
1040 Returns
1041 -------
1042 `Table`
1043 The table with dereferenced column groups.
1044 """
1045 for group in self.column_groups:
1046 group.table = self
1047 group._dereference_columns()
1048 return self
1050 @field_serializer("columns")
1051 def _serialize_columns(self, columns: list[Column]) -> list[dict[str, Any]]:
1052 """Serialize only non-resource columns."""
1053 return [
1054 col.model_dump(
1055 by_alias=True,
1056 exclude_none=True,
1057 exclude_defaults=True,
1058 )
1059 for col in columns
1060 if not col._is_resource_ref
1061 ]
1064class SchemaVersion(BaseModel):
1065 """Schema version model."""
1067 current: str
1068 """The current version of the schema."""
1070 compatible: list[str] = Field(default_factory=list)
1071 """The compatible versions of the schema."""
1073 read_compatible: list[str] = Field(default_factory=list)
1074 """The read compatible versions of the schema."""
1077class SchemaIdVisitor:
1078 """Visit a schema and build the map of IDs to objects.
1080 Notes
1081 -----
1082 Duplicates are added to a set when they are encountered, which can be
1083 accessed via the ``duplicates`` attribute. The presence of duplicates will
1084 not throw an error. Only the first object with a given ID will be added to
1085 the map, but this should not matter, since a ``ValidationError`` will be
1086 thrown by the ``model_validator`` method if any duplicates are found in the
1087 schema.
1088 """
1090 def __init__(self) -> None:
1091 """Create a new SchemaVisitor."""
1092 self.schema: Schema | None = None
1093 self.duplicates: set[str] = set()
1095 def add(self, obj: BaseObject) -> None:
1096 """Add an object to the ID map.
1098 Parameters
1099 ----------
1100 obj
1101 The object to add to the ID map.
1102 """
1103 if hasattr(obj, "id"):
1104 obj_id = getattr(obj, "id")
1105 if self.schema is not None:
1106 if obj_id in self.schema._id_map:
1107 self.duplicates.add(obj_id)
1108 else:
1109 self.schema._id_map[obj_id] = obj
1111 def visit_schema(self, schema: Schema) -> None:
1112 """Visit the objects in a schema and build the ID map.
1114 Parameters
1115 ----------
1116 schema
1117 The schema object to visit.
1119 Notes
1120 -----
1121 This will set an internal variable pointing to the schema object.
1122 """
1123 self.schema = schema
1124 self.duplicates.clear()
1125 self.add(self.schema)
1126 for table in self.schema.tables:
1127 self.visit_table(table)
1129 def visit_table(self, table: Table) -> None:
1130 """Visit a table object.
1132 Parameters
1133 ----------
1134 table
1135 The table object to visit.
1136 """
1137 self.add(table)
1138 for column in table.columns:
1139 self.visit_column(column)
1140 for constraint in table.constraints:
1141 self.visit_constraint(constraint)
1143 def visit_column(self, column: Column) -> None:
1144 """Visit a column object.
1146 Parameters
1147 ----------
1148 column
1149 The column object to visit.
1150 """
1151 self.add(column)
1153 def visit_constraint(self, constraint: Constraint) -> None:
1154 """Visit a constraint object.
1156 Parameters
1157 ----------
1158 constraint
1159 The constraint object to visit.
1160 """
1161 self.add(constraint)
1164T = TypeVar("T", bound=BaseObject)
1167def _strip_ids(data: Any) -> Any:
1168 """Recursively strip '@id' fields from a dictionary or list.
1170 Parameters
1171 ----------
1172 data
1173 The data to strip IDs from, which can be a dictionary, list, or any
1174 other type. Other types will be returned unchanged.
1175 """
1176 if isinstance(data, dict):
1177 data.pop("@id", None)
1178 for k, v in data.items():
1179 data[k] = _strip_ids(v)
1180 return data
1181 elif isinstance(data, list):
1182 return [_strip_ids(item) for item in data]
1183 else:
1184 return data
1187def _append_error(
1188 errors: list[InitErrorDetails],
1189 loc: tuple,
1190 input_value: Any,
1191 error_message: str,
1192 error_type: str = "value_error",
1193) -> None:
1194 """Append an error to the errors list.
1196 Parameters
1197 ----------
1198 errors : list[InitErrorDetails]
1199 The list of errors to append to.
1200 loc : tuple
1201 The location of the error in the schema.
1202 input_value : Any
1203 The input value that caused the error.
1204 error_message : str
1205 The error message to include in the context.
1206 """
1207 errors.append(
1208 {
1209 "type": error_type,
1210 "loc": loc,
1211 "input": input_value,
1212 "ctx": {"error": error_message},
1213 }
1214 )
1217class Resource(BaseModel):
1218 """A resource definition referencing an external schema."""
1220 uri: str = Field(..., description="Resource URI or path")
1221 """URI of the schema resource which may be a local path, ``resource://``,
1222 or remote URL."""
1225class Schema(BaseObject, Generic[T]):
1226 """Database schema model.
1228 This represents a database schema, which contains one or more tables.
1229 """
1231 version: SchemaVersion | str | None = None
1232 """The version of the schema."""
1234 tables: Sequence[Table]
1235 """The tables in the schema."""
1237 resources: dict[str, Resource] = Field(default_factory=dict)
1238 """External resources referenced by this schema."""
1240 _id_map: dict[str, Any] = PrivateAttr(default_factory=dict)
1241 """Map of IDs to objects."""
1243 _resource_map: dict[str, Schema] = PrivateAttr(default_factory=dict)
1244 """Map of resource names to loaded schemas."""
1246 @model_validator(mode="after")
1247 def _load_resources(self: Schema, info: ValidationInfo) -> Schema:
1248 """Load external resources referenced by this schema into an internal
1249 mapping of resource names to their `Schema` objects.
1251 Returns
1252 -------
1253 `Schema`
1254 The schema being validated.
1256 Raises
1257 ------
1258 ValueError
1259 Raised if a resource cannot be loaded.
1260 """
1261 if info.context:
1262 context = info.context.copy()
1263 # Ignore this flag for loading the resources themselves
1264 context.pop("dereference_resources", None)
1265 else:
1266 context = {}
1268 for resource_name, resource in self.resources.items():
1269 uri = resource.uri
1270 try:
1271 loaded_schema = Schema.from_uri(uri, context=context)
1272 self._resource_map[resource_name] = loaded_schema
1273 logger.debug(f"Loaded resource '{resource_name}' from URI '{uri}'")
1274 except Exception as e:
1275 raise ValueError(f"Failed to load resource '{resource_name}' from URI '{uri}': {e}") from e
1276 return self
1278 def _find_table_by_name(self, name: str) -> Table:
1279 """Find a table by name.
1281 Parameters
1282 ----------
1283 name
1284 The name of the table to find.
1286 Returns
1287 -------
1288 `Table`
1289 The table with the given name.
1291 Raises
1292 ------
1293 KeyError
1294 Raised if the table is not found.
1295 """
1296 for table in self.tables:
1297 if table.name == name:
1298 return table
1299 raise KeyError(f"Table '{name}' not found in schema '{self.name}'")
1301 @model_validator(mode="after")
1302 def _dereference_resource_columns(self: Schema, info: ValidationInfo) -> Schema:
1303 """Dereference columns from external resources and add them to the
1304 tables in this schema.
1305 """
1306 context = info.context
1307 column_ref_index_increment: int | None = None
1308 dereference_resources = False
1309 if context is not None:
1310 dereference_resources = context.get("dereference_resources", False)
1311 column_ref_index_increment = context.get("column_ref_index_increment", None)
1313 for table in self.tables:
1314 if column_refs := table.column_refs:
1315 for resource_name, tables in column_refs.items():
1316 resource_schema = self._resource_map.get(resource_name)
1317 if resource_schema is None:
1318 raise ValueError(f"Schema resource '{resource_name}' was not found in resources")
1319 self._process_column_refs(
1320 table,
1321 tables,
1322 resource_schema,
1323 dereference_resources,
1324 column_ref_index_increment,
1325 )
1326 if dereference_resources and len(table.column_refs) > 0:
1327 # Clear column refs in table if fully dereferencing
1328 logger.debug(
1329 f"Clearing columnRefs in table '{table.name}' after dereferencing resource columns"
1330 )
1331 table.column_refs = {}
1332 return self
1334 @classmethod
1335 def _copy_overrides_to_column(
1336 cls, column_ref: ColumnResourceRef, column_copy: Column
1337 ) -> ColumnOverrides | None:
1338 """Copy overrides from a column ref to a column."""
1339 if column_ref.overrides is not None:
1340 overrides = column_ref.overrides
1341 override_fields = overrides.model_fields_set
1342 for field_name in override_fields:
1343 if hasattr(column_copy, field_name):
1344 # Use attribute assignment to avoid type conversion issues
1345 # which can occur with using model_dump and model_copy
1346 setattr(column_copy, field_name, getattr(overrides, field_name))
1347 return column_ref.overrides
1349 @classmethod
1350 def _process_column_refs(
1351 cls,
1352 table: Table,
1353 ref_tables: ResourceTableMap,
1354 resource_schema: Schema,
1355 dereference_resources: bool = False,
1356 column_ref_index_increment: int | None = None,
1357 ) -> None:
1358 """Process column references from an external resource and add them
1359 to the given table as columns.
1360 """
1361 current_column_index = column_ref_index_increment if column_ref_index_increment is not None else -1
1363 for table_name, columns in ref_tables.items():
1364 try:
1365 resource_table = resource_schema._find_table_by_name(table_name)
1366 except KeyError as e:
1367 raise ValueError(
1368 f"Table '{table_name}' not found in resource '{resource_schema.name}'"
1369 ) from e
1370 for local_column_name, column_ref in columns.items():
1371 if column_ref is not None and column_ref.ref_name is not None:
1372 # Use specified ref_name
1373 ref_column_name = column_ref.ref_name
1374 else:
1375 # Use the local column name if no ref_name
1376 # specified
1377 ref_column_name = local_column_name
1379 # Check if referenced column exists in resource
1380 try:
1381 base_column = resource_table._find_column_by_name(ref_column_name)
1382 except KeyError:
1383 # The ref_name is specified but column is not
1384 # found
1385 if column_ref is not None and column_ref.ref_name is not None:
1386 raise ValueError(
1387 f"Column '{ref_column_name}' not found in table '{table_name}' "
1388 f"from resource '{resource_schema.name}'"
1389 )
1390 # The ref_name is not specified and the local
1391 # column name is not found
1392 raise ValueError(
1393 f"Column '{local_column_name}' not found in table '{table_name}' "
1394 f"from resource '{resource_schema.name}' and no ref_name provided"
1395 )
1397 # Create a copy of the base column and apply
1398 # overrides
1399 column_copy = base_column.model_copy()
1401 # Set the local name (key from the mapping)
1402 column_copy.name = local_column_name
1404 if not dereference_resources:
1405 # Flag the column as a resource reference so it will not be
1406 # written out during serialization
1407 column_copy._is_resource_ref = True
1409 # Apply overrides to the original column definition
1410 overrides: ColumnOverrides | None = None
1411 if column_ref is not None:
1412 overrides = cls._copy_overrides_to_column(column_ref, column_copy)
1414 # Manually set the ID of the copied column as ID generation has
1415 # already occurred by now
1416 column_copy.id = f"{table.id}.{local_column_name}"
1418 # Apply automatic assignment of 'tap:column_index', if enabled
1419 if column_ref_index_increment is not None:
1420 if (not overrides) or (overrides.tap_column_index is None):
1421 column_copy.tap_column_index = current_column_index
1422 current_column_index += column_ref_index_increment
1423 logger.debug(
1424 f"Automatically assigned 'tap:column_index' {column_copy.tap_column_index} to "
1425 f"column '{local_column_name}' in table '{table_name}' from resource "
1426 f"'{resource_schema.name}'"
1427 )
1428 else:
1429 # Skip automatic assignment of 'tap:column_index' if it
1430 # is already overridden
1431 logger.debug(
1432 f"Skipping automatic assignment of 'tap:column_index' for column "
1433 f"'{local_column_name}' in table '{table_name}' from resource "
1434 f"'{resource_schema.name}' as it is already overridden to "
1435 f"{column_copy.tap_column_index}"
1436 )
1437 table.columns.append(column_copy)
1438 logger.debug(
1439 f"Dereferenced column '{local_column_name}' from table '{table_name}' "
1440 f"in resource '{resource_schema.name}' into table '{table.name}'"
1441 )
1443 @model_validator(mode="before")
1444 @classmethod
1445 def generate_ids(cls, values: dict[str, Any], info: ValidationInfo) -> dict[str, Any]:
1446 """Generate IDs for objects that do not have them.
1448 Parameters
1449 ----------
1450 values
1451 The values of the schema.
1452 info
1453 Validation context used to determine if ID generation is enabled.
1455 Returns
1456 -------
1457 `dict` [ `str`, `Any` ]
1458 The values of the schema with generated IDs.
1459 """
1460 context = info.context
1461 if not context or not context.get("id_generation", False):
1462 logger.debug("Skipping ID generation")
1463 return values
1464 schema_name = values["name"]
1465 if "@id" not in values:
1466 values["@id"] = f"#{schema_name}"
1467 logger.debug(f"Generated ID '{values['@id']}' for schema '{schema_name}'")
1468 if "tables" in values:
1469 for table in values["tables"]:
1470 if "@id" not in table:
1471 table["@id"] = f"#{table['name']}"
1472 logger.debug(f"Generated ID '{table['@id']}' for table '{table['name']}'")
1473 if "columns" in table:
1474 for column in table["columns"]:
1475 if "@id" not in column:
1476 column["@id"] = f"#{table['name']}.{column['name']}"
1477 logger.debug(f"Generated ID '{column['@id']}' for column '{column['name']}'")
1478 if "columnGroups" in table:
1479 for column_group in table["columnGroups"]:
1480 if "@id" not in column_group:
1481 column_group["@id"] = f"#{table['name']}.{column_group['name']}"
1482 logger.debug(
1483 f"Generated ID '{column_group['@id']}' for column group "
1484 f"'{column_group['name']}'"
1485 )
1486 if "constraints" in table:
1487 for constraint in table["constraints"]:
1488 if "@id" not in constraint:
1489 constraint["@id"] = f"#{constraint['name']}"
1490 logger.debug(
1491 f"Generated ID '{constraint['@id']}' for constraint '{constraint['name']}'"
1492 )
1493 if "indexes" in table:
1494 for index in table["indexes"]:
1495 if "@id" not in index:
1496 index["@id"] = f"#{index['name']}"
1497 logger.debug(f"Generated ID '{index['@id']}' for index '{index['name']}'")
1498 return values
1500 @field_validator("tables", mode="after")
1501 @classmethod
1502 def check_unique_table_names(cls, tables: list[Table]) -> list[Table]:
1503 """Check that table names are unique.
1505 Parameters
1506 ----------
1507 tables
1508 The tables to check.
1510 Returns
1511 -------
1512 `list` [ `Table` ]
1513 The tables if they are unique.
1515 Raises
1516 ------
1517 ValueError
1518 Raised if table names are not unique.
1519 """
1520 if len(tables) != len(set(table.name for table in tables)):
1521 raise ValueError("Table names must be unique")
1522 return tables
1524 @model_validator(mode="after")
1525 def check_tap_table_indexes(self, info: ValidationInfo) -> Schema:
1526 """Check that the TAP table indexes are unique.
1528 Parameters
1529 ----------
1530 info
1531 The validation context used to determine if the check is enabled.
1533 Returns
1534 -------
1535 `Schema`
1536 The schema being validated.
1537 """
1538 context = info.context
1539 if not context or not context.get("check_tap_table_indexes", False):
1540 return self
1541 table_indicies = set()
1542 for table in self.tables:
1543 table_index = table.tap_table_index
1544 if table_index is not None:
1545 if table_index in table_indicies:
1546 raise ValueError(f"Duplicate 'tap:table_index' value {table_index} found in schema")
1547 table_indicies.add(table_index)
1548 return self
1550 @model_validator(mode="after")
1551 def check_unique_constraint_names(self: Schema) -> Schema:
1552 """Check for duplicate constraint names in the schema.
1554 Returns
1555 -------
1556 `Schema`
1557 The schema being validated.
1559 Raises
1560 ------
1561 ValueError
1562 Raised if duplicate constraint names are found in the schema.
1563 """
1564 constraint_names = set()
1565 duplicate_names = []
1567 for table in self.tables:
1568 for constraint in table.constraints:
1569 constraint_name = constraint.name
1570 if constraint_name in constraint_names:
1571 duplicate_names.append(constraint_name)
1572 else:
1573 constraint_names.add(constraint_name)
1575 if duplicate_names:
1576 raise ValueError(f"Duplicate constraint names found in schema: {duplicate_names}")
1578 return self
1580 @model_validator(mode="after")
1581 def check_unique_index_names(self: Schema) -> Schema:
1582 """Check for duplicate index names in the schema.
1584 Returns
1585 -------
1586 `Schema`
1587 The schema being validated.
1589 Raises
1590 ------
1591 ValueError
1592 Raised if duplicate index names are found in the schema.
1593 """
1594 index_names = set()
1595 duplicate_names = []
1597 for table in self.tables:
1598 for index in table.indexes:
1599 index_name = index.name
1600 if index_name in index_names:
1601 duplicate_names.append(index_name)
1602 else:
1603 index_names.add(index_name)
1605 if duplicate_names:
1606 raise ValueError(f"Duplicate index names found in schema: {duplicate_names}")
1608 return self
1610 @model_validator(mode="after")
1611 def create_id_map(self: Schema) -> Schema:
1612 """Create a map of IDs to objects.
1614 Returns
1615 -------
1616 `Schema`
1617 The schema with the ID map created.
1619 Raises
1620 ------
1621 ValueError
1622 Raised if duplicate identifiers are found in the schema.
1623 """
1624 if self._id_map:
1625 logger.debug("Ignoring call to create_id_map() - ID map was already populated")
1626 return self
1627 visitor: SchemaIdVisitor = SchemaIdVisitor()
1628 visitor.visit_schema(self)
1629 if len(visitor.duplicates):
1630 raise ValueError(
1631 "Duplicate IDs found in schema:\n " + "\n ".join(visitor.duplicates) + "\n"
1632 )
1633 logger.debug("Created ID map with %d entries", len(self._id_map))
1634 return self
1636 def _validate_column_id(
1637 self: Schema,
1638 column_id: str,
1639 loc: tuple,
1640 errors: list[InitErrorDetails],
1641 ) -> None:
1642 """Validate a column ID from a constraint and append errors if invalid.
1644 Parameters
1645 ----------
1646 schema : Schema
1647 The schema being validated.
1648 column_id : str
1649 The column ID to validate.
1650 loc : tuple
1651 The location of the error in the schema.
1652 errors : list[InitErrorDetails]
1653 The list of errors to append to.
1654 """
1655 if column_id not in self:
1656 _append_error(
1657 errors,
1658 loc,
1659 column_id,
1660 f"Column ID '{column_id}' not found in schema",
1661 )
1662 elif not isinstance(self[column_id], Column):
1663 _append_error(
1664 errors,
1665 loc,
1666 column_id,
1667 f"ID '{column_id}' does not refer to a Column object",
1668 )
1670 def _validate_foreign_key_column(
1671 self: Schema,
1672 column_id: str,
1673 table: Table,
1674 loc: tuple,
1675 errors: list[InitErrorDetails],
1676 ) -> None:
1677 """Validate a foreign key column ID from a constraint and append errors
1678 if invalid.
1680 Parameters
1681 ----------
1682 schema : Schema
1683 The schema being validated.
1684 column_id : str
1685 The foreign key column ID to validate.
1686 loc : tuple
1687 The location of the error in the schema.
1688 errors : list[InitErrorDetails]
1689 The list of errors to append to.
1690 """
1691 try:
1692 table._find_column_by_id(column_id)
1693 except KeyError:
1694 _append_error(
1695 errors,
1696 loc,
1697 column_id,
1698 f"Column '{column_id}' not found in table '{table.name}'",
1699 )
1701 @model_validator(mode="after")
1702 def check_constraints(self: Schema) -> Schema:
1703 """Check constraint objects for validity. This needs to be deferred
1704 until after the schema is fully loaded and the ID map is created.
1706 Raises
1707 ------
1708 pydantic.ValidationError
1709 Raised if any constraints are invalid.
1711 Returns
1712 -------
1713 `Schema`
1714 The schema being validated.
1715 """
1716 errors: list[InitErrorDetails] = []
1718 for table_index, table in enumerate(self.tables):
1719 for constraint_index, constraint in enumerate(table.constraints):
1720 column_ids: list[str] = []
1721 referenced_column_ids: list[str] = []
1723 if isinstance(constraint, ForeignKeyConstraint):
1724 column_ids += constraint.columns
1725 referenced_column_ids += constraint.referenced_columns
1726 elif isinstance(constraint, UniqueConstraint):
1727 column_ids += constraint.columns
1728 # No extra checks are required on CheckConstraint objects.
1730 # Validate the foreign key columns
1731 for column_id in column_ids:
1732 self._validate_column_id(
1733 column_id,
1734 (
1735 "tables",
1736 table_index,
1737 "constraints",
1738 constraint_index,
1739 "columns",
1740 column_id,
1741 ),
1742 errors,
1743 )
1744 # Check that the foreign key column is within the source
1745 # table.
1746 self._validate_foreign_key_column(
1747 column_id,
1748 table,
1749 (
1750 "tables",
1751 table_index,
1752 "constraints",
1753 constraint_index,
1754 "columns",
1755 column_id,
1756 ),
1757 errors,
1758 )
1760 # Validate the primary key (reference) columns
1761 for referenced_column_id in referenced_column_ids:
1762 self._validate_column_id(
1763 referenced_column_id,
1764 (
1765 "tables",
1766 table_index,
1767 "constraints",
1768 constraint_index,
1769 "referenced_columns",
1770 referenced_column_id,
1771 ),
1772 errors,
1773 )
1775 if errors:
1776 raise ValidationError.from_exception_data("Schema validation failed", errors)
1778 return self
1780 def __getitem__(self, id: str) -> BaseObject:
1781 """Get an object by its ID.
1783 Parameters
1784 ----------
1785 id
1786 The ID of the object to get.
1788 Raises
1789 ------
1790 KeyError
1791 Raised if the object with the given ID is not found in the schema.
1792 """
1793 if id not in self:
1794 raise KeyError(f"Object with ID '{id}' not found in schema")
1795 return self._id_map[id]
1797 def __contains__(self, id: str) -> bool:
1798 """Check if an object with the given ID is in the schema.
1800 Parameters
1801 ----------
1802 id
1803 The ID of the object to check.
1804 """
1805 return id in self._id_map
1807 def find_object_by_id(self, id: str, obj_type: type[T]) -> T:
1808 """Find an object with the given type by its ID.
1810 Parameters
1811 ----------
1812 id
1813 The ID of the object to find.
1814 obj_type
1815 The type of the object to find.
1817 Returns
1818 -------
1819 BaseObject
1820 The object with the given ID and type.
1822 Raises
1823 ------
1824 KeyError
1825 If the object with the given ID is not found in the schema.
1826 TypeError
1827 If the object that is found does not have the right type.
1829 Notes
1830 -----
1831 The actual return type is the user-specified argument ``T``, which is
1832 expected to be a subclass of `BaseObject`.
1833 """
1834 obj = self[id]
1835 if not isinstance(obj, obj_type):
1836 raise TypeError(f"Object with ID '{id}' is not of type '{obj_type.__name__}'")
1837 return obj
1839 def get_table_by_column(self, column: Column) -> Table:
1840 """Find the table that contains a column.
1842 Parameters
1843 ----------
1844 column
1845 The column to find.
1847 Returns
1848 -------
1849 `Table`
1850 The table that contains the column.
1852 Raises
1853 ------
1854 ValueError
1855 If the column is not found in any table.
1856 """
1857 for table in self.tables:
1858 if column in table.columns:
1859 return table
1860 raise ValueError(f"Column '{column.name}' not found in any table")
1862 @classmethod
1863 def from_uri(cls, resource_path: ResourcePathExpression, context: dict[str, Any] = {}) -> Schema:
1864 """Load a `Schema` from a string representing a ``ResourcePath``.
1866 Parameters
1867 ----------
1868 resource_path
1869 The ``ResourcePath`` pointing to a YAML file.
1870 context
1871 Pydantic context to be used in validation.
1873 Returns
1874 -------
1875 `str`
1876 The ID of the object.
1878 Raises
1879 ------
1880 yaml.YAMLError
1881 Raised if there is an error loading the YAML data.
1882 ValueError
1883 Raised if there is an error reading the resource.
1884 pydantic.ValidationError
1885 Raised if the schema fails validation.
1886 """
1887 try:
1888 rp_stream = ResourcePath(resource_path).read()
1889 except Exception as e:
1890 raise ValueError(f"Error reading resource from '{resource_path}' : {e}") from e
1891 yaml_data = yaml.safe_load(rp_stream)
1892 return Schema.model_validate(yaml_data, context=context)
1894 @classmethod
1895 def from_stream(cls, source: IO[str], context: dict[str, Any] = {}) -> Schema:
1896 """Load a `Schema` from a file stream which should contain YAML data.
1898 Parameters
1899 ----------
1900 source
1901 The file stream to read from.
1902 context
1903 Pydantic context to be used in validation.
1905 Returns
1906 -------
1907 `Schema`
1908 The Felis schema loaded from the stream.
1910 Raises
1911 ------
1912 yaml.YAMLError
1913 Raised if there is an error loading the YAML file.
1914 pydantic.ValidationError
1915 Raised if the schema fails validation.
1916 """
1917 logger.debug("Loading schema from: '%s'", source)
1918 yaml_data = yaml.safe_load(source)
1919 return Schema.model_validate(yaml_data, context=context)
1921 def _model_dump(self, strip_ids: bool = False) -> dict[str, Any]:
1922 """Dump the schema as a dictionary with some default arguments
1923 applied.
1925 Parameters
1926 ----------
1927 strip_ids
1928 Whether to strip the IDs from the dumped data. Defaults to `False`.
1930 Returns
1931 -------
1932 `dict` [ `str`, `Any` ]
1933 The dumped schema data as a dictionary.
1934 """
1935 data = self.model_dump(by_alias=True, exclude_none=True, exclude_defaults=True)
1936 if strip_ids:
1937 data = _strip_ids(data)
1938 return data
1940 def dump_yaml(self, stream: IO[str] = sys.stdout, strip_ids: bool = False) -> None:
1941 """Pretty print the schema as YAML.
1943 Parameters
1944 ----------
1945 stream
1946 The stream to write the YAML data to.
1947 strip_ids
1948 Whether to strip the IDs from the dumped data. Defaults to `False`.
1949 """
1950 data = self._model_dump(strip_ids=strip_ids)
1951 yaml.safe_dump(
1952 data,
1953 stream,
1954 default_flow_style=False,
1955 sort_keys=False,
1956 )
1958 def dump_json(self, stream: IO[str] = sys.stdout, strip_ids: bool = False) -> None:
1959 """Pretty print the schema as JSON.
1961 Parameters
1962 ----------
1963 stream
1964 The stream to write the JSON data to.
1965 strip_ids
1966 Whether to strip the IDs from the dumped data. Defaults to `False`.
1967 """
1968 data = self._model_dump(strip_ids=strip_ids)
1969 json.dump(
1970 data,
1971 stream,
1972 indent=4,
1973 sort_keys=False,
1974 )