Coverage for python/felis/datamodel.py: 98%
732 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-01 08:46 +0000
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 "ColumnOverrides",
62 "ColumnResourceRef",
63 "Constraint",
64 "DataType",
65 "ForeignKeyConstraint",
66 "Index",
67 "Resource",
68 "Schema",
69 "SchemaVersion",
70 "Table",
71 "UniqueConstraint",
72)
74CONFIG = ConfigDict(
75 populate_by_name=True, # Populate attributes by name.
76 extra="forbid", # Do not allow extra fields.
77 str_strip_whitespace=True, # Strip whitespace from string fields.
78 use_enum_values=False, # Do not use enum values during serialization.
79)
80"""Pydantic model configuration as described in:
81https://docs.pydantic.dev/2.0/api/config/#pydantic.config.ConfigDict
82"""
85class BaseObject(BaseModel):
86 """Base model.
88 All classes representing objects in the Felis data model should inherit
89 from this class.
90 """
92 model_config = CONFIG
93 """Pydantic model configuration."""
95 name: str
96 """Name of the database object."""
98 id: str = Field(alias="@id")
99 """Unique identifier of the database object."""
101 description: str | None = None
102 """Description of the database object."""
104 votable_utype: str | None = Field(None, alias="votable:utype")
105 """VOTable utype (usage-specific or unique type) of the object."""
107 @model_validator(mode="after")
108 def check_description(self, info: ValidationInfo) -> BaseObject:
109 """Check that the description is present and contains at least one
110 non-whitespace character, if the validation context indicates that this
111 check is enabled. By default, descriptions may be omitted or empty.
113 Parameters
114 ----------
115 info
116 Validation context used to determine if the check is enabled.
118 Returns
119 -------
120 `BaseObject`
121 The object being validated.
122 """
123 context = info.context
124 if not context or not context.get("check_description", False):
125 return self
126 if self.description is None or self.description == "":
127 raise ValueError("Description is required and must be non-empty")
128 return self
131class DataType(StrEnum):
132 """``Enum`` representing the data types supported by Felis."""
134 boolean = auto()
135 byte = auto()
136 short = auto()
137 int = auto()
138 long = auto()
139 float = auto()
140 double = auto()
141 char = auto()
142 string = auto()
143 unicode = auto()
144 text = auto()
145 binary = auto()
146 timestamp = auto()
149def validate_ivoa_ucd(ivoa_ucd: str) -> str:
150 """Validate IVOA UCD values.
152 Parameters
153 ----------
154 ivoa_ucd
155 IVOA UCD value to check.
157 Returns
158 -------
159 `str`
160 The IVOA UCD value if it is valid.
162 Raises
163 ------
164 ValueError
165 If the IVOA UCD value is invalid.
166 """
167 if ivoa_ucd is not None: 167 ↛ 172line 167 didn't jump to line 172 because the condition on line 167 was always true
168 try:
169 ucd.parse_ucd(ivoa_ucd, check_controlled_vocabulary=True, has_colon=";" in ivoa_ucd)
170 except ValueError as e:
171 raise ValueError(f"Invalid IVOA UCD: {e}")
172 return ivoa_ucd
175class Column(BaseObject):
176 """Column model."""
178 datatype: DataType
179 """Datatype of the column."""
181 length: int | None = Field(None, gt=0)
182 """Length of the column."""
184 precision: int | None = Field(None, ge=0)
185 """The numerical precision of the column.
187 For timestamps, this is the number of fractional digits retained in the
188 seconds field.
189 """
191 nullable: bool = True
192 """Whether the column can be ``NULL``."""
194 value: str | int | float | bool | None = None
195 """Default value of the column."""
197 autoincrement: bool | None = None
198 """Whether the column is autoincremented."""
200 ivoa_ucd: str | None = Field(None, alias="ivoa:ucd")
201 """IVOA UCD of the column."""
203 fits_tunit: str | None = Field(None, alias="fits:tunit")
204 """FITS TUNIT of the column."""
206 ivoa_unit: str | None = Field(None, alias="ivoa:unit")
207 """IVOA unit of the column."""
209 tap_column_index: int | None = Field(None, alias="tap:column_index")
210 """TAP_SCHEMA column index of the column."""
212 tap_principal: int | None = Field(0, alias="tap:principal", ge=0, le=1)
213 """Whether this is a TAP_SCHEMA principal column."""
215 votable_arraysize: int | str | None = Field(None, alias="votable:arraysize")
216 """VOTable arraysize of the column."""
218 tap_std: int | None = Field(0, alias="tap:std", ge=0, le=1)
219 """TAP_SCHEMA indication that this column is defined by an IVOA standard.
220 """
222 votable_xtype: str | None = Field(None, alias="votable:xtype")
223 """VOTable xtype (extended type) of the column."""
225 votable_datatype: str | None = Field(None, alias="votable:datatype")
226 """VOTable datatype of the column."""
228 mysql_datatype: str | None = Field(None, alias="mysql:datatype")
229 """MySQL datatype override on the column."""
231 postgresql_datatype: str | None = Field(None, alias="postgresql:datatype")
232 """PostgreSQL datatype override on the column."""
234 _is_resource_ref: bool = PrivateAttr(False)
235 """Whether this column is a resource reference column."""
237 @model_validator(mode="after")
238 def check_value(self) -> Column:
239 """Check that the default value is valid.
241 Returns
242 -------
243 `Column`
244 The column being validated.
245 """
246 if (value := self.value) is not None:
247 if value is not None and self.autoincrement is True:
248 raise ValueError("Column cannot have both a default value and be autoincremented")
249 felis_type = FelisType.felis_type(self.datatype)
250 if felis_type.is_numeric:
251 if felis_type in (Byte, Short, Int, Long) and not isinstance(value, int):
252 raise ValueError("Default value must be an int for integer type columns")
253 elif felis_type in (Float, Double) and not isinstance(value, float):
254 raise ValueError("Default value must be a decimal number for float and double columns")
255 elif felis_type in (String, Char, Unicode, Text):
256 if not isinstance(value, str):
257 raise ValueError("Default value must be a string for string columns")
258 if not len(value):
259 raise ValueError("Default value must be a non-empty string for string columns")
260 elif felis_type is Boolean and not isinstance(value, bool):
261 raise ValueError("Default value must be a boolean for boolean columns")
262 return self
264 @field_validator("ivoa_ucd")
265 @classmethod
266 def check_ivoa_ucd(cls, ivoa_ucd: str) -> str:
267 """Check that IVOA UCD values are valid.
269 Parameters
270 ----------
271 ivoa_ucd
272 IVOA UCD value to check.
274 Returns
275 -------
276 `str`
277 The IVOA UCD value if it is valid.
278 """
279 return validate_ivoa_ucd(ivoa_ucd)
281 @model_validator(mode="after")
282 def check_units(self) -> Column:
283 """Check that the ``fits:tunit`` or ``ivoa:unit`` field has valid
284 units according to astropy. Only one may be provided.
286 Returns
287 -------
288 `Column`
289 The column being validated.
291 Raises
292 ------
293 ValueError
294 Raised if both FITS and IVOA units are provided, or if the unit is
295 invalid.
296 """
297 fits_unit = self.fits_tunit
298 ivoa_unit = self.ivoa_unit
300 if fits_unit and ivoa_unit:
301 raise ValueError("Column cannot have both FITS and IVOA units")
302 unit = fits_unit or ivoa_unit
304 if unit is not None:
305 try:
306 units.Unit(unit)
307 except ValueError as e:
308 raise ValueError(f"Invalid unit: {e}")
310 return self
312 @model_validator(mode="before")
313 @classmethod
314 def check_length(cls, values: dict[str, Any]) -> dict[str, Any]:
315 """Check that a valid length is provided for sized types.
317 Parameters
318 ----------
319 values
320 Values of the column.
322 Returns
323 -------
324 `dict` [ `str`, `Any` ]
325 The values of the column.
327 Raises
328 ------
329 ValueError
330 Raised if a length is not provided for a sized type.
331 """
332 datatype = values.get("datatype")
333 if datatype is None:
334 # Skip this validation if datatype is not provided
335 return values
336 length = values.get("length")
337 felis_type = FelisType.felis_type(datatype)
338 if felis_type.is_sized and length is None:
339 raise ValueError(
340 f"Length must be provided for type '{datatype}'"
341 + (f" in column '{values['@id']}'" if "@id" in values else "")
342 )
343 elif not felis_type.is_sized and length is not None:
344 msg = f"The datatype '{datatype}' does not support a specified length"
345 if "@id" in values:
346 msg += f" in column '{values['@id']}'"
347 logger.warning("%s", msg)
348 return values
350 @model_validator(mode="after")
351 def check_redundant_datatypes(self, info: ValidationInfo) -> Column:
352 """Check for redundant datatypes on columns.
354 Parameters
355 ----------
356 info
357 Validation context used to determine if the check is enabled.
359 Returns
360 -------
361 `Column`
362 The column being validated.
364 Raises
365 ------
366 ValueError
367 Raised if a datatype override is redundant.
368 """
369 context = info.context
370 if not context or not context.get("check_redundant_datatypes", False):
371 return self
372 if all( 372 ↛ 376line 372 didn't jump to line 376 because the condition on line 372 was never true
373 getattr(self, f"{dialect}:datatype", None) is not None
374 for dialect in get_supported_dialects().keys()
375 ):
376 return self
378 datatype = self.datatype
379 length: int | None = self.length or None
381 datatype_func = get_type_func(datatype)
382 felis_type = FelisType.felis_type(datatype)
383 if felis_type.is_sized:
384 datatype_obj = datatype_func(length)
385 else:
386 datatype_obj = datatype_func()
388 for dialect_name, dialect in get_supported_dialects().items():
389 db_annotation = f"{dialect_name}_datatype"
390 if datatype_string := self.model_dump().get(db_annotation):
391 db_datatype_obj = string_to_typeengine(datatype_string, dialect, length)
392 if datatype_obj.compile(dialect) == db_datatype_obj.compile(dialect):
393 raise ValueError(
394 "'{}: {}' is a redundant override of 'datatype: {}' in column '{}'{}".format(
395 db_annotation,
396 datatype_string,
397 self.datatype,
398 self.id,
399 "" if length is None else f" with length {length}",
400 )
401 )
402 else:
403 logger.debug(
404 "Type override of 'datatype: %s' with '%s: %s' in column '%s' "
405 "compiled to '%s' and '%s'",
406 self.datatype,
407 db_annotation,
408 datatype_string,
409 self.id,
410 datatype_obj.compile(dialect),
411 db_datatype_obj.compile(dialect),
412 )
413 return self
415 @model_validator(mode="after")
416 def check_precision(self) -> Column:
417 """Check that precision is only valid for timestamp columns.
419 Returns
420 -------
421 `Column`
422 The column being validated.
423 """
424 if self.precision is not None and self.datatype != "timestamp":
425 raise ValueError("Precision is only valid for timestamp columns")
426 return self
428 @model_validator(mode="before")
429 @classmethod
430 def check_votable_arraysize(cls, values: dict[str, Any], info: ValidationInfo) -> dict[str, Any]:
431 """Set the default value for the ``votable_arraysize`` field, which
432 corresponds to ``arraysize`` in the IVOA VOTable standard.
434 Parameters
435 ----------
436 values
437 Values of the column.
438 info
439 Validation context used to determine if the check is enabled.
441 Returns
442 -------
443 `dict` [ `str`, `Any` ]
444 The values of the column.
446 Notes
447 -----
448 Following the IVOA VOTable standard, an ``arraysize`` of 1 should not
449 be used.
450 """
451 if values.get("name", None) is None or values.get("datatype", None) is None:
452 # Skip bad column data that will not validate
453 return values
454 context = info.context if info.context else {}
455 arraysize = values.get("votable:arraysize", None)
456 if arraysize is None:
457 length = values.get("length", None)
458 datatype = values.get("datatype")
459 if length is not None and length > 1:
460 # Following the IVOA standard, arraysize of 1 is disallowed
461 if datatype == "char":
462 arraysize = str(length)
463 elif datatype in ("string", "unicode", "binary"):
464 if context.get("force_unbounded_arraysize", False):
465 arraysize = "*"
466 logger.debug(
467 "Forced VOTable's 'arraysize' to '*' on column '%s' with datatype "
468 "'%s' and length '%s'",
469 values["name"],
470 values["datatype"],
471 length,
472 )
473 else:
474 arraysize = f"{length}*"
475 elif datatype in ("timestamp", "text"):
476 arraysize = "*"
477 if arraysize is not None:
478 values["votable:arraysize"] = arraysize
479 logger.debug(
480 "Set default 'votable:arraysize' to '%s' on column '%s'"
481 " with datatype '%s' and length '%s'",
482 arraysize,
483 values["name"],
484 values["datatype"],
485 values.get("length", None),
486 )
487 else:
488 logger.debug(
489 "Using existing 'votable:arraysize' of '%s' on column '%s'", arraysize, values["name"]
490 )
491 if isinstance(values["votable:arraysize"], int): 491 ↛ 492line 491 didn't jump to line 492 because the condition on line 491 was never true
492 logger.warning(
493 "Usage of an integer value for 'votable:arraysize' in column '%s' is deprecated",
494 values["name"],
495 )
496 values["votable:arraysize"] = str(arraysize)
497 return values
499 @field_serializer("datatype")
500 def serialize_datatype(self, value: DataType) -> str:
501 """Convert `DataType` to string when serializing to JSON/YAML.
503 Parameters
504 ----------
505 value
506 The `DataType` value to serialize.
508 Returns
509 -------
510 `str`
511 The serialized `DataType` value.
512 """
513 return str(value)
515 @field_validator("datatype", mode="before")
516 @classmethod
517 def deserialize_datatype(cls, value: str) -> DataType:
518 """Convert string back into `DataType` when loading from JSON/YAML.
520 Parameters
521 ----------
522 value
523 The string value to deserialize.
525 Returns
526 -------
527 `DataType`
528 The deserialized `DataType` value.
529 """
530 return DataType(value)
532 @model_validator(mode="after")
533 def check_votable_xtype(self) -> Column:
534 """Set the default value for the ``votable_xtype`` field, which
535 corresponds to an Extended Datatype or ``xtype`` in the IVOA VOTable
536 standard.
538 Returns
539 -------
540 `Column`
541 The column being validated.
543 Notes
544 -----
545 This is currently only set automatically for the Felis ``timestamp``
546 datatype.
547 """
548 if self.datatype == DataType.timestamp and self.votable_xtype is None:
549 self.votable_xtype = "timestamp"
550 return self
552 def _update_from_overrides(self, overrides: ColumnOverrides) -> None:
553 """Update the column attributes from the given overrides.
555 Parameters
556 ----------
557 overrides
558 The column overrides to apply or `None` to skip applying overrides.
560 Notes
561 -----
562 Using ``model_fields_set`` allows updating only the fields that are
563 explicitly set in the `overrides` object. This prevents overwriting
564 existing column attributes which were not explicitly provided.
565 """
566 if overrides.model_fields_set:
567 logger.debug("Applying overrides to column '%s': %s", self.id, overrides.model_fields_set)
568 for field in overrides.model_fields_set:
569 setattr(self, field, getattr(overrides, field))
572class Constraint(BaseObject):
573 """Table constraint model."""
575 deferrable: bool = False
576 """Whether this constraint will be declared as deferrable."""
578 initially: Literal["IMMEDIATE", "DEFERRED"] | None = None
579 """Value for ``INITIALLY`` clause; only used if `deferrable` is
580 `True`."""
582 @model_validator(mode="after")
583 def check_deferrable(self) -> Constraint:
584 """Check that the ``INITIALLY`` clause is only used if `deferrable` is
585 `True`.
587 Returns
588 -------
589 `Constraint`
590 The constraint being validated.
591 """
592 if self.initially is not None and not self.deferrable:
593 raise ValueError("INITIALLY clause can only be used if deferrable is True")
594 return self
597class CheckConstraint(Constraint):
598 """Table check constraint model."""
600 type: Literal["Check"] = Field("Check", alias="@type")
601 """Type of the constraint."""
603 expression: str
604 """Expression for the check constraint."""
606 @field_serializer("type")
607 def serialize_type(self, value: str) -> str:
608 """Ensure '@type' is included in serialized output.
610 Parameters
611 ----------
612 value
613 The value to serialize.
615 Returns
616 -------
617 `str`
618 The serialized value.
619 """
620 return value
623class UniqueConstraint(Constraint):
624 """Table unique constraint model."""
626 type: Literal["Unique"] = Field("Unique", alias="@type")
627 """Type of the constraint."""
629 columns: list[str]
630 """Columns in the unique constraint."""
632 @field_serializer("type")
633 def serialize_type(self, value: str) -> str:
634 """Ensure '@type' is included in serialized output.
636 Parameters
637 ----------
638 value
639 The value to serialize.
641 Returns
642 -------
643 `str`
644 The serialized value.
645 """
646 return value
649class ForeignKeyConstraint(Constraint):
650 """Table foreign key constraint model.
652 This constraint is used to define a foreign key relationship between two
653 tables in the schema. There must be at least one column in the
654 `columns` list, and at least one column in the `referenced_columns` list
655 or a validation error will be raised.
657 Notes
658 -----
659 These relationships will be reflected in the TAP_SCHEMA ``keys`` and
660 ``key_columns`` data.
661 """
663 type: Literal["ForeignKey"] = Field("ForeignKey", alias="@type")
664 """Type of the constraint."""
666 columns: list[str] = Field(min_length=1)
667 """The columns comprising the foreign key."""
669 referenced_columns: list[str] = Field(alias="referencedColumns", min_length=1)
670 """The columns referenced by the foreign key."""
672 on_delete: Literal["CASCADE", "SET NULL", "SET DEFAULT", "RESTRICT", "NO ACTION"] | None = None
673 """Action to take when the referenced row is deleted."""
675 on_update: Literal["CASCADE", "SET NULL", "SET DEFAULT", "RESTRICT", "NO ACTION"] | None = None
676 """Action to take when the referenced row is updated."""
678 @field_serializer("type")
679 def serialize_type(self, value: str) -> str:
680 """Ensure '@type' is included in serialized output.
682 Parameters
683 ----------
684 value
685 The value to serialize.
687 Returns
688 -------
689 `str`
690 The serialized value.
691 """
692 return value
694 @model_validator(mode="after")
695 def check_column_lengths(self) -> ForeignKeyConstraint:
696 """Check that the `columns` and `referenced_columns` lists have the
697 same length.
699 Returns
700 -------
701 `ForeignKeyConstraint`
702 The foreign key constraint being validated.
704 Raises
705 ------
706 ValueError
707 Raised if the `columns` and `referenced_columns` lists do not have
708 the same length.
709 """
710 if len(self.columns) != len(self.referenced_columns):
711 raise ValueError(
712 "Columns and referencedColumns must have the same length for a ForeignKey constraint"
713 )
714 return self
717_ConstraintType = Annotated[
718 CheckConstraint | ForeignKeyConstraint | UniqueConstraint, Field(discriminator="type")
719]
720"""Type alias for a constraint type."""
723class Index(BaseObject):
724 """Table index model.
726 An index can be defined on either columns or expressions, but not both.
727 """
729 columns: list[str] | None = None
730 """Columns in the index."""
732 expressions: list[str] | None = None
733 """Expressions in the index."""
735 @model_validator(mode="before")
736 @classmethod
737 def check_columns_or_expressions(cls, values: dict[str, Any]) -> dict[str, Any]:
738 """Check that columns or expressions are specified, but not both.
740 Parameters
741 ----------
742 values
743 Values of the index.
745 Returns
746 -------
747 `dict` [ `str`, `Any` ]
748 The values of the index.
750 Raises
751 ------
752 ValueError
753 Raised if both columns and expressions are specified, or if neither
754 are specified.
755 """
756 if "columns" in values and "expressions" in values:
757 raise ValueError("Defining columns and expressions is not valid")
758 elif "columns" not in values and "expressions" not in values:
759 raise ValueError("Must define columns or expressions")
760 return values
763ColumnRef: TypeAlias = str
764"""Type alias for a column reference."""
767class ColumnGroup(BaseObject):
768 """Column group model."""
770 columns: list[ColumnRef | Column] = Field(..., min_length=1)
771 """Columns in the group."""
773 ivoa_ucd: str | None = Field(None, alias="ivoa:ucd")
774 """IVOA UCD of the column."""
776 table: Table | None = Field(None, exclude=True)
777 """Reference to the parent table."""
779 @field_validator("ivoa_ucd")
780 @classmethod
781 def check_ivoa_ucd(cls, ivoa_ucd: str) -> str:
782 """Check that IVOA UCD values are valid.
784 Parameters
785 ----------
786 ivoa_ucd
787 IVOA UCD value to check.
789 Returns
790 -------
791 `str`
792 The IVOA UCD value if it is valid.
793 """
794 return validate_ivoa_ucd(ivoa_ucd)
796 @model_validator(mode="after")
797 def check_unique_columns(self) -> ColumnGroup:
798 """Check that the columns list contains unique items.
800 Returns
801 -------
802 `ColumnGroup`
803 The column group being validated.
804 """
805 column_ids = [col if isinstance(col, str) else col.id for col in self.columns]
806 if len(column_ids) != len(set(column_ids)):
807 raise ValueError("Columns in the group must be unique")
808 return self
810 def _dereference_columns(self) -> None:
811 """Dereference ColumnRef to Column objects."""
812 if self.table is None:
813 raise ValueError("ColumnGroup must have a reference to its parent table")
815 dereferenced_columns: list[ColumnRef | Column] = []
816 for col in self.columns:
817 if isinstance(col, str):
818 # Dereference ColumnRef to Column object
819 try:
820 col_obj = self.table._find_column_by_id(col)
821 except KeyError as e:
822 raise ValueError(f"Column '{col}' not found in table '{self.table.name}'") from e
823 dereferenced_columns.append(col_obj)
824 else:
825 dereferenced_columns.append(col)
827 self.columns = dereferenced_columns
829 @field_serializer("columns")
830 def serialize_columns(self, columns: list[ColumnRef | Column]) -> list[str]:
831 """Serialize columns as their IDs.
833 Parameters
834 ----------
835 columns
836 The columns to serialize.
838 Returns
839 -------
840 `list` [ `str` ]
841 The serialized column IDs.
842 """
843 return [col if isinstance(col, str) else col.id for col in columns]
846class ColumnOverrides(BaseModel):
847 """Allowed overrides for a referenced column.
849 Notes
850 -----
851 All of these fields are optional. Values of None may be explicitly set to
852 override the corresponding attribute in the referenced column but only
853 for certain fields (see validation in `_check_non_nullable_overrides`).
854 """
856 model_config = CONFIG.copy()
858 datatype: DataType | None = None
859 """New datatype for the column."""
861 length: int | None = None
862 """New length for the column."""
864 description: str | None = None
865 """New description for the column."""
867 nullable: bool | None = None
868 """New nullable flag for the column."""
870 tap_principal: int | None = Field(default=None, alias="tap:principal")
871 """Override for the TAP_SCHEMA 'principal' flag."""
873 tap_column_index: int | None = Field(default=None, alias="tap:column_index")
874 """Override for the TAP_SCHEMA column index."""
876 @model_validator(mode="before")
877 @classmethod
878 def _check_non_nullable_overrides(cls, data: Any) -> Any:
879 """Check that certain fields are not overridden to null."""
880 if not isinstance(data, dict):
881 return data
882 non_nullable_fields = ("datatype", "length", "nullable", "tap_principal")
883 for name in non_nullable_fields:
884 if name in data and data[name] is None:
885 raise ValueError(f"The '{name}' field cannot be overridden to null")
886 return data
888 @field_serializer("datatype")
889 def serialize_datatype(self, value: DataType | None) -> str | None:
890 """Convert `DataType` to string when serializing to JSON/YAML.
892 Parameters
893 ----------
894 value
895 The `DataType` value to serialize, or None.
897 Returns
898 -------
899 `str` | None
900 The serialized `DataType` value, or None if the input was None.
901 """
902 if value is None:
903 return None
904 return str(value)
906 @field_validator("datatype", mode="before")
907 @classmethod
908 def deserialize_datatype(cls, value: str | None) -> DataType | None:
909 """Convert string back into `DataType` when loading from JSON/YAML.
911 Parameters
912 ----------
913 value
914 The string value to deserialize, or None.
916 Returns
917 -------
918 `DataType` | None
919 The deserialized `DataType` value, or None if the input was None.
920 """
921 if value is None: 921 ↛ 922line 921 didn't jump to line 922 because the condition on line 921 was never true
922 return None
923 return DataType(value)
926class ColumnResourceRef(BaseModel):
927 """A column which is dervived from an external resource."""
929 ref_name: str | None = None
930 """Name of the referenced column in the resource
931 (if different from the key)."""
933 overrides: ColumnOverrides | None = None
934 """Optional overrides of the referenced column's attributes."""
937# Type aliases for the nested mapping structure of referenced columns
938ResourceColumnMap: TypeAlias = dict[str, ColumnResourceRef | None]
939ResourceTableMap: TypeAlias = dict[str, ResourceColumnMap]
940ResourceMap: TypeAlias = dict[str, ResourceTableMap]
943class Table(BaseObject):
944 """Table model."""
946 primary_key: str | list[str] | None = Field(None, alias="primaryKey")
947 """Primary key of the table."""
949 tap_table_index: int | None = Field(None, alias="tap:table_index")
950 """IVOA TAP_SCHEMA table index of the table."""
952 mysql_engine: str | None = Field("MyISAM", alias="mysql:engine")
953 """MySQL engine to use for the table."""
955 mysql_charset: str | None = Field(None, alias="mysql:charset")
956 """MySQL charset to use for the table."""
958 column_refs: ResourceMap = Field(default_factory=dict, alias="columnRefs")
959 """Referenced columns from external resources."""
961 columns: list[Column] = Field(default_factory=list)
962 """Columns in the table."""
964 column_groups: list[ColumnGroup] = Field(default_factory=list, alias="columnGroups")
965 """Column groups in the table."""
967 constraints: list[_ConstraintType] = Field(default_factory=list)
968 """Constraints on the table."""
970 indexes: list[Index] = Field(default_factory=list)
971 """Indexes on the table."""
973 @field_validator("columns", mode="after")
974 @classmethod
975 def check_unique_column_names(cls, columns: list[Column]) -> list[Column]:
976 """Check that column names are unique.
978 Parameters
979 ----------
980 columns
981 The columns to check.
983 Returns
984 -------
985 `list` [ `Column` ]
986 The columns if they are unique.
988 Raises
989 ------
990 ValueError
991 Raised if column names are not unique.
992 """
993 if len(columns) != len(set(column.name for column in columns)):
994 raise ValueError("Column names must be unique")
995 return columns
997 @model_validator(mode="after")
998 def check_tap_table_index(self, info: ValidationInfo) -> Table:
999 """Check that the table has a TAP table index.
1001 Parameters
1002 ----------
1003 info
1004 Validation context used to determine if the check is enabled.
1006 Returns
1007 -------
1008 `Table`
1009 The table being validated.
1011 Raises
1012 ------
1013 ValueError
1014 Raised If the table is missing a TAP table index.
1015 """
1016 context = info.context
1017 if not context or not context.get("check_tap_table_indexes", False):
1018 return self
1019 if self.tap_table_index is None:
1020 raise ValueError("Table is missing a TAP table index")
1021 return self
1023 @model_validator(mode="after")
1024 def check_tap_principal(self, info: ValidationInfo) -> Table:
1025 """Check that at least one column is flagged as 'principal' for TAP
1026 purposes.
1028 Parameters
1029 ----------
1030 info
1031 Validation context used to determine if the check is enabled.
1033 Returns
1034 -------
1035 `Table`
1036 The table being validated.
1038 Raises
1039 ------
1040 ValueError
1041 Raised if the table is missing a column flagged as 'principal'.
1042 """
1043 context = info.context
1044 if not context or not context.get("check_tap_principal", False):
1045 return self
1046 for col in self.columns:
1047 if col.tap_principal == 1:
1048 return self
1049 raise ValueError(f"Table '{self.name}' is missing at least one column designated as 'tap:principal'")
1051 def _find_column_by_id(self, id: str) -> Column:
1052 """Find a column by ID.
1054 Parameters
1055 ----------
1056 id
1057 The ID of the column to find.
1059 Returns
1060 -------
1061 `Column`
1062 The column with the given ID.
1064 Raises
1065 ------
1066 ValueError
1067 Raised if the column is not found.
1068 """
1069 for column in self.columns:
1070 if column.id == id:
1071 return column
1072 raise KeyError(f"Column '{id}' not found in table '{self.name}'")
1074 def _find_column_by_name(self, name: str) -> Column:
1075 for column in self.columns:
1076 if column.name == name:
1077 return column
1078 raise KeyError(f"Column '{name}' not found in table '{self.name}'")
1080 @model_validator(mode="after")
1081 def dereference_column_groups(self: Table) -> Table:
1082 """Dereference columns in column groups.
1084 Returns
1085 -------
1086 `Table`
1087 The table with dereferenced column groups.
1088 """
1089 for group in self.column_groups:
1090 group.table = self
1091 group._dereference_columns()
1092 return self
1094 @field_serializer("columns")
1095 def _serialize_columns(self, columns: list[Column]) -> list[dict[str, Any]]:
1096 """Serialize only non-resource columns."""
1097 return [
1098 col.model_dump(
1099 by_alias=True,
1100 exclude_none=True,
1101 exclude_defaults=True,
1102 )
1103 for col in columns
1104 if not col._is_resource_ref
1105 ]
1108class SchemaVersion(BaseModel):
1109 """Schema version model."""
1111 current: str
1112 """The current version of the schema."""
1114 compatible: list[str] = Field(default_factory=list)
1115 """The compatible versions of the schema."""
1117 read_compatible: list[str] = Field(default_factory=list)
1118 """The read compatible versions of the schema."""
1121class SchemaIdVisitor:
1122 """Visit a schema and build the map of IDs to objects.
1124 Notes
1125 -----
1126 Duplicates are added to a set when they are encountered, which can be
1127 accessed via the ``duplicates`` attribute. The presence of duplicates will
1128 not throw an error. Only the first object with a given ID will be added to
1129 the map, but this should not matter, since a ``ValidationError`` will be
1130 thrown by the ``model_validator`` method if any duplicates are found in the
1131 schema.
1132 """
1134 def __init__(self) -> None:
1135 """Create a new SchemaVisitor."""
1136 self.schema: Schema | None = None
1137 self.duplicates: set[str] = set()
1139 def add(self, obj: BaseObject) -> None:
1140 """Add an object to the ID map.
1142 Parameters
1143 ----------
1144 obj
1145 The object to add to the ID map.
1146 """
1147 if hasattr(obj, "id"): 1147 ↛ exitline 1147 didn't return from function 'add' because the condition on line 1147 was always true
1148 obj_id = getattr(obj, "id")
1149 if self.schema is not None: 1149 ↛ exitline 1149 didn't return from function 'add' because the condition on line 1149 was always true
1150 if obj_id in self.schema._id_map: 1150 ↛ 1151line 1150 didn't jump to line 1151 because the condition on line 1150 was never true
1151 self.duplicates.add(obj_id)
1152 else:
1153 self.schema._id_map[obj_id] = obj
1155 def visit_schema(self, schema: Schema) -> None:
1156 """Visit the objects in a schema and build the ID map.
1158 Parameters
1159 ----------
1160 schema
1161 The schema object to visit.
1163 Notes
1164 -----
1165 This will set an internal variable pointing to the schema object.
1166 """
1167 self.schema = schema
1168 self.duplicates.clear()
1169 self.add(self.schema)
1170 for table in self.schema.tables:
1171 self.visit_table(table)
1173 def visit_table(self, table: Table) -> None:
1174 """Visit a table object.
1176 Parameters
1177 ----------
1178 table
1179 The table object to visit.
1180 """
1181 self.add(table)
1182 for column in table.columns:
1183 self.visit_column(column)
1184 for constraint in table.constraints:
1185 self.visit_constraint(constraint)
1187 def visit_column(self, column: Column) -> None:
1188 """Visit a column object.
1190 Parameters
1191 ----------
1192 column
1193 The column object to visit.
1194 """
1195 self.add(column)
1197 def visit_constraint(self, constraint: Constraint) -> None:
1198 """Visit a constraint object.
1200 Parameters
1201 ----------
1202 constraint
1203 The constraint object to visit.
1204 """
1205 self.add(constraint)
1208T = TypeVar("T", bound=BaseObject)
1211def _strip_ids(data: Any) -> Any:
1212 """Recursively strip '@id' fields from a dictionary or list.
1214 Parameters
1215 ----------
1216 data
1217 The data to strip IDs from, which can be a dictionary, list, or any
1218 other type. Other types will be returned unchanged.
1219 """
1220 if isinstance(data, dict):
1221 data.pop("@id", None)
1222 for k, v in data.items():
1223 data[k] = _strip_ids(v)
1224 return data
1225 elif isinstance(data, list):
1226 return [_strip_ids(item) for item in data]
1227 else:
1228 return data
1231def _append_error(
1232 errors: list[InitErrorDetails],
1233 loc: tuple,
1234 input_value: Any,
1235 error_message: str,
1236 error_type: str = "value_error",
1237) -> None:
1238 """Append an error to the errors list.
1240 Parameters
1241 ----------
1242 errors : list[InitErrorDetails]
1243 The list of errors to append to.
1244 loc : tuple
1245 The location of the error in the schema.
1246 input_value : Any
1247 The input value that caused the error.
1248 error_message : str
1249 The error message to include in the context.
1250 """
1251 errors.append(
1252 {
1253 "type": error_type,
1254 "loc": loc,
1255 "input": input_value,
1256 "ctx": {"error": error_message},
1257 }
1258 )
1261class Resource(BaseModel):
1262 """A resource definition referencing an external schema."""
1264 uri: str = Field(..., description="Resource URI or path")
1265 """URI of the schema resource which may be a local path, ``resource://``,
1266 or remote URL."""
1269class Schema(BaseObject, Generic[T]):
1270 """Database schema model.
1272 This represents a database schema, which contains one or more tables.
1273 """
1275 version: SchemaVersion | str | None = None
1276 """The version of the schema."""
1278 resources: dict[str, Resource] = Field(default_factory=dict)
1279 """External resources referenced by this schema."""
1281 tables: Sequence[Table]
1282 """The tables in the schema."""
1284 _id_map: dict[str, Any] = PrivateAttr(default_factory=dict)
1285 """Map of IDs to objects."""
1287 _resource_map: dict[str, Schema] = PrivateAttr(default_factory=dict)
1288 """Map of resource names to loaded schemas."""
1290 @model_validator(mode="after")
1291 def _load_resources(self: Schema, info: ValidationInfo) -> Schema:
1292 """Load external resources referenced by this schema into an internal
1293 mapping of resource names to their `Schema` objects.
1295 Returns
1296 -------
1297 `Schema`
1298 The schema being validated.
1300 Raises
1301 ------
1302 ValueError
1303 Raised if a resource cannot be loaded.
1304 """
1305 if info.context:
1306 context = info.context.copy()
1307 # Ignore this flag for loading the resources themselves
1308 context.pop("dereference_resources", None)
1309 else:
1310 context = {}
1312 # Get the base URI for resolving relative resource paths from the
1313 # validation context, if available.
1314 resource_path = context.pop("resource_path", None)
1315 base_uri = None
1316 if resource_path is not None:
1317 base_uri = resource_path.parent()
1319 for resource_name, resource in self.resources.items():
1320 uri = resource.uri
1322 # Apply the base URI to the resource URI, if available.
1323 if base_uri is not None:
1324 orig_uri = uri
1325 uri = base_uri.join(uri, forceDirectory=False)
1326 if uri != orig_uri: 1326 ↛ 1335line 1326 didn't jump to line 1335 because the condition on line 1326 was always true
1327 logger.info(
1328 "Resolved relative URI '%s' for resource '%s' to '%s' using base URI '%s'",
1329 resource.uri,
1330 resource_name,
1331 uri,
1332 base_uri,
1333 )
1335 try:
1336 loaded_schema = Schema.from_uri(uri, context=context)
1337 self._resource_map[resource_name] = loaded_schema
1338 logger.debug("Loaded resource '%s' from URI '%s'", resource_name, uri)
1339 except Exception as e:
1340 raise ValueError(f"Failed to load resource '{resource_name}' from URI '{uri}': {e}") from e
1341 return self
1343 def _find_table_by_name(self, name: str) -> Table:
1344 """Find a table by name.
1346 Parameters
1347 ----------
1348 name
1349 The name of the table to find.
1351 Returns
1352 -------
1353 `Table`
1354 The table with the given name.
1356 Raises
1357 ------
1358 KeyError
1359 Raised if the table is not found.
1360 """
1361 for table in self.tables:
1362 if table.name == name:
1363 return table
1364 raise KeyError(f"Table '{name}' not found in schema '{self.name}'")
1366 @model_validator(mode="after")
1367 def _dereference_resource_columns(self: Schema, info: ValidationInfo) -> Schema:
1368 """Dereference columns from external resources and add them to the
1369 tables in this schema.
1370 """
1371 context = info.context
1372 column_ref_index_increment: int | None = None
1373 dereference_resources = False
1374 if context is not None:
1375 dereference_resources = context.get("dereference_resources", False)
1376 column_ref_index_increment = context.get("column_ref_index_increment", None)
1378 for table in self.tables:
1379 if column_refs := table.column_refs:
1380 for resource_name, tables in column_refs.items():
1381 resource_schema = self._resource_map.get(resource_name)
1382 if resource_schema is None:
1383 raise ValueError(f"Schema resource '{resource_name}' was not found in resources")
1384 self._process_column_refs(
1385 table,
1386 tables,
1387 resource_schema,
1388 dereference_resources,
1389 column_ref_index_increment,
1390 )
1391 if dereference_resources and len(table.column_refs) > 0:
1392 # Clear column refs in table if fully dereferencing
1393 logger.debug(
1394 "Clearing columnRefs in table '%s' after dereferencing resource columns",
1395 table.name,
1396 )
1397 table.column_refs = {}
1398 return self
1400 @classmethod
1401 def _process_column_refs(
1402 cls,
1403 table: Table,
1404 ref_tables: ResourceTableMap,
1405 resource_schema: Schema,
1406 dereference_resources: bool = False,
1407 column_ref_index_increment: int | None = None,
1408 ) -> None:
1409 """Process column references from an external resource and add them
1410 to the given table as columns.
1411 """
1412 current_column_index = column_ref_index_increment if column_ref_index_increment is not None else -1
1414 for table_name, columns in ref_tables.items():
1415 try:
1416 resource_table = resource_schema._find_table_by_name(table_name)
1417 except KeyError as e:
1418 raise ValueError(
1419 f"Table '{table_name}' not found in resource '{resource_schema.name}'"
1420 ) from e
1421 for local_column_name, column_ref in columns.items():
1422 if column_ref is not None and column_ref.ref_name is not None:
1423 # Use specified ref_name
1424 ref_column_name = column_ref.ref_name
1425 else:
1426 # Use the local column name if no ref_name
1427 # specified
1428 ref_column_name = local_column_name
1430 # Check if referenced column exists in resource
1431 try:
1432 base_column = resource_table._find_column_by_name(ref_column_name)
1433 except KeyError:
1434 # The ref_name is specified but column is not
1435 # found
1436 if column_ref is not None and column_ref.ref_name is not None:
1437 raise ValueError(
1438 f"Column '{ref_column_name}' not found in table '{table_name}' "
1439 f"from resource '{resource_schema.name}'"
1440 )
1441 # The ref_name is not specified and the local
1442 # column name is not found
1443 raise ValueError(
1444 f"Column '{local_column_name}' not found in table '{table_name}' "
1445 f"from resource '{resource_schema.name}' and no ref_name provided"
1446 )
1448 # Create a copy of the base column
1449 column_copy = base_column.model_copy()
1451 # Set the local name (key from the mapping)
1452 column_copy.name = local_column_name
1454 if not dereference_resources:
1455 # Flag the column as a resource reference so it will not be
1456 # written out during serialization
1457 column_copy._is_resource_ref = True
1459 # Apply overrides to the referenced column definition
1460 overrides = column_ref.overrides if column_ref is not None else None
1461 if overrides is not None:
1462 column_copy._update_from_overrides(overrides)
1464 # Manually set the ID of the copied column as ID generation has
1465 # already occurred by now
1466 column_copy.id = f"{table.id}.{local_column_name}"
1468 # Apply automatic assignment of 'tap:column_index', if enabled
1469 if column_ref_index_increment is not None:
1470 if (not overrides) or (overrides.tap_column_index is None):
1471 column_copy.tap_column_index = current_column_index
1472 current_column_index += column_ref_index_increment
1473 logger.debug(
1474 "Automatically assigned 'tap:column_index' %s to "
1475 "column '%s' in table '%s' from resource '%s'",
1476 column_copy.tap_column_index,
1477 local_column_name,
1478 table_name,
1479 resource_schema.name,
1480 )
1481 else:
1482 # Skip automatic assignment of 'tap:column_index' if it
1483 # is already overridden
1484 logger.debug(
1485 "Skipping automatic assignment of 'tap:column_index' for column "
1486 "'%s' in table '%s' from resource '%s' as it is already overridden to %s",
1487 local_column_name,
1488 table_name,
1489 resource_schema.name,
1490 column_copy.tap_column_index,
1491 )
1492 table.columns.append(column_copy)
1493 logger.debug(
1494 "Dereferenced column '%s' from table '%s' in resource '%s' into table '%s'",
1495 local_column_name,
1496 table_name,
1497 resource_schema.name,
1498 table.name,
1499 )
1501 @model_validator(mode="before")
1502 @classmethod
1503 def generate_ids(cls, values: dict[str, Any], info: ValidationInfo) -> dict[str, Any]:
1504 """Generate IDs for objects that do not have them.
1506 Parameters
1507 ----------
1508 values
1509 The values of the schema.
1510 info
1511 Validation context used to determine if ID generation is enabled.
1513 Returns
1514 -------
1515 `dict` [ `str`, `Any` ]
1516 The values of the schema with generated IDs.
1517 """
1518 context = info.context
1519 if not context or not context.get("id_generation", False):
1520 logger.debug("Skipping ID generation")
1521 return values
1522 schema_name = values["name"]
1523 if "@id" not in values:
1524 values["@id"] = f"#{schema_name}"
1525 logger.debug("Generated ID '%s' for schema '%s'", values["@id"], schema_name)
1526 if "tables" in values: 1526 ↛ 1559line 1526 didn't jump to line 1559 because the condition on line 1526 was always true
1527 for table in values["tables"]:
1528 if "@id" not in table:
1529 table["@id"] = f"#{table['name']}"
1530 logger.debug("Generated ID '%s' for table '%s'", table["@id"], table["name"])
1531 if "columns" in table:
1532 for column in table["columns"]:
1533 if "@id" not in column:
1534 column["@id"] = f"#{table['name']}.{column['name']}"
1535 logger.debug("Generated ID '%s' for column '%s'", column["@id"], column["name"])
1536 if "columnGroups" in table:
1537 for column_group in table["columnGroups"]:
1538 if "@id" not in column_group:
1539 column_group["@id"] = f"#{table['name']}.{column_group['name']}"
1540 logger.debug(
1541 "Generated ID '%s' for column group '%s'",
1542 column_group["@id"],
1543 column_group["name"],
1544 )
1545 if "constraints" in table:
1546 for constraint in table["constraints"]:
1547 if "@id" not in constraint:
1548 constraint["@id"] = f"#{constraint['name']}"
1549 logger.debug(
1550 "Generated ID '%s' for constraint '%s'",
1551 constraint["@id"],
1552 constraint["name"],
1553 )
1554 if "indexes" in table:
1555 for index in table["indexes"]:
1556 if "@id" not in index:
1557 index["@id"] = f"#{index['name']}"
1558 logger.debug("Generated ID '%s' for index '%s'", index["@id"], index["name"])
1559 return values
1561 @field_validator("tables", mode="after")
1562 @classmethod
1563 def check_unique_table_names(cls, tables: list[Table]) -> list[Table]:
1564 """Check that table names are unique.
1566 Parameters
1567 ----------
1568 tables
1569 The tables to check.
1571 Returns
1572 -------
1573 `list` [ `Table` ]
1574 The tables if they are unique.
1576 Raises
1577 ------
1578 ValueError
1579 Raised if table names are not unique.
1580 """
1581 if len(tables) != len(set(table.name for table in tables)):
1582 raise ValueError("Table names must be unique")
1583 return tables
1585 @model_validator(mode="after")
1586 def check_tap_table_indexes(self, info: ValidationInfo) -> Schema:
1587 """Check that the TAP table indexes are unique.
1589 Parameters
1590 ----------
1591 info
1592 The validation context used to determine if the check is enabled.
1594 Returns
1595 -------
1596 `Schema`
1597 The schema being validated.
1598 """
1599 context = info.context
1600 if not context or not context.get("check_tap_table_indexes", False):
1601 return self
1602 table_indicies = set()
1603 for table in self.tables:
1604 table_index = table.tap_table_index
1605 if table_index is not None: 1605 ↛ 1603line 1605 didn't jump to line 1603 because the condition on line 1605 was always true
1606 if table_index in table_indicies:
1607 raise ValueError(f"Duplicate 'tap:table_index' value {table_index} found in schema")
1608 table_indicies.add(table_index)
1609 return self
1611 @model_validator(mode="after")
1612 def check_unique_constraint_names(self: Schema) -> Schema:
1613 """Check for duplicate constraint names in the schema.
1615 Returns
1616 -------
1617 `Schema`
1618 The schema being validated.
1620 Raises
1621 ------
1622 ValueError
1623 Raised if duplicate constraint names are found in the schema.
1624 """
1625 constraint_names = set()
1626 duplicate_names = []
1628 for table in self.tables:
1629 for constraint in table.constraints:
1630 constraint_name = constraint.name
1631 if constraint_name in constraint_names:
1632 duplicate_names.append(constraint_name)
1633 else:
1634 constraint_names.add(constraint_name)
1636 if duplicate_names:
1637 raise ValueError(f"Duplicate constraint names found in schema: {duplicate_names}")
1639 return self
1641 @model_validator(mode="after")
1642 def check_unique_index_names(self: Schema) -> Schema:
1643 """Check for duplicate index names in the schema.
1645 Returns
1646 -------
1647 `Schema`
1648 The schema being validated.
1650 Raises
1651 ------
1652 ValueError
1653 Raised if duplicate index names are found in the schema.
1654 """
1655 index_names = set()
1656 duplicate_names = []
1658 for table in self.tables:
1659 for index in table.indexes:
1660 index_name = index.name
1661 if index_name in index_names:
1662 duplicate_names.append(index_name)
1663 else:
1664 index_names.add(index_name)
1666 if duplicate_names:
1667 raise ValueError(f"Duplicate index names found in schema: {duplicate_names}")
1669 return self
1671 @model_validator(mode="after")
1672 def create_id_map(self: Schema) -> Schema:
1673 """Create a map of IDs to objects.
1675 Returns
1676 -------
1677 `Schema`
1678 The schema with the ID map created.
1680 Raises
1681 ------
1682 ValueError
1683 Raised if duplicate identifiers are found in the schema.
1684 """
1685 if self._id_map: 1685 ↛ 1686line 1685 didn't jump to line 1686 because the condition on line 1685 was never true
1686 logger.debug("Ignoring call to create_id_map() - ID map was already populated")
1687 return self
1688 visitor: SchemaIdVisitor = SchemaIdVisitor()
1689 visitor.visit_schema(self)
1690 if len(visitor.duplicates): 1690 ↛ 1691line 1690 didn't jump to line 1691 because the condition on line 1690 was never true
1691 raise ValueError(
1692 "Duplicate IDs found in schema:\n " + "\n ".join(visitor.duplicates) + "\n"
1693 )
1694 logger.debug("Created ID map with %d entries", len(self._id_map))
1695 return self
1697 def _validate_column_id(
1698 self: Schema,
1699 column_id: str,
1700 loc: tuple,
1701 errors: list[InitErrorDetails],
1702 ) -> None:
1703 """Validate a column ID from a constraint and append errors if invalid.
1705 Parameters
1706 ----------
1707 schema : Schema
1708 The schema being validated.
1709 column_id : str
1710 The column ID to validate.
1711 loc : tuple
1712 The location of the error in the schema.
1713 errors : list[InitErrorDetails]
1714 The list of errors to append to.
1715 """
1716 if column_id not in self:
1717 _append_error(
1718 errors,
1719 loc,
1720 column_id,
1721 f"Column ID '{column_id}' not found in schema",
1722 )
1723 elif not isinstance(self[column_id], Column):
1724 _append_error(
1725 errors,
1726 loc,
1727 column_id,
1728 f"ID '{column_id}' does not refer to a Column object",
1729 )
1731 def _validate_foreign_key_column(
1732 self: Schema,
1733 column_id: str,
1734 table: Table,
1735 loc: tuple,
1736 errors: list[InitErrorDetails],
1737 ) -> None:
1738 """Validate a foreign key column ID from a constraint and append errors
1739 if invalid.
1741 Parameters
1742 ----------
1743 schema : Schema
1744 The schema being validated.
1745 column_id : str
1746 The foreign key column ID to validate.
1747 loc : tuple
1748 The location of the error in the schema.
1749 errors : list[InitErrorDetails]
1750 The list of errors to append to.
1751 """
1752 try:
1753 table._find_column_by_id(column_id)
1754 except KeyError:
1755 _append_error(
1756 errors,
1757 loc,
1758 column_id,
1759 f"Column '{column_id}' not found in table '{table.name}'",
1760 )
1762 @model_validator(mode="after")
1763 def check_constraints(self: Schema) -> Schema:
1764 """Check constraint objects for validity. This needs to be deferred
1765 until after the schema is fully loaded and the ID map is created.
1767 Raises
1768 ------
1769 pydantic.ValidationError
1770 Raised if any constraints are invalid.
1772 Returns
1773 -------
1774 `Schema`
1775 The schema being validated.
1776 """
1777 errors: list[InitErrorDetails] = []
1779 for table_index, table in enumerate(self.tables):
1780 for constraint_index, constraint in enumerate(table.constraints):
1781 column_ids: list[str] = []
1782 referenced_column_ids: list[str] = []
1784 if isinstance(constraint, ForeignKeyConstraint):
1785 column_ids += constraint.columns
1786 referenced_column_ids += constraint.referenced_columns
1787 elif isinstance(constraint, UniqueConstraint):
1788 column_ids += constraint.columns
1789 # No extra checks are required on CheckConstraint objects.
1791 # Validate the foreign key columns
1792 for column_id in column_ids:
1793 self._validate_column_id(
1794 column_id,
1795 (
1796 "tables",
1797 table_index,
1798 "constraints",
1799 constraint_index,
1800 "columns",
1801 column_id,
1802 ),
1803 errors,
1804 )
1805 # Check that the foreign key column is within the source
1806 # table.
1807 self._validate_foreign_key_column(
1808 column_id,
1809 table,
1810 (
1811 "tables",
1812 table_index,
1813 "constraints",
1814 constraint_index,
1815 "columns",
1816 column_id,
1817 ),
1818 errors,
1819 )
1821 # Validate the primary key (reference) columns
1822 for referenced_column_id in referenced_column_ids:
1823 self._validate_column_id(
1824 referenced_column_id,
1825 (
1826 "tables",
1827 table_index,
1828 "constraints",
1829 constraint_index,
1830 "referenced_columns",
1831 referenced_column_id,
1832 ),
1833 errors,
1834 )
1836 if errors:
1837 raise ValidationError.from_exception_data("Schema validation failed", errors)
1839 return self
1841 def __getitem__(self, id: str) -> BaseObject:
1842 """Get an object by its ID.
1844 Parameters
1845 ----------
1846 id
1847 The ID of the object to get.
1849 Raises
1850 ------
1851 KeyError
1852 Raised if the object with the given ID is not found in the schema.
1853 """
1854 if id not in self:
1855 raise KeyError(f"Object with ID '{id}' not found in schema")
1856 return self._id_map[id]
1858 def __contains__(self, id: str) -> bool:
1859 """Check if an object with the given ID is in the schema.
1861 Parameters
1862 ----------
1863 id
1864 The ID of the object to check.
1865 """
1866 return id in self._id_map
1868 def find_object_by_id(self, id: str, obj_type: type[T]) -> T:
1869 """Find an object with the given type by its ID.
1871 Parameters
1872 ----------
1873 id
1874 The ID of the object to find.
1875 obj_type
1876 The type of the object to find.
1878 Returns
1879 -------
1880 BaseObject
1881 The object with the given ID and type.
1883 Raises
1884 ------
1885 KeyError
1886 If the object with the given ID is not found in the schema.
1887 TypeError
1888 If the object that is found does not have the right type.
1890 Notes
1891 -----
1892 The actual return type is the user-specified argument ``T``, which is
1893 expected to be a subclass of `BaseObject`.
1894 """
1895 obj = self[id]
1896 if not isinstance(obj, obj_type):
1897 raise TypeError(f"Object with ID '{id}' is not of type '{obj_type.__name__}'")
1898 return obj
1900 def get_table_by_column(self, column: Column) -> Table:
1901 """Find the table that contains a column.
1903 Parameters
1904 ----------
1905 column
1906 The column to find.
1908 Returns
1909 -------
1910 `Table`
1911 The table that contains the column.
1913 Raises
1914 ------
1915 ValueError
1916 If the column is not found in any table.
1917 """
1918 for table in self.tables:
1919 if column in table.columns:
1920 return table
1921 raise ValueError(f"Column '{column.name}' not found in any table")
1923 @classmethod
1924 def from_uri(cls, resource_path: ResourcePathExpression, context: dict[str, Any] = {}) -> Schema:
1925 """Load a `Schema` from a string representing a ``ResourcePath``.
1927 Parameters
1928 ----------
1929 resource_path
1930 The ``ResourcePath`` pointing to a YAML file.
1931 context
1932 Pydantic context to be used in validation.
1934 Returns
1935 -------
1936 `str`
1937 The ID of the object.
1939 Raises
1940 ------
1941 yaml.YAMLError
1942 Raised if there is an error loading the YAML data.
1943 ValueError
1944 Raised if there is an error reading the resource.
1945 pydantic.ValidationError
1946 Raised if the schema fails validation.
1947 """
1948 try:
1949 rp = ResourcePath(resource_path, forceAbsolute=False, forceDirectory=False)
1950 rp_data = rp.read()
1951 except Exception as e:
1952 raise ValueError(f"Error reading resource from '{resource_path}' : {e}") from e
1953 yaml_data = yaml.safe_load(rp_data)
1954 context = dict(context)
1955 # Append the resource path to the context for resolving resource URLs.
1956 context["resource_path"] = rp
1957 return Schema.model_validate(yaml_data, context=context)
1959 @classmethod
1960 def from_stream(cls, source: IO[str], context: dict[str, Any] = {}) -> Schema:
1961 """Load a `Schema` from a file stream which should contain YAML data.
1963 Parameters
1964 ----------
1965 source
1966 The file stream to read from.
1967 context
1968 Pydantic context to be used in validation.
1970 Returns
1971 -------
1972 `Schema`
1973 The Felis schema loaded from the stream.
1975 Raises
1976 ------
1977 yaml.YAMLError
1978 Raised if there is an error loading the YAML file.
1979 pydantic.ValidationError
1980 Raised if the schema fails validation.
1981 """
1982 logger.debug("Loading schema from: '%s'", source)
1983 yaml_data = yaml.safe_load(source)
1984 return Schema.model_validate(yaml_data, context=context)
1986 def _model_dump(self, strip_ids: bool = False, sort_columns: bool = False) -> dict[str, Any]:
1987 """Dump the schema as a dictionary with some default arguments
1988 applied.
1990 Parameters
1991 ----------
1992 strip_ids
1993 Whether to strip the IDs from the dumped data. Defaults to `False`.
1994 sort_columns
1995 Whether to sort columns alphabetically by name. Defaults to
1996 `False`.
1998 Returns
1999 -------
2000 `dict` [ `str`, `Any` ]
2001 The dumped schema data as a dictionary.
2002 """
2003 data = self.model_dump(by_alias=True, exclude_none=True, exclude_defaults=True)
2004 if strip_ids:
2005 data = _strip_ids(data)
2006 if sort_columns:
2007 for table in data.get("tables", []):
2008 table["columns"] = sorted(table.get("columns", []), key=itemgetter("name"))
2009 return data
2011 def dump_yaml(
2012 self, stream: IO[str] = sys.stdout, strip_ids: bool = False, sort_columns: bool = False
2013 ) -> None:
2014 """Pretty print the schema as YAML.
2016 Parameters
2017 ----------
2018 stream
2019 The stream to write the YAML data to.
2020 strip_ids
2021 Whether to strip the IDs from the dumped data. Defaults to `False`.
2022 sort_columns
2023 Whether to sort columns alphabetically by name. Defaults to
2024 `False`.
2025 """
2026 data = self._model_dump(strip_ids=strip_ids, sort_columns=sort_columns)
2027 yaml.safe_dump(
2028 data,
2029 stream,
2030 default_flow_style=False,
2031 sort_keys=False,
2032 )
2034 def dump_json(
2035 self, stream: IO[str] = sys.stdout, strip_ids: bool = False, sort_columns: bool = False
2036 ) -> None:
2037 """Pretty print the schema as JSON.
2039 Parameters
2040 ----------
2041 stream
2042 The stream to write the JSON data to.
2043 strip_ids
2044 Whether to strip the IDs from the dumped data. Defaults to `False`.
2045 sort_columns
2046 Whether to sort columns alphabetically by name. Defaults to
2047 `False`.
2048 """
2049 data = self._model_dump(strip_ids=strip_ids, sort_columns=sort_columns)
2050 json.dump(
2051 data,
2052 stream,
2053 indent=4,
2054 sort_keys=False,
2055 )