Coverage for python/felis/datamodel.py: 98%
732 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-11-04 01:32 -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 operator import itemgetter
32from typing import IO, Annotated, Any, Generic, Literal, TypeAlias, TypeVar
34import yaml
35from astropy import units as units # type: ignore
36from astropy.io.votable import ucd # type: ignore
37from lsst.resources import ResourcePath, ResourcePathExpression
38from pydantic import (
39 BaseModel,
40 ConfigDict,
41 Field,
42 PrivateAttr,
43 ValidationError,
44 ValidationInfo,
45 field_serializer,
46 field_validator,
47 model_validator,
48)
49from pydantic_core import InitErrorDetails
51from .db._dialects import get_supported_dialects, string_to_typeengine
52from .db._sqltypes import get_type_func
53from .types import Boolean, Byte, Char, Double, FelisType, Float, Int, Long, Short, String, Text, Unicode
55logger = logging.getLogger(__name__)
57__all__ = (
58 "BaseObject",
59 "CheckConstraint",
60 "Column",
61 "ColumnGroup",
62 "ColumnOverrides",
63 "ColumnResourceRef",
64 "Constraint",
65 "DataType",
66 "ForeignKeyConstraint",
67 "Index",
68 "Resource",
69 "Schema",
70 "SchemaVersion",
71 "Table",
72 "UniqueConstraint",
73)
75CONFIG = ConfigDict(
76 populate_by_name=True, # Populate attributes by name.
77 extra="forbid", # Do not allow extra fields.
78 str_strip_whitespace=True, # Strip whitespace from string fields.
79 use_enum_values=False, # Do not use enum values during serialization.
80)
81"""Pydantic model configuration as described in:
82https://docs.pydantic.dev/2.0/api/config/#pydantic.config.ConfigDict
83"""
86class BaseObject(BaseModel):
87 """Base model.
89 All classes representing objects in the Felis data model should inherit
90 from this class.
91 """
93 model_config = CONFIG
94 """Pydantic model configuration."""
96 name: str
97 """Name of the database object."""
99 id: str = Field(alias="@id")
100 """Unique identifier of the database object."""
102 description: str | None = None
103 """Description of the database object."""
105 votable_utype: str | None = Field(None, alias="votable:utype")
106 """VOTable utype (usage-specific or unique type) of the object."""
108 @model_validator(mode="after")
109 def check_description(self, info: ValidationInfo) -> BaseObject:
110 """Check that the description is present and contains at least one
111 non-whitespace character, if the validation context indicates that this
112 check is enabled. By default, descriptions may be omitted or empty.
114 Parameters
115 ----------
116 info
117 Validation context used to determine if the check is enabled.
119 Returns
120 -------
121 `BaseObject`
122 The object being validated.
123 """
124 context = info.context
125 if not context or not context.get("check_description", False):
126 return self
127 if self.description is None or self.description == "":
128 raise ValueError("Description is required and must be non-empty")
129 return self
132class DataType(StrEnum):
133 """``Enum`` representing the data types supported by Felis."""
135 boolean = auto()
136 byte = auto()
137 short = auto()
138 int = auto()
139 long = auto()
140 float = auto()
141 double = auto()
142 char = auto()
143 string = auto()
144 unicode = auto()
145 text = auto()
146 binary = auto()
147 timestamp = auto()
150def validate_ivoa_ucd(ivoa_ucd: str) -> str:
151 """Validate IVOA UCD values.
153 Parameters
154 ----------
155 ivoa_ucd
156 IVOA UCD value to check.
158 Returns
159 -------
160 `str`
161 The IVOA UCD value if it is valid.
163 Raises
164 ------
165 ValueError
166 If the IVOA UCD value is invalid.
167 """
168 if ivoa_ucd is not None: 168 ↛ 173line 168 didn't jump to line 173 because the condition on line 168 was always true
169 try:
170 ucd.parse_ucd(ivoa_ucd, check_controlled_vocabulary=True, has_colon=";" in ivoa_ucd)
171 except ValueError as e:
172 raise ValueError(f"Invalid IVOA UCD: {e}")
173 return ivoa_ucd
176class Column(BaseObject):
177 """Column model."""
179 datatype: DataType
180 """Datatype of the column."""
182 length: int | None = Field(None, gt=0)
183 """Length of the column."""
185 precision: int | None = Field(None, ge=0)
186 """The numerical precision of the column.
188 For timestamps, this is the number of fractional digits retained in the
189 seconds field.
190 """
192 nullable: bool = True
193 """Whether the column can be ``NULL``."""
195 value: str | int | float | bool | None = None
196 """Default value of the column."""
198 autoincrement: bool | None = None
199 """Whether the column is autoincremented."""
201 ivoa_ucd: str | None = Field(None, alias="ivoa:ucd")
202 """IVOA UCD of the column."""
204 fits_tunit: str | None = Field(None, alias="fits:tunit")
205 """FITS TUNIT of the column."""
207 ivoa_unit: str | None = Field(None, alias="ivoa:unit")
208 """IVOA unit of the column."""
210 tap_column_index: int | None = Field(None, alias="tap:column_index")
211 """TAP_SCHEMA column index of the column."""
213 tap_principal: int | None = Field(0, alias="tap:principal", ge=0, le=1)
214 """Whether this is a TAP_SCHEMA principal column."""
216 votable_arraysize: int | str | None = Field(None, alias="votable:arraysize")
217 """VOTable arraysize of the column."""
219 tap_std: int | None = Field(0, alias="tap:std", ge=0, le=1)
220 """TAP_SCHEMA indication that this column is defined by an IVOA standard.
221 """
223 votable_xtype: str | None = Field(None, alias="votable:xtype")
224 """VOTable xtype (extended type) of the column."""
226 votable_datatype: str | None = Field(None, alias="votable:datatype")
227 """VOTable datatype of the column."""
229 mysql_datatype: str | None = Field(None, alias="mysql:datatype")
230 """MySQL datatype override on the column."""
232 postgresql_datatype: str | None = Field(None, alias="postgresql:datatype")
233 """PostgreSQL datatype override on the column."""
235 _is_resource_ref: bool = PrivateAttr(False)
236 """Whether this column is a resource reference column."""
238 @model_validator(mode="after")
239 def check_value(self) -> Column:
240 """Check that the default value is valid.
242 Returns
243 -------
244 `Column`
245 The column being validated.
246 """
247 if (value := self.value) is not None:
248 if value is not None and self.autoincrement is True:
249 raise ValueError("Column cannot have both a default value and be autoincremented")
250 felis_type = FelisType.felis_type(self.datatype)
251 if felis_type.is_numeric:
252 if felis_type in (Byte, Short, Int, Long) and not isinstance(value, int):
253 raise ValueError("Default value must be an int for integer type columns")
254 elif felis_type in (Float, Double) and not isinstance(value, float):
255 raise ValueError("Default value must be a decimal number for float and double columns")
256 elif felis_type in (String, Char, Unicode, Text):
257 if not isinstance(value, str):
258 raise ValueError("Default value must be a string for string columns")
259 if not len(value):
260 raise ValueError("Default value must be a non-empty string for string columns")
261 elif felis_type is Boolean and not isinstance(value, bool):
262 raise ValueError("Default value must be a boolean for boolean columns")
263 return self
265 @field_validator("ivoa_ucd")
266 @classmethod
267 def check_ivoa_ucd(cls, ivoa_ucd: str) -> str:
268 """Check that IVOA UCD values are valid.
270 Parameters
271 ----------
272 ivoa_ucd
273 IVOA UCD value to check.
275 Returns
276 -------
277 `str`
278 The IVOA UCD value if it is valid.
279 """
280 return validate_ivoa_ucd(ivoa_ucd)
282 @model_validator(mode="after")
283 def check_units(self) -> Column:
284 """Check that the ``fits:tunit`` or ``ivoa:unit`` field has valid
285 units according to astropy. Only one may be provided.
287 Returns
288 -------
289 `Column`
290 The column being validated.
292 Raises
293 ------
294 ValueError
295 Raised if both FITS and IVOA units are provided, or if the unit is
296 invalid.
297 """
298 fits_unit = self.fits_tunit
299 ivoa_unit = self.ivoa_unit
301 if fits_unit and ivoa_unit:
302 raise ValueError("Column cannot have both FITS and IVOA units")
303 unit = fits_unit or ivoa_unit
305 if unit is not None:
306 try:
307 units.Unit(unit)
308 except ValueError as e:
309 raise ValueError(f"Invalid unit: {e}")
311 return self
313 @model_validator(mode="before")
314 @classmethod
315 def check_length(cls, values: dict[str, Any]) -> dict[str, Any]:
316 """Check that a valid length is provided for sized types.
318 Parameters
319 ----------
320 values
321 Values of the column.
323 Returns
324 -------
325 `dict` [ `str`, `Any` ]
326 The values of the column.
328 Raises
329 ------
330 ValueError
331 Raised if a length is not provided for a sized type.
332 """
333 datatype = values.get("datatype")
334 if datatype is None:
335 # Skip this validation if datatype is not provided
336 return values
337 length = values.get("length")
338 felis_type = FelisType.felis_type(datatype)
339 if felis_type.is_sized and length is None:
340 raise ValueError(
341 f"Length must be provided for type '{datatype}'"
342 + (f" in column '{values['@id']}'" if "@id" in values else "")
343 )
344 elif not felis_type.is_sized and length is not None:
345 msg = f"The datatype '{datatype}' does not support a specified length"
346 if "@id" in values:
347 msg += f" in column '{values['@id']}'"
348 logger.warning("%s", msg)
349 return values
351 @model_validator(mode="after")
352 def check_redundant_datatypes(self, info: ValidationInfo) -> Column:
353 """Check for redundant datatypes on columns.
355 Parameters
356 ----------
357 info
358 Validation context used to determine if the check is enabled.
360 Returns
361 -------
362 `Column`
363 The column being validated.
365 Raises
366 ------
367 ValueError
368 Raised if a datatype override is redundant.
369 """
370 context = info.context
371 if not context or not context.get("check_redundant_datatypes", False):
372 return self
373 if all( 373 ↛ 377line 373 didn't jump to line 377 because the condition on line 373 was never true
374 getattr(self, f"{dialect}:datatype", None) is not None
375 for dialect in get_supported_dialects().keys()
376 ):
377 return self
379 datatype = self.datatype
380 length: int | None = self.length or None
382 datatype_func = get_type_func(datatype)
383 felis_type = FelisType.felis_type(datatype)
384 if felis_type.is_sized:
385 datatype_obj = datatype_func(length)
386 else:
387 datatype_obj = datatype_func()
389 for dialect_name, dialect in get_supported_dialects().items():
390 db_annotation = f"{dialect_name}_datatype"
391 if datatype_string := self.model_dump().get(db_annotation):
392 db_datatype_obj = string_to_typeengine(datatype_string, dialect, length)
393 if datatype_obj.compile(dialect) == db_datatype_obj.compile(dialect):
394 raise ValueError(
395 "'{}: {}' is a redundant override of 'datatype: {}' in column '{}'{}".format(
396 db_annotation,
397 datatype_string,
398 self.datatype,
399 self.id,
400 "" if length is None else f" with length {length}",
401 )
402 )
403 else:
404 logger.debug(
405 "Type override of 'datatype: %s' with '%s: %s' in column '%s' "
406 "compiled to '%s' and '%s'",
407 self.datatype,
408 db_annotation,
409 datatype_string,
410 self.id,
411 datatype_obj.compile(dialect),
412 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 "Forced VOTable's 'arraysize' to '*' on column '%s' with datatype "
469 "'%s' and length '%s'",
470 values["name"],
471 values["datatype"],
472 length,
473 )
474 else:
475 arraysize = f"{length}*"
476 elif datatype in ("timestamp", "text"):
477 arraysize = "*"
478 if arraysize is not None:
479 values["votable:arraysize"] = arraysize
480 logger.debug(
481 "Set default 'votable:arraysize' to '%s' on column '%s'"
482 " with datatype '%s' and length '%s'",
483 arraysize,
484 values["name"],
485 values["datatype"],
486 values.get("length", None),
487 )
488 else:
489 logger.debug(
490 "Using existing 'votable:arraysize' of '%s' on column '%s'", arraysize, values["name"]
491 )
492 if isinstance(values["votable:arraysize"], int): 492 ↛ 493line 492 didn't jump to line 493 because the condition on line 492 was never true
493 logger.warning(
494 "Usage of an integer value for 'votable:arraysize' in column '%s' is deprecated",
495 values["name"],
496 )
497 values["votable:arraysize"] = str(arraysize)
498 return values
500 @field_serializer("datatype")
501 def serialize_datatype(self, value: DataType) -> str:
502 """Convert `DataType` to string when serializing to JSON/YAML.
504 Parameters
505 ----------
506 value
507 The `DataType` value to serialize.
509 Returns
510 -------
511 `str`
512 The serialized `DataType` value.
513 """
514 return str(value)
516 @field_validator("datatype", mode="before")
517 @classmethod
518 def deserialize_datatype(cls, value: str) -> DataType:
519 """Convert string back into `DataType` when loading from JSON/YAML.
521 Parameters
522 ----------
523 value
524 The string value to deserialize.
526 Returns
527 -------
528 `DataType`
529 The deserialized `DataType` value.
530 """
531 return DataType(value)
533 @model_validator(mode="after")
534 def check_votable_xtype(self) -> Column:
535 """Set the default value for the ``votable_xtype`` field, which
536 corresponds to an Extended Datatype or ``xtype`` in the IVOA VOTable
537 standard.
539 Returns
540 -------
541 `Column`
542 The column being validated.
544 Notes
545 -----
546 This is currently only set automatically for the Felis ``timestamp``
547 datatype.
548 """
549 if self.datatype == DataType.timestamp and self.votable_xtype is None:
550 self.votable_xtype = "timestamp"
551 return self
553 def _update_from_overrides(self, overrides: ColumnOverrides) -> None:
554 """Update the column attributes from the given overrides.
556 Parameters
557 ----------
558 overrides
559 The column overrides to apply or `None` to skip applying overrides.
561 Notes
562 -----
563 Using ``model_fields_set`` allows updating only the fields that are
564 explicitly set in the `overrides` object. This prevents overwriting
565 existing column attributes which were not explicitly provided.
566 """
567 if overrides.model_fields_set:
568 logger.debug("Applying overrides to column '%s': %s", self.id, overrides.model_fields_set)
569 for field in overrides.model_fields_set:
570 setattr(self, field, getattr(overrides, field))
573class Constraint(BaseObject):
574 """Table constraint model."""
576 deferrable: bool = False
577 """Whether this constraint will be declared as deferrable."""
579 initially: Literal["IMMEDIATE", "DEFERRED"] | None = None
580 """Value for ``INITIALLY`` clause; only used if `deferrable` is
581 `True`."""
583 @model_validator(mode="after")
584 def check_deferrable(self) -> Constraint:
585 """Check that the ``INITIALLY`` clause is only used if `deferrable` is
586 `True`.
588 Returns
589 -------
590 `Constraint`
591 The constraint being validated.
592 """
593 if self.initially is not None and not self.deferrable:
594 raise ValueError("INITIALLY clause can only be used if deferrable is True")
595 return self
598class CheckConstraint(Constraint):
599 """Table check constraint model."""
601 type: Literal["Check"] = Field("Check", alias="@type")
602 """Type of the constraint."""
604 expression: str
605 """Expression for the check constraint."""
607 @field_serializer("type")
608 def serialize_type(self, value: str) -> str:
609 """Ensure '@type' is included in serialized output.
611 Parameters
612 ----------
613 value
614 The value to serialize.
616 Returns
617 -------
618 `str`
619 The serialized value.
620 """
621 return value
624class UniqueConstraint(Constraint):
625 """Table unique constraint model."""
627 type: Literal["Unique"] = Field("Unique", alias="@type")
628 """Type of the constraint."""
630 columns: list[str]
631 """Columns in the unique constraint."""
633 @field_serializer("type")
634 def serialize_type(self, value: str) -> str:
635 """Ensure '@type' is included in serialized output.
637 Parameters
638 ----------
639 value
640 The value to serialize.
642 Returns
643 -------
644 `str`
645 The serialized value.
646 """
647 return value
650class ForeignKeyConstraint(Constraint):
651 """Table foreign key constraint model.
653 This constraint is used to define a foreign key relationship between two
654 tables in the schema. There must be at least one column in the
655 `columns` list, and at least one column in the `referenced_columns` list
656 or a validation error will be raised.
658 Notes
659 -----
660 These relationships will be reflected in the TAP_SCHEMA ``keys`` and
661 ``key_columns`` data.
662 """
664 type: Literal["ForeignKey"] = Field("ForeignKey", alias="@type")
665 """Type of the constraint."""
667 columns: list[str] = Field(min_length=1)
668 """The columns comprising the foreign key."""
670 referenced_columns: list[str] = Field(alias="referencedColumns", min_length=1)
671 """The columns referenced by the foreign key."""
673 on_delete: Literal["CASCADE", "SET NULL", "SET DEFAULT", "RESTRICT", "NO ACTION"] | None = None
674 """Action to take when the referenced row is deleted."""
676 on_update: Literal["CASCADE", "SET NULL", "SET DEFAULT", "RESTRICT", "NO ACTION"] | None = None
677 """Action to take when the referenced row is updated."""
679 @field_serializer("type")
680 def serialize_type(self, value: str) -> str:
681 """Ensure '@type' is included in serialized output.
683 Parameters
684 ----------
685 value
686 The value to serialize.
688 Returns
689 -------
690 `str`
691 The serialized value.
692 """
693 return value
695 @model_validator(mode="after")
696 def check_column_lengths(self) -> ForeignKeyConstraint:
697 """Check that the `columns` and `referenced_columns` lists have the
698 same length.
700 Returns
701 -------
702 `ForeignKeyConstraint`
703 The foreign key constraint being validated.
705 Raises
706 ------
707 ValueError
708 Raised if the `columns` and `referenced_columns` lists do not have
709 the same length.
710 """
711 if len(self.columns) != len(self.referenced_columns):
712 raise ValueError(
713 "Columns and referencedColumns must have the same length for a ForeignKey constraint"
714 )
715 return self
718_ConstraintType = Annotated[
719 CheckConstraint | ForeignKeyConstraint | UniqueConstraint, Field(discriminator="type")
720]
721"""Type alias for a constraint type."""
724class Index(BaseObject):
725 """Table index model.
727 An index can be defined on either columns or expressions, but not both.
728 """
730 columns: list[str] | None = None
731 """Columns in the index."""
733 expressions: list[str] | None = None
734 """Expressions in the index."""
736 @model_validator(mode="before")
737 @classmethod
738 def check_columns_or_expressions(cls, values: dict[str, Any]) -> dict[str, Any]:
739 """Check that columns or expressions are specified, but not both.
741 Parameters
742 ----------
743 values
744 Values of the index.
746 Returns
747 -------
748 `dict` [ `str`, `Any` ]
749 The values of the index.
751 Raises
752 ------
753 ValueError
754 Raised if both columns and expressions are specified, or if neither
755 are specified.
756 """
757 if "columns" in values and "expressions" in values:
758 raise ValueError("Defining columns and expressions is not valid")
759 elif "columns" not in values and "expressions" not in values:
760 raise ValueError("Must define columns or expressions")
761 return values
764ColumnRef: TypeAlias = str
765"""Type alias for a column reference."""
768class ColumnGroup(BaseObject):
769 """Column group model."""
771 columns: list[ColumnRef | Column] = Field(..., min_length=1)
772 """Columns in the group."""
774 ivoa_ucd: str | None = Field(None, alias="ivoa:ucd")
775 """IVOA UCD of the column."""
777 table: Table | None = Field(None, exclude=True)
778 """Reference to the parent table."""
780 @field_validator("ivoa_ucd")
781 @classmethod
782 def check_ivoa_ucd(cls, ivoa_ucd: str) -> str:
783 """Check that IVOA UCD values are valid.
785 Parameters
786 ----------
787 ivoa_ucd
788 IVOA UCD value to check.
790 Returns
791 -------
792 `str`
793 The IVOA UCD value if it is valid.
794 """
795 return validate_ivoa_ucd(ivoa_ucd)
797 @model_validator(mode="after")
798 def check_unique_columns(self) -> ColumnGroup:
799 """Check that the columns list contains unique items.
801 Returns
802 -------
803 `ColumnGroup`
804 The column group being validated.
805 """
806 column_ids = [col if isinstance(col, str) else col.id for col in self.columns]
807 if len(column_ids) != len(set(column_ids)):
808 raise ValueError("Columns in the group must be unique")
809 return self
811 def _dereference_columns(self) -> None:
812 """Dereference ColumnRef to Column objects."""
813 if self.table is None:
814 raise ValueError("ColumnGroup must have a reference to its parent table")
816 dereferenced_columns: list[ColumnRef | Column] = []
817 for col in self.columns:
818 if isinstance(col, str):
819 # Dereference ColumnRef to Column object
820 try:
821 col_obj = self.table._find_column_by_id(col)
822 except KeyError as e:
823 raise ValueError(f"Column '{col}' not found in table '{self.table.name}'") from e
824 dereferenced_columns.append(col_obj)
825 else:
826 dereferenced_columns.append(col)
828 self.columns = dereferenced_columns
830 @field_serializer("columns")
831 def serialize_columns(self, columns: list[ColumnRef | Column]) -> list[str]:
832 """Serialize columns as their IDs.
834 Parameters
835 ----------
836 columns
837 The columns to serialize.
839 Returns
840 -------
841 `list` [ `str` ]
842 The serialized column IDs.
843 """
844 return [col if isinstance(col, str) else col.id for col in columns]
847class ColumnOverrides(BaseModel):
848 """Allowed overrides for a referenced column.
850 Notes
851 -----
852 All of these fields are optional. Values of None may be explicitly set to
853 override the corresponding attribute in the referenced column but only
854 for certain fields (see validation in `_check_non_nullable_overrides`).
855 """
857 model_config = CONFIG.copy()
859 datatype: DataType | None = None
860 """New datatype for the column."""
862 length: int | None = None
863 """New length for the column."""
865 description: str | None = None
866 """New description for the column."""
868 nullable: bool | None = None
869 """New nullable flag for the column."""
871 tap_principal: int | None = Field(default=None, alias="tap:principal")
872 """Override for the TAP_SCHEMA 'principal' flag."""
874 tap_column_index: int | None = Field(default=None, alias="tap:column_index")
875 """Override for the TAP_SCHEMA column index."""
877 @model_validator(mode="before")
878 @classmethod
879 def _check_non_nullable_overrides(cls, data: Any) -> Any:
880 """Check that certain fields are not overridden to null."""
881 if not isinstance(data, dict):
882 return data
883 non_nullable_fields = ("datatype", "length", "nullable", "tap_principal")
884 for name in non_nullable_fields:
885 if name in data and data[name] is None:
886 raise ValueError(f"The '{name}' field cannot be overridden to null")
887 return data
889 @field_serializer("datatype")
890 def serialize_datatype(self, value: DataType | None) -> str | None:
891 """Convert `DataType` to string when serializing to JSON/YAML.
893 Parameters
894 ----------
895 value
896 The `DataType` value to serialize, or None.
898 Returns
899 -------
900 `str` | None
901 The serialized `DataType` value, or None if the input was None.
902 """
903 if value is None:
904 return None
905 return str(value)
907 @field_validator("datatype", mode="before")
908 @classmethod
909 def deserialize_datatype(cls, value: str | None) -> DataType | None:
910 """Convert string back into `DataType` when loading from JSON/YAML.
912 Parameters
913 ----------
914 value
915 The string value to deserialize, or None.
917 Returns
918 -------
919 `DataType` | None
920 The deserialized `DataType` value, or None if the input was None.
921 """
922 if value is None: 922 ↛ 923line 922 didn't jump to line 923 because the condition on line 922 was never true
923 return None
924 return DataType(value)
927class ColumnResourceRef(BaseModel):
928 """A column which is dervived from an external resource."""
930 ref_name: str | None = None
931 """Name of the referenced column in the resource
932 (if different from the key)."""
934 overrides: ColumnOverrides | None = None
935 """Optional overrides of the referenced column's attributes."""
938# Type aliases for the nested mapping structure of referenced columns
939ResourceColumnMap: TypeAlias = dict[str, ColumnResourceRef | None]
940ResourceTableMap: TypeAlias = dict[str, ResourceColumnMap]
941ResourceMap: TypeAlias = dict[str, ResourceTableMap]
944class Table(BaseObject):
945 """Table model."""
947 primary_key: str | list[str] | None = Field(None, alias="primaryKey")
948 """Primary key of the table."""
950 tap_table_index: int | None = Field(None, alias="tap:table_index")
951 """IVOA TAP_SCHEMA table index of the table."""
953 mysql_engine: str | None = Field("MyISAM", alias="mysql:engine")
954 """MySQL engine to use for the table."""
956 mysql_charset: str | None = Field(None, alias="mysql:charset")
957 """MySQL charset to use for the table."""
959 column_refs: ResourceMap = Field(default_factory=dict, alias="columnRefs")
960 """Referenced columns from external resources."""
962 columns: list[Column] = Field(default_factory=list)
963 """Columns in the table."""
965 column_groups: list[ColumnGroup] = Field(default_factory=list, alias="columnGroups")
966 """Column groups in the table."""
968 constraints: list[_ConstraintType] = Field(default_factory=list)
969 """Constraints on the table."""
971 indexes: list[Index] = Field(default_factory=list)
972 """Indexes on the table."""
974 @field_validator("columns", mode="after")
975 @classmethod
976 def check_unique_column_names(cls, columns: list[Column]) -> list[Column]:
977 """Check that column names are unique.
979 Parameters
980 ----------
981 columns
982 The columns to check.
984 Returns
985 -------
986 `list` [ `Column` ]
987 The columns if they are unique.
989 Raises
990 ------
991 ValueError
992 Raised if column names are not unique.
993 """
994 if len(columns) != len(set(column.name for column in columns)):
995 raise ValueError("Column names must be unique")
996 return columns
998 @model_validator(mode="after")
999 def check_tap_table_index(self, info: ValidationInfo) -> Table:
1000 """Check that the table has a TAP table index.
1002 Parameters
1003 ----------
1004 info
1005 Validation context used to determine if the check is enabled.
1007 Returns
1008 -------
1009 `Table`
1010 The table being validated.
1012 Raises
1013 ------
1014 ValueError
1015 Raised If the table is missing a TAP table index.
1016 """
1017 context = info.context
1018 if not context or not context.get("check_tap_table_indexes", False):
1019 return self
1020 if self.tap_table_index is None:
1021 raise ValueError("Table is missing a TAP table index")
1022 return self
1024 @model_validator(mode="after")
1025 def check_tap_principal(self, info: ValidationInfo) -> Table:
1026 """Check that at least one column is flagged as 'principal' for TAP
1027 purposes.
1029 Parameters
1030 ----------
1031 info
1032 Validation context used to determine if the check is enabled.
1034 Returns
1035 -------
1036 `Table`
1037 The table being validated.
1039 Raises
1040 ------
1041 ValueError
1042 Raised if the table is missing a column flagged as 'principal'.
1043 """
1044 context = info.context
1045 if not context or not context.get("check_tap_principal", False):
1046 return self
1047 for col in self.columns:
1048 if col.tap_principal == 1:
1049 return self
1050 raise ValueError(f"Table '{self.name}' is missing at least one column designated as 'tap:principal'")
1052 def _find_column_by_id(self, id: str) -> Column:
1053 """Find a column by ID.
1055 Parameters
1056 ----------
1057 id
1058 The ID of the column to find.
1060 Returns
1061 -------
1062 `Column`
1063 The column with the given ID.
1065 Raises
1066 ------
1067 ValueError
1068 Raised if the column is not found.
1069 """
1070 for column in self.columns:
1071 if column.id == id:
1072 return column
1073 raise KeyError(f"Column '{id}' not found in table '{self.name}'")
1075 def _find_column_by_name(self, name: str) -> Column:
1076 for column in self.columns:
1077 if column.name == name:
1078 return column
1079 raise KeyError(f"Column '{name}' not found in table '{self.name}'")
1081 @model_validator(mode="after")
1082 def dereference_column_groups(self: Table) -> Table:
1083 """Dereference columns in column groups.
1085 Returns
1086 -------
1087 `Table`
1088 The table with dereferenced column groups.
1089 """
1090 for group in self.column_groups:
1091 group.table = self
1092 group._dereference_columns()
1093 return self
1095 @field_serializer("columns")
1096 def _serialize_columns(self, columns: list[Column]) -> list[dict[str, Any]]:
1097 """Serialize only non-resource columns."""
1098 return [
1099 col.model_dump(
1100 by_alias=True,
1101 exclude_none=True,
1102 exclude_defaults=True,
1103 )
1104 for col in columns
1105 if not col._is_resource_ref
1106 ]
1109class SchemaVersion(BaseModel):
1110 """Schema version model."""
1112 current: str
1113 """The current version of the schema."""
1115 compatible: list[str] = Field(default_factory=list)
1116 """The compatible versions of the schema."""
1118 read_compatible: list[str] = Field(default_factory=list)
1119 """The read compatible versions of the schema."""
1122class SchemaIdVisitor:
1123 """Visit a schema and build the map of IDs to objects.
1125 Notes
1126 -----
1127 Duplicates are added to a set when they are encountered, which can be
1128 accessed via the ``duplicates`` attribute. The presence of duplicates will
1129 not throw an error. Only the first object with a given ID will be added to
1130 the map, but this should not matter, since a ``ValidationError`` will be
1131 thrown by the ``model_validator`` method if any duplicates are found in the
1132 schema.
1133 """
1135 def __init__(self) -> None:
1136 """Create a new SchemaVisitor."""
1137 self.schema: Schema | None = None
1138 self.duplicates: set[str] = set()
1140 def add(self, obj: BaseObject) -> None:
1141 """Add an object to the ID map.
1143 Parameters
1144 ----------
1145 obj
1146 The object to add to the ID map.
1147 """
1148 if hasattr(obj, "id"): 1148 ↛ exitline 1148 didn't return from function 'add' because the condition on line 1148 was always true
1149 obj_id = getattr(obj, "id")
1150 if self.schema is not None: 1150 ↛ exitline 1150 didn't return from function 'add' because the condition on line 1150 was always true
1151 if obj_id in self.schema._id_map: 1151 ↛ 1152line 1151 didn't jump to line 1152 because the condition on line 1151 was never true
1152 self.duplicates.add(obj_id)
1153 else:
1154 self.schema._id_map[obj_id] = obj
1156 def visit_schema(self, schema: Schema) -> None:
1157 """Visit the objects in a schema and build the ID map.
1159 Parameters
1160 ----------
1161 schema
1162 The schema object to visit.
1164 Notes
1165 -----
1166 This will set an internal variable pointing to the schema object.
1167 """
1168 self.schema = schema
1169 self.duplicates.clear()
1170 self.add(self.schema)
1171 for table in self.schema.tables:
1172 self.visit_table(table)
1174 def visit_table(self, table: Table) -> None:
1175 """Visit a table object.
1177 Parameters
1178 ----------
1179 table
1180 The table object to visit.
1181 """
1182 self.add(table)
1183 for column in table.columns:
1184 self.visit_column(column)
1185 for constraint in table.constraints:
1186 self.visit_constraint(constraint)
1188 def visit_column(self, column: Column) -> None:
1189 """Visit a column object.
1191 Parameters
1192 ----------
1193 column
1194 The column object to visit.
1195 """
1196 self.add(column)
1198 def visit_constraint(self, constraint: Constraint) -> None:
1199 """Visit a constraint object.
1201 Parameters
1202 ----------
1203 constraint
1204 The constraint object to visit.
1205 """
1206 self.add(constraint)
1209T = TypeVar("T", bound=BaseObject)
1212def _strip_ids(data: Any) -> Any:
1213 """Recursively strip '@id' fields from a dictionary or list.
1215 Parameters
1216 ----------
1217 data
1218 The data to strip IDs from, which can be a dictionary, list, or any
1219 other type. Other types will be returned unchanged.
1220 """
1221 if isinstance(data, dict):
1222 data.pop("@id", None)
1223 for k, v in data.items():
1224 data[k] = _strip_ids(v)
1225 return data
1226 elif isinstance(data, list):
1227 return [_strip_ids(item) for item in data]
1228 else:
1229 return data
1232def _append_error(
1233 errors: list[InitErrorDetails],
1234 loc: tuple,
1235 input_value: Any,
1236 error_message: str,
1237 error_type: str = "value_error",
1238) -> None:
1239 """Append an error to the errors list.
1241 Parameters
1242 ----------
1243 errors : list[InitErrorDetails]
1244 The list of errors to append to.
1245 loc : tuple
1246 The location of the error in the schema.
1247 input_value : Any
1248 The input value that caused the error.
1249 error_message : str
1250 The error message to include in the context.
1251 """
1252 errors.append(
1253 {
1254 "type": error_type,
1255 "loc": loc,
1256 "input": input_value,
1257 "ctx": {"error": error_message},
1258 }
1259 )
1262class Resource(BaseModel):
1263 """A resource definition referencing an external schema."""
1265 uri: str = Field(..., description="Resource URI or path")
1266 """URI of the schema resource which may be a local path, ``resource://``,
1267 or remote URL."""
1270class Schema(BaseObject, Generic[T]):
1271 """Database schema model.
1273 This represents a database schema, which contains one or more tables.
1274 """
1276 version: SchemaVersion | str | None = None
1277 """The version of the schema."""
1279 resources: dict[str, Resource] = Field(default_factory=dict)
1280 """External resources referenced by this schema."""
1282 tables: Sequence[Table]
1283 """The tables in the schema."""
1285 _id_map: dict[str, Any] = PrivateAttr(default_factory=dict)
1286 """Map of IDs to objects."""
1288 _resource_map: dict[str, Schema] = PrivateAttr(default_factory=dict)
1289 """Map of resource names to loaded schemas."""
1291 @model_validator(mode="after")
1292 def _load_resources(self: Schema, info: ValidationInfo) -> Schema:
1293 """Load external resources referenced by this schema into an internal
1294 mapping of resource names to their `Schema` objects.
1296 Returns
1297 -------
1298 `Schema`
1299 The schema being validated.
1301 Raises
1302 ------
1303 ValueError
1304 Raised if a resource cannot be loaded.
1305 """
1306 if info.context:
1307 context = info.context.copy()
1308 # Ignore this flag for loading the resources themselves
1309 context.pop("dereference_resources", None)
1310 else:
1311 context = {}
1313 # Get the base URI for resolving relative resource paths from the
1314 # validation context, if available.
1315 resource_path = context.pop("resource_path", None)
1316 base_uri = None
1317 if resource_path is not None:
1318 base_uri = resource_path.parent()
1320 for resource_name, resource in self.resources.items():
1321 uri = resource.uri
1323 # Apply the base URI to the resource URI, if available.
1324 if base_uri is not None:
1325 orig_uri = uri
1326 uri = base_uri.join(uri, forceDirectory=False)
1327 if uri != orig_uri: 1327 ↛ 1336line 1327 didn't jump to line 1336 because the condition on line 1327 was always true
1328 logger.debug(
1329 "Resolved relative URI '%s' for resource '%s' to '%s' using base URI '%s'",
1330 resource.uri,
1331 resource_name,
1332 uri,
1333 base_uri,
1334 )
1336 try:
1337 loaded_schema = Schema.from_uri(uri, context=context)
1338 self._resource_map[resource_name] = loaded_schema
1339 logger.debug("Loaded resource '%s' from URI '%s'", resource_name, uri)
1340 except Exception as e:
1341 raise ValueError(f"Failed to load resource '{resource_name}' from URI '{uri}': {e}") from e
1342 return self
1344 def _find_table_by_name(self, name: str) -> Table:
1345 """Find a table by name.
1347 Parameters
1348 ----------
1349 name
1350 The name of the table to find.
1352 Returns
1353 -------
1354 `Table`
1355 The table with the given name.
1357 Raises
1358 ------
1359 KeyError
1360 Raised if the table is not found.
1361 """
1362 for table in self.tables:
1363 if table.name == name:
1364 return table
1365 raise KeyError(f"Table '{name}' not found in schema '{self.name}'")
1367 @model_validator(mode="after")
1368 def _dereference_resource_columns(self: Schema, info: ValidationInfo) -> Schema:
1369 """Dereference columns from external resources and add them to the
1370 tables in this schema.
1371 """
1372 context = info.context
1373 column_ref_index_increment: int | None = None
1374 dereference_resources = False
1375 if context is not None:
1376 dereference_resources = context.get("dereference_resources", False)
1377 column_ref_index_increment = context.get("column_ref_index_increment", None)
1379 for table in self.tables:
1380 if column_refs := table.column_refs:
1381 for resource_name, tables in column_refs.items():
1382 resource_schema = self._resource_map.get(resource_name)
1383 if resource_schema is None:
1384 raise ValueError(f"Schema resource '{resource_name}' was not found in resources")
1385 self._process_column_refs(
1386 table,
1387 tables,
1388 resource_schema,
1389 dereference_resources,
1390 column_ref_index_increment,
1391 )
1392 if dereference_resources and len(table.column_refs) > 0:
1393 # Clear column refs in table if fully dereferencing
1394 logger.debug(
1395 "Clearing columnRefs in table '%s' after dereferencing resource columns",
1396 table.name,
1397 )
1398 table.column_refs = {}
1399 return self
1401 @classmethod
1402 def _process_column_refs(
1403 cls,
1404 table: Table,
1405 ref_tables: ResourceTableMap,
1406 resource_schema: Schema,
1407 dereference_resources: bool = False,
1408 column_ref_index_increment: int | None = None,
1409 ) -> None:
1410 """Process column references from an external resource and add them
1411 to the given table as columns.
1412 """
1413 current_column_index = column_ref_index_increment if column_ref_index_increment is not None else -1
1415 for table_name, columns in ref_tables.items():
1416 try:
1417 resource_table = resource_schema._find_table_by_name(table_name)
1418 except KeyError as e:
1419 raise ValueError(
1420 f"Table '{table_name}' not found in resource '{resource_schema.name}'"
1421 ) from e
1422 for local_column_name, column_ref in columns.items():
1423 if column_ref is not None and column_ref.ref_name is not None:
1424 # Use specified ref_name
1425 ref_column_name = column_ref.ref_name
1426 else:
1427 # Use the local column name if no ref_name
1428 # specified
1429 ref_column_name = local_column_name
1431 # Check if referenced column exists in resource
1432 try:
1433 base_column = resource_table._find_column_by_name(ref_column_name)
1434 except KeyError:
1435 # The ref_name is specified but column is not
1436 # found
1437 if column_ref is not None and column_ref.ref_name is not None:
1438 raise ValueError(
1439 f"Column '{ref_column_name}' not found in table '{table_name}' "
1440 f"from resource '{resource_schema.name}'"
1441 )
1442 # The ref_name is not specified and the local
1443 # column name is not found
1444 raise ValueError(
1445 f"Column '{local_column_name}' not found in table '{table_name}' "
1446 f"from resource '{resource_schema.name}' and no ref_name provided"
1447 )
1449 # Create a copy of the base column
1450 column_copy = base_column.model_copy()
1452 # Set the local name (key from the mapping)
1453 column_copy.name = local_column_name
1455 if not dereference_resources:
1456 # Flag the column as a resource reference so it will not be
1457 # written out during serialization
1458 column_copy._is_resource_ref = True
1460 # Apply overrides to the referenced column definition
1461 overrides = column_ref.overrides if column_ref is not None else None
1462 if overrides is not None:
1463 column_copy._update_from_overrides(overrides)
1465 # Manually set the ID of the copied column as ID generation has
1466 # already occurred by now
1467 column_copy.id = f"{table.id}.{local_column_name}"
1469 # Apply automatic assignment of 'tap:column_index', if enabled
1470 if column_ref_index_increment is not None:
1471 if (not overrides) or (overrides.tap_column_index is None):
1472 column_copy.tap_column_index = current_column_index
1473 current_column_index += column_ref_index_increment
1474 logger.debug(
1475 "Automatically assigned 'tap:column_index' %s to "
1476 "column '%s' in table '%s' from resource '%s'",
1477 column_copy.tap_column_index,
1478 local_column_name,
1479 table_name,
1480 resource_schema.name,
1481 )
1482 else:
1483 # Skip automatic assignment of 'tap:column_index' if it
1484 # is already overridden
1485 logger.debug(
1486 "Skipping automatic assignment of 'tap:column_index' for column "
1487 "'%s' in table '%s' from resource '%s' as it is already overridden to %s",
1488 local_column_name,
1489 table_name,
1490 resource_schema.name,
1491 column_copy.tap_column_index,
1492 )
1493 table.columns.append(column_copy)
1494 logger.debug(
1495 "Dereferenced column '%s' from table '%s' in resource '%s' into table '%s'",
1496 local_column_name,
1497 table_name,
1498 resource_schema.name,
1499 table.name,
1500 )
1502 @model_validator(mode="before")
1503 @classmethod
1504 def generate_ids(cls, values: dict[str, Any], info: ValidationInfo) -> dict[str, Any]:
1505 """Generate IDs for objects that do not have them.
1507 Parameters
1508 ----------
1509 values
1510 The values of the schema.
1511 info
1512 Validation context used to determine if ID generation is enabled.
1514 Returns
1515 -------
1516 `dict` [ `str`, `Any` ]
1517 The values of the schema with generated IDs.
1518 """
1519 context = info.context
1520 if not context or not context.get("id_generation", False):
1521 logger.debug("Skipping ID generation")
1522 return values
1523 schema_name = values["name"]
1524 if "@id" not in values:
1525 values["@id"] = f"#{schema_name}"
1526 logger.debug("Generated ID '%s' for schema '%s'", values["@id"], schema_name)
1527 if "tables" in values: 1527 ↛ 1560line 1527 didn't jump to line 1560 because the condition on line 1527 was always true
1528 for table in values["tables"]:
1529 if "@id" not in table:
1530 table["@id"] = f"#{table['name']}"
1531 logger.debug("Generated ID '%s' for table '%s'", table["@id"], table["name"])
1532 if "columns" in table:
1533 for column in table["columns"]:
1534 if "@id" not in column:
1535 column["@id"] = f"#{table['name']}.{column['name']}"
1536 logger.debug("Generated ID '%s' for column '%s'", column["@id"], column["name"])
1537 if "columnGroups" in table:
1538 for column_group in table["columnGroups"]:
1539 if "@id" not in column_group:
1540 column_group["@id"] = f"#{table['name']}.{column_group['name']}"
1541 logger.debug(
1542 "Generated ID '%s' for column group '%s'",
1543 column_group["@id"],
1544 column_group["name"],
1545 )
1546 if "constraints" in table:
1547 for constraint in table["constraints"]:
1548 if "@id" not in constraint:
1549 constraint["@id"] = f"#{constraint['name']}"
1550 logger.debug(
1551 "Generated ID '%s' for constraint '%s'",
1552 constraint["@id"],
1553 constraint["name"],
1554 )
1555 if "indexes" in table:
1556 for index in table["indexes"]:
1557 if "@id" not in index:
1558 index["@id"] = f"#{index['name']}"
1559 logger.debug("Generated ID '%s' for index '%s'", index["@id"], index["name"])
1560 return values
1562 @field_validator("tables", mode="after")
1563 @classmethod
1564 def check_unique_table_names(cls, tables: list[Table]) -> list[Table]:
1565 """Check that table names are unique.
1567 Parameters
1568 ----------
1569 tables
1570 The tables to check.
1572 Returns
1573 -------
1574 `list` [ `Table` ]
1575 The tables if they are unique.
1577 Raises
1578 ------
1579 ValueError
1580 Raised if table names are not unique.
1581 """
1582 if len(tables) != len(set(table.name for table in tables)):
1583 raise ValueError("Table names must be unique")
1584 return tables
1586 @model_validator(mode="after")
1587 def check_tap_table_indexes(self, info: ValidationInfo) -> Schema:
1588 """Check that the TAP table indexes are unique.
1590 Parameters
1591 ----------
1592 info
1593 The validation context used to determine if the check is enabled.
1595 Returns
1596 -------
1597 `Schema`
1598 The schema being validated.
1599 """
1600 context = info.context
1601 if not context or not context.get("check_tap_table_indexes", False):
1602 return self
1603 table_indicies = set()
1604 for table in self.tables:
1605 table_index = table.tap_table_index
1606 if table_index is not None: 1606 ↛ 1604line 1606 didn't jump to line 1604 because the condition on line 1606 was always true
1607 if table_index in table_indicies:
1608 raise ValueError(f"Duplicate 'tap:table_index' value {table_index} found in schema")
1609 table_indicies.add(table_index)
1610 return self
1612 @model_validator(mode="after")
1613 def check_unique_constraint_names(self: Schema) -> Schema:
1614 """Check for duplicate constraint names in the schema.
1616 Returns
1617 -------
1618 `Schema`
1619 The schema being validated.
1621 Raises
1622 ------
1623 ValueError
1624 Raised if duplicate constraint names are found in the schema.
1625 """
1626 constraint_names = set()
1627 duplicate_names = []
1629 for table in self.tables:
1630 for constraint in table.constraints:
1631 constraint_name = constraint.name
1632 if constraint_name in constraint_names:
1633 duplicate_names.append(constraint_name)
1634 else:
1635 constraint_names.add(constraint_name)
1637 if duplicate_names:
1638 raise ValueError(f"Duplicate constraint names found in schema: {duplicate_names}")
1640 return self
1642 @model_validator(mode="after")
1643 def check_unique_index_names(self: Schema) -> Schema:
1644 """Check for duplicate index names in the schema.
1646 Returns
1647 -------
1648 `Schema`
1649 The schema being validated.
1651 Raises
1652 ------
1653 ValueError
1654 Raised if duplicate index names are found in the schema.
1655 """
1656 index_names = set()
1657 duplicate_names = []
1659 for table in self.tables:
1660 for index in table.indexes:
1661 index_name = index.name
1662 if index_name in index_names:
1663 duplicate_names.append(index_name)
1664 else:
1665 index_names.add(index_name)
1667 if duplicate_names:
1668 raise ValueError(f"Duplicate index names found in schema: {duplicate_names}")
1670 return self
1672 @model_validator(mode="after")
1673 def create_id_map(self: Schema) -> Schema:
1674 """Create a map of IDs to objects.
1676 Returns
1677 -------
1678 `Schema`
1679 The schema with the ID map created.
1681 Raises
1682 ------
1683 ValueError
1684 Raised if duplicate identifiers are found in the schema.
1685 """
1686 if self._id_map: 1686 ↛ 1687line 1686 didn't jump to line 1687 because the condition on line 1686 was never true
1687 logger.debug("Ignoring call to create_id_map() - ID map was already populated")
1688 return self
1689 visitor: SchemaIdVisitor = SchemaIdVisitor()
1690 visitor.visit_schema(self)
1691 if len(visitor.duplicates): 1691 ↛ 1692line 1691 didn't jump to line 1692 because the condition on line 1691 was never true
1692 raise ValueError(
1693 "Duplicate IDs found in schema:\n " + "\n ".join(visitor.duplicates) + "\n"
1694 )
1695 logger.debug("Created ID map with %d entries", len(self._id_map))
1696 return self
1698 def _validate_column_id(
1699 self: Schema,
1700 column_id: str,
1701 loc: tuple,
1702 errors: list[InitErrorDetails],
1703 ) -> None:
1704 """Validate a column ID from a constraint and append errors if invalid.
1706 Parameters
1707 ----------
1708 schema : Schema
1709 The schema being validated.
1710 column_id : str
1711 The column ID to validate.
1712 loc : tuple
1713 The location of the error in the schema.
1714 errors : list[InitErrorDetails]
1715 The list of errors to append to.
1716 """
1717 if column_id not in self:
1718 _append_error(
1719 errors,
1720 loc,
1721 column_id,
1722 f"Column ID '{column_id}' not found in schema",
1723 )
1724 elif not isinstance(self[column_id], Column):
1725 _append_error(
1726 errors,
1727 loc,
1728 column_id,
1729 f"ID '{column_id}' does not refer to a Column object",
1730 )
1732 def _validate_foreign_key_column(
1733 self: Schema,
1734 column_id: str,
1735 table: Table,
1736 loc: tuple,
1737 errors: list[InitErrorDetails],
1738 ) -> None:
1739 """Validate a foreign key column ID from a constraint and append errors
1740 if invalid.
1742 Parameters
1743 ----------
1744 schema : Schema
1745 The schema being validated.
1746 column_id : str
1747 The foreign key column ID to validate.
1748 loc : tuple
1749 The location of the error in the schema.
1750 errors : list[InitErrorDetails]
1751 The list of errors to append to.
1752 """
1753 try:
1754 table._find_column_by_id(column_id)
1755 except KeyError:
1756 _append_error(
1757 errors,
1758 loc,
1759 column_id,
1760 f"Column '{column_id}' not found in table '{table.name}'",
1761 )
1763 @model_validator(mode="after")
1764 def check_constraints(self: Schema) -> Schema:
1765 """Check constraint objects for validity. This needs to be deferred
1766 until after the schema is fully loaded and the ID map is created.
1768 Raises
1769 ------
1770 pydantic.ValidationError
1771 Raised if any constraints are invalid.
1773 Returns
1774 -------
1775 `Schema`
1776 The schema being validated.
1777 """
1778 errors: list[InitErrorDetails] = []
1780 for table_index, table in enumerate(self.tables):
1781 for constraint_index, constraint in enumerate(table.constraints):
1782 column_ids: list[str] = []
1783 referenced_column_ids: list[str] = []
1785 if isinstance(constraint, ForeignKeyConstraint):
1786 column_ids += constraint.columns
1787 referenced_column_ids += constraint.referenced_columns
1788 elif isinstance(constraint, UniqueConstraint):
1789 column_ids += constraint.columns
1790 # No extra checks are required on CheckConstraint objects.
1792 # Validate the foreign key columns
1793 for column_id in column_ids:
1794 self._validate_column_id(
1795 column_id,
1796 (
1797 "tables",
1798 table_index,
1799 "constraints",
1800 constraint_index,
1801 "columns",
1802 column_id,
1803 ),
1804 errors,
1805 )
1806 # Check that the foreign key column is within the source
1807 # table.
1808 self._validate_foreign_key_column(
1809 column_id,
1810 table,
1811 (
1812 "tables",
1813 table_index,
1814 "constraints",
1815 constraint_index,
1816 "columns",
1817 column_id,
1818 ),
1819 errors,
1820 )
1822 # Validate the primary key (reference) columns
1823 for referenced_column_id in referenced_column_ids:
1824 self._validate_column_id(
1825 referenced_column_id,
1826 (
1827 "tables",
1828 table_index,
1829 "constraints",
1830 constraint_index,
1831 "referenced_columns",
1832 referenced_column_id,
1833 ),
1834 errors,
1835 )
1837 if errors:
1838 raise ValidationError.from_exception_data("Schema validation failed", errors)
1840 return self
1842 def __getitem__(self, id: str) -> BaseObject:
1843 """Get an object by its ID.
1845 Parameters
1846 ----------
1847 id
1848 The ID of the object to get.
1850 Raises
1851 ------
1852 KeyError
1853 Raised if the object with the given ID is not found in the schema.
1854 """
1855 if id not in self:
1856 raise KeyError(f"Object with ID '{id}' not found in schema")
1857 return self._id_map[id]
1859 def __contains__(self, id: str) -> bool:
1860 """Check if an object with the given ID is in the schema.
1862 Parameters
1863 ----------
1864 id
1865 The ID of the object to check.
1866 """
1867 return id in self._id_map
1869 def find_object_by_id(self, id: str, obj_type: type[T]) -> T:
1870 """Find an object with the given type by its ID.
1872 Parameters
1873 ----------
1874 id
1875 The ID of the object to find.
1876 obj_type
1877 The type of the object to find.
1879 Returns
1880 -------
1881 BaseObject
1882 The object with the given ID and type.
1884 Raises
1885 ------
1886 KeyError
1887 If the object with the given ID is not found in the schema.
1888 TypeError
1889 If the object that is found does not have the right type.
1891 Notes
1892 -----
1893 The actual return type is the user-specified argument ``T``, which is
1894 expected to be a subclass of `BaseObject`.
1895 """
1896 obj = self[id]
1897 if not isinstance(obj, obj_type):
1898 raise TypeError(f"Object with ID '{id}' is not of type '{obj_type.__name__}'")
1899 return obj
1901 def get_table_by_column(self, column: Column) -> Table:
1902 """Find the table that contains a column.
1904 Parameters
1905 ----------
1906 column
1907 The column to find.
1909 Returns
1910 -------
1911 `Table`
1912 The table that contains the column.
1914 Raises
1915 ------
1916 ValueError
1917 If the column is not found in any table.
1918 """
1919 for table in self.tables:
1920 if column in table.columns:
1921 return table
1922 raise ValueError(f"Column '{column.name}' not found in any table")
1924 @classmethod
1925 def from_uri(cls, resource_path: ResourcePathExpression, context: dict[str, Any] = {}) -> Schema:
1926 """Load a `Schema` from a string representing a ``ResourcePath``.
1928 Parameters
1929 ----------
1930 resource_path
1931 The ``ResourcePath`` pointing to a YAML file.
1932 context
1933 Pydantic context to be used in validation.
1935 Returns
1936 -------
1937 `str`
1938 The ID of the object.
1940 Raises
1941 ------
1942 yaml.YAMLError
1943 Raised if there is an error loading the YAML data.
1944 ValueError
1945 Raised if there is an error reading the resource.
1946 pydantic.ValidationError
1947 Raised if the schema fails validation.
1948 """
1949 try:
1950 rp = ResourcePath(resource_path, forceAbsolute=False, forceDirectory=False)
1951 rp_data = rp.read()
1952 except Exception as e:
1953 raise ValueError(f"Error reading resource from '{resource_path}' : {e}") from e
1954 yaml_data = yaml.safe_load(rp_data)
1955 context = dict(context)
1956 # Append the resource path to the context for resolving resource URLs.
1957 context["resource_path"] = rp
1958 return Schema.model_validate(yaml_data, context=context)
1960 @classmethod
1961 def from_stream(cls, source: IO[str], context: dict[str, Any] = {}) -> Schema:
1962 """Load a `Schema` from a file stream which should contain YAML data.
1964 Parameters
1965 ----------
1966 source
1967 The file stream to read from.
1968 context
1969 Pydantic context to be used in validation.
1971 Returns
1972 -------
1973 `Schema`
1974 The Felis schema loaded from the stream.
1976 Raises
1977 ------
1978 yaml.YAMLError
1979 Raised if there is an error loading the YAML file.
1980 pydantic.ValidationError
1981 Raised if the schema fails validation.
1982 """
1983 logger.debug("Loading schema from: '%s'", source)
1984 yaml_data = yaml.safe_load(source)
1985 return Schema.model_validate(yaml_data, context=context)
1987 def _model_dump(self, strip_ids: bool = False, sort_columns: bool = False) -> dict[str, Any]:
1988 """Dump the schema as a dictionary with some default arguments
1989 applied.
1991 Parameters
1992 ----------
1993 strip_ids
1994 Whether to strip the IDs from the dumped data. Defaults to `False`.
1995 sort_columns
1996 Whether to sort columns alphabetically by name. Defaults to
1997 `False`.
1999 Returns
2000 -------
2001 `dict` [ `str`, `Any` ]
2002 The dumped schema data as a dictionary.
2003 """
2004 data = self.model_dump(by_alias=True, exclude_none=True, exclude_defaults=True)
2005 if strip_ids:
2006 data = _strip_ids(data)
2007 if sort_columns:
2008 for table in data.get("tables", []):
2009 table["columns"] = sorted(table.get("columns", []), key=itemgetter("name"))
2010 return data
2012 def dump_yaml(
2013 self, stream: IO[str] = sys.stdout, strip_ids: bool = False, sort_columns: bool = False
2014 ) -> None:
2015 """Pretty print the schema as YAML.
2017 Parameters
2018 ----------
2019 stream
2020 The stream to write the YAML data to.
2021 strip_ids
2022 Whether to strip the IDs from the dumped data. Defaults to `False`.
2023 sort_columns
2024 Whether to sort columns alphabetically by name. Defaults to
2025 `False`.
2026 """
2027 data = self._model_dump(strip_ids=strip_ids, sort_columns=sort_columns)
2028 yaml.safe_dump(
2029 data,
2030 stream,
2031 default_flow_style=False,
2032 sort_keys=False,
2033 )
2035 def dump_json(
2036 self, stream: IO[str] = sys.stdout, strip_ids: bool = False, sort_columns: bool = False
2037 ) -> None:
2038 """Pretty print the schema as JSON.
2040 Parameters
2041 ----------
2042 stream
2043 The stream to write the JSON data to.
2044 strip_ids
2045 Whether to strip the IDs from the dumped data. Defaults to `False`.
2046 sort_columns
2047 Whether to sort columns alphabetically by name. Defaults to
2048 `False`.
2049 """
2050 data = self._model_dump(strip_ids=strip_ids, sort_columns=sort_columns)
2051 json.dump(
2052 data,
2053 stream,
2054 indent=4,
2055 sort_keys=False,
2056 )