Coverage for python/felis/datamodel.py: 31%

1"""Define Pydantic data models for Felis."""

3# This file is part of felis.

5# Developed for the LSST Data Management System.

6# This product includes software developed by the LSST Project

7# (https://www.lsst.org).

8# See the COPYRIGHT file at the top-level directory of this distribution

9# for details of code ownership.

10#

11# This program is free software: you can redistribute it and/or modify

12# it under the terms of the GNU General Public License as published by

13# the Free Software Foundation, either version 3 of the License, or

14# (at your option) any later version.

15#

16# This program is distributed in the hope that it will be useful,

17# but WITHOUT ANY WARRANTY; without even the implied warranty of

18# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

19# GNU General Public License for more details.

20#

21# You should have received a copy of the GNU General Public License

22# along with this program. If not, see <https://www.gnu.org/licenses/>.

24from __future__ import annotations

26import json

27import logging

28import sys

29from collections.abc import Sequence

30from enum import StrEnum, auto

31from typing import IO, Annotated, Any, Generic, Literal, TypeAlias, TypeVar

33import yaml

34from astropy import units as units # type: ignore

35from astropy.io.votable import ucd # type: ignore

36from lsst.resources import ResourcePath, ResourcePathExpression

37from pydantic import (

38 BaseModel,

39 ConfigDict,

40 Field,

41 PrivateAttr,

42 ValidationError,

43 ValidationInfo,

44 field_serializer,

45 field_validator,

46 model_validator,

47)

48from pydantic_core import InitErrorDetails

50from .db._dialects import get_supported_dialects, string_to_typeengine

51from .db._sqltypes import get_type_func

52from .types import Boolean, Byte, Char, Double, FelisType, Float, Int, Long, Short, String, Text, Unicode

54logger = logging.getLogger(__name__)

56__all__ = (

57 "BaseObject",

58 "CheckConstraint",

59 "Column",

60 "ColumnOverrides",

61 "ColumnResourceRef",

62 "Constraint",

63 "DataType",

64 "ForeignKeyConstraint",

65 "Index",

66 "Resource",

67 "Schema",

68 "SchemaVersion",

69 "Table",

70 "UniqueConstraint",

71)

73CONFIG = ConfigDict(

74 populate_by_name=True, # Populate attributes by name.

75 extra="forbid", # Do not allow extra fields.

76 str_strip_whitespace=True, # Strip whitespace from string fields.

77 use_enum_values=False, # Do not use enum values during serialization.

78)

79"""Pydantic model configuration as described in:

80https://docs.pydantic.dev/2.0/api/config/#pydantic.config.ConfigDict

81"""

83DESCR_MIN_LENGTH = 3

84"""Minimum length for a description field."""

86DescriptionStr: TypeAlias = Annotated[str, Field(min_length=DESCR_MIN_LENGTH)]

87"""Type for a description, which must be three or more characters long."""

90class BaseObject(BaseModel):

91 """Base model.

93 All classes representing objects in the Felis data model should inherit

94 from this class.

95 """

97 model_config = CONFIG

98 """Pydantic model configuration."""

100 name: str

101 """Name of the database object."""

102

103 id: str = Field(alias="@id")

104 """Unique identifier of the database object."""

105

106 description: DescriptionStr | None = None

107 """Description of the database object."""

108

109 votable_utype: str | None = Field(None, alias="votable:utype")

110 """VOTable utype (usage-specific or unique type) of the object."""

111

112 @model_validator(mode="after")

113 def check_description(self, info: ValidationInfo) -> BaseObject:

114 """Check that the description is present if required.

115

116 Parameters

117 ----------

118 info

119 Validation context used to determine if the check is enabled.

120

121 Returns

122 -------

123 `BaseObject`

124 The object being validated.

125 """

126 context = info.context

127 if not context or not context.get("check_description", False):

128 return self

129 if self.description is None or self.description == "":

130 raise ValueError("Description is required and must be non-empty")

131 if len(self.description) < DESCR_MIN_LENGTH:

132 raise ValueError(f"Description must be at least {DESCR_MIN_LENGTH} characters long")

133 return self

134

135

136class DataType(StrEnum):

137 """``Enum`` representing the data types supported by Felis."""

138

139 boolean = auto()

140 byte = auto()

141 short = auto()

142 int = auto()

143 long = auto()

144 float = auto()

145 double = auto()

146 char = auto()

147 string = auto()

148 unicode = auto()

149 text = auto()

150 binary = auto()

151 timestamp = auto()

152

153

154def validate_ivoa_ucd(ivoa_ucd: str) -> str:

155 """Validate IVOA UCD values.

156

157 Parameters

158 ----------

159 ivoa_ucd

160 IVOA UCD value to check.

161

162 Returns

163 -------

164 `str`

165 The IVOA UCD value if it is valid.

166

167 Raises

168 ------

169 ValueError

170 If the IVOA UCD value is invalid.

171 """

172 if ivoa_ucd is not None:

173 try:

174 ucd.parse_ucd(ivoa_ucd, check_controlled_vocabulary=True, has_colon=";" in ivoa_ucd)

175 except ValueError as e:

176 raise ValueError(f"Invalid IVOA UCD: {e}")

177 return ivoa_ucd

178

179

180class Column(BaseObject):

181 """Column model."""

182

183 datatype: DataType

184 """Datatype of the column."""

185

186 length: int | None = Field(None, gt=0)

187 """Length of the column."""

188

189 precision: int | None = Field(None, ge=0)

190 """The numerical precision of the column.

191

192 For timestamps, this is the number of fractional digits retained in the

193 seconds field.

194 """

195

196 nullable: bool = True

197 """Whether the column can be ``NULL``."""

198

199 value: str | int | float | bool | None = None

200 """Default value of the column."""

201

202 autoincrement: bool | None = None

203 """Whether the column is autoincremented."""

204

205 ivoa_ucd: str | None = Field(None, alias="ivoa:ucd")

206 """IVOA UCD of the column."""

207

208 fits_tunit: str | None = Field(None, alias="fits:tunit")

209 """FITS TUNIT of the column."""

210

211 ivoa_unit: str | None = Field(None, alias="ivoa:unit")

212 """IVOA unit of the column."""

213

214 tap_column_index: int | None = Field(None, alias="tap:column_index")

215 """TAP_SCHEMA column index of the column."""

216

217 tap_principal: int | None = Field(0, alias="tap:principal", ge=0, le=1)

218 """Whether this is a TAP_SCHEMA principal column."""

219

220 votable_arraysize: int | str | None = Field(None, alias="votable:arraysize")

221 """VOTable arraysize of the column."""

222

223 tap_std: int | None = Field(0, alias="tap:std", ge=0, le=1)

224 """TAP_SCHEMA indication that this column is defined by an IVOA standard.

225 """

226

227 votable_xtype: str | None = Field(None, alias="votable:xtype")

228 """VOTable xtype (extended type) of the column."""

229

230 votable_datatype: str | None = Field(None, alias="votable:datatype")

231 """VOTable datatype of the column."""

232

233 mysql_datatype: str | None = Field(None, alias="mysql:datatype")

234 """MySQL datatype override on the column."""

235

236 postgresql_datatype: str | None = Field(None, alias="postgresql:datatype")

237 """PostgreSQL datatype override on the column."""

238

239 _is_resource_ref: bool = PrivateAttr(False)

240 """Whether this column is a resource reference column."""

241

242 @model_validator(mode="after")

243 def check_value(self) -> Column:

244 """Check that the default value is valid.

245

246 Returns

247 -------

248 `Column`

249 The column being validated.

250 """

251 if (value := self.value) is not None:

252 if value is not None and self.autoincrement is True:

253 raise ValueError("Column cannot have both a default value and be autoincremented")

254 felis_type = FelisType.felis_type(self.datatype)

255 if felis_type.is_numeric:

256 if felis_type in (Byte, Short, Int, Long) and not isinstance(value, int):

257 raise ValueError("Default value must be an int for integer type columns")

258 elif felis_type in (Float, Double) and not isinstance(value, float):

259 raise ValueError("Default value must be a decimal number for float and double columns")

260 elif felis_type in (String, Char, Unicode, Text):

261 if not isinstance(value, str):

262 raise ValueError("Default value must be a string for string columns")

263 if not len(value):

264 raise ValueError("Default value must be a non-empty string for string columns")

265 elif felis_type is Boolean and not isinstance(value, bool):

266 raise ValueError("Default value must be a boolean for boolean columns")

267 return self

268

269 @field_validator("ivoa_ucd")

270 @classmethod

271 def check_ivoa_ucd(cls, ivoa_ucd: str) -> str:

272 """Check that IVOA UCD values are valid.

273

274 Parameters

275 ----------

276 ivoa_ucd

277 IVOA UCD value to check.

278

279 Returns

280 -------

281 `str`

282 The IVOA UCD value if it is valid.

283 """

284 return validate_ivoa_ucd(ivoa_ucd)

285

286 @model_validator(mode="after")

287 def check_units(self) -> Column:

288 """Check that the ``fits:tunit`` or ``ivoa:unit`` field has valid

289 units according to astropy. Only one may be provided.

290

291 Returns

292 -------

293 `Column`

294 The column being validated.

295

296 Raises

297 ------

298 ValueError

299 Raised if both FITS and IVOA units are provided, or if the unit is

300 invalid.

301 """

302 fits_unit = self.fits_tunit

303 ivoa_unit = self.ivoa_unit

304

305 if fits_unit and ivoa_unit:

306 raise ValueError("Column cannot have both FITS and IVOA units")

307 unit = fits_unit or ivoa_unit

308

309 if unit is not None:

310 try:

311 units.Unit(unit)

312 except ValueError as e:

313 raise ValueError(f"Invalid unit: {e}")

314

315 return self

316

317 @model_validator(mode="before")

318 @classmethod

319 def check_length(cls, values: dict[str, Any]) -> dict[str, Any]:

320 """Check that a valid length is provided for sized types.

321

322 Parameters

323 ----------

324 values

325 Values of the column.

326

327 Returns

328 -------

329 `dict` [ `str`, `Any` ]

330 The values of the column.

331

332 Raises

333 ------

334 ValueError

335 Raised if a length is not provided for a sized type.

336 """

337 datatype = values.get("datatype")

338 if datatype is None:

339 # Skip this validation if datatype is not provided

340 return values

341 length = values.get("length")

342 felis_type = FelisType.felis_type(datatype)

343 if felis_type.is_sized and length is None:

344 raise ValueError(

345 f"Length must be provided for type '{datatype}'"

346 + (f" in column '{values['@id']}'" if "@id" in values else "")

347 )

348 elif not felis_type.is_sized and length is not None:

349 logger.warning(

350 f"The datatype '{datatype}' does not support a specified length"

351 + (f" in column '{values['@id']}'" if "@id" in values else "")

352 )

353 return values

354

355 @model_validator(mode="after")

356 def check_redundant_datatypes(self, info: ValidationInfo) -> Column:

357 """Check for redundant datatypes on columns.

358

359 Parameters

360 ----------

361 info

362 Validation context used to determine if the check is enabled.

363

364 Returns

365 -------

366 `Column`

367 The column being validated.

368

369 Raises

370 ------

371 ValueError

372 Raised if a datatype override is redundant.

373 """

374 context = info.context

375 if not context or not context.get("check_redundant_datatypes", False):

376 return self

377 if all(

378 getattr(self, f"{dialect}:datatype", None) is not None

379 for dialect in get_supported_dialects().keys()

380 ):

381 return self

382

383 datatype = self.datatype

384 length: int | None = self.length or None

385

386 datatype_func = get_type_func(datatype)

387 felis_type = FelisType.felis_type(datatype)

388 if felis_type.is_sized:

389 datatype_obj = datatype_func(length)

390 else:

391 datatype_obj = datatype_func()

392

393 for dialect_name, dialect in get_supported_dialects().items():

394 db_annotation = f"{dialect_name}_datatype"

395 if datatype_string := self.model_dump().get(db_annotation):

396 db_datatype_obj = string_to_typeengine(datatype_string, dialect, length)

397 if datatype_obj.compile(dialect) == db_datatype_obj.compile(dialect):

398 raise ValueError(

399 "'{}: {}' is a redundant override of 'datatype: {}' in column '{}'{}".format(

400 db_annotation,

401 datatype_string,

402 self.datatype,

403 self.id,

404 "" if length is None else f" with length {length}",

405 )

406 )

407 else:

408 logger.debug(

409 f"Type override of 'datatype: {self.datatype}' "

410 f"with '{db_annotation}: {datatype_string}' in column '{self.id}' "

411 f"compiled to '{datatype_obj.compile(dialect)}' and "

412 f"'{db_datatype_obj.compile(dialect)}'"

413 )

414 return self

415

416 @model_validator(mode="after")

417 def check_precision(self) -> Column:

418 """Check that precision is only valid for timestamp columns.

419

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

428

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.

434

435 Parameters

436 ----------

437 values

438 Values of the column.

439 info

440 Validation context used to determine if the check is enabled.

441

442 Returns

443 -------

444 `dict` [ `str`, `Any` ]

445 The values of the column.

446

447 Notes

448 -----

449 Following the IVOA VOTable standard, an ``arraysize`` of 1 should not

450 be used.

451 """

452 if values.get("name", None) is None or values.get("datatype", None) is None:

453 # Skip bad column data that will not validate

454 return values

455 context = info.context if info.context else {}

456 arraysize = values.get("votable:arraysize", None)

457 if arraysize is None:

458 length = values.get("length", None)

459 datatype = values.get("datatype")

460 if length is not None and length > 1:

461 # Following the IVOA standard, arraysize of 1 is disallowed

462 if datatype == "char":

463 arraysize = str(length)

464 elif datatype in ("string", "unicode", "binary"):

465 if context.get("force_unbounded_arraysize", False):

466 arraysize = "*"

467 logger.debug(

468 f"Forced VOTable's 'arraysize' to '*' on column '{values['name']}' with datatype "

469 + f"'{values['datatype']}' and length '{length}'"

470 )

471 else:

472 arraysize = f"{length}*"

473 elif datatype in ("timestamp", "text"):

474 arraysize = "*"

475 if arraysize is not None:

476 values["votable:arraysize"] = arraysize

477 logger.debug(

478 f"Set default 'votable:arraysize' to '{arraysize}' on column '{values['name']}'"

479 + f" with datatype '{values['datatype']}' and length '{values.get('length', None)}'"

480 )

481 else:

482 logger.debug(f"Using existing 'votable:arraysize' of '{arraysize}' on column '{values['name']}'")

483 if isinstance(values["votable:arraysize"], int):

484 logger.warning(

485 f"Usage of an integer value for 'votable:arraysize' in column '{values['name']}' is "

486 + "deprecated"

487 )

488 values["votable:arraysize"] = str(arraysize)

489 return values

490

491 @field_serializer("datatype")

492 def serialize_datatype(self, value: DataType) -> str:

493 """Convert `DataType` to string when serializing to JSON/YAML.

494

495 Parameters

496 ----------

497 value

498 The `DataType` value to serialize.

499

500 Returns

501 -------

502 `str`

503 The serialized `DataType` value.

504 """

505 return str(value)

506

507 @field_validator("datatype", mode="before")

508 @classmethod

509 def deserialize_datatype(cls, value: str) -> DataType:

510 """Convert string back into `DataType` when loading from JSON/YAML.

511

512 Parameters

513 ----------

514 value

515 The string value to deserialize.

516

517 Returns

518 -------

519 `DataType`

520 The deserialized `DataType` value.

521 """

522 return DataType(value)

523

524 @model_validator(mode="after")

525 def check_votable_xtype(self) -> Column:

526 """Set the default value for the ``votable_xtype`` field, which

527 corresponds to an Extended Datatype or ``xtype`` in the IVOA VOTable

528 standard.

529

530 Returns

531 -------

532 `Column`

533 The column being validated.

534

535 Notes

536 -----

537 This is currently only set automatically for the Felis ``timestamp``

538 datatype.

539 """

540 if self.datatype == DataType.timestamp and self.votable_xtype is None:

541 self.votable_xtype = "timestamp"

542 return self

543

544

545class Constraint(BaseObject):

546 """Table constraint model."""

547

548 deferrable: bool = False

549 """Whether this constraint will be declared as deferrable."""

550

551 initially: Literal["IMMEDIATE", "DEFERRED"] | None = None

552 """Value for ``INITIALLY`` clause; only used if `deferrable` is

553 `True`."""

554

555 @model_validator(mode="after")

556 def check_deferrable(self) -> Constraint:

557 """Check that the ``INITIALLY`` clause is only used if `deferrable` is

558 `True`.

559

560 Returns

561 -------

562 `Constraint`

563 The constraint being validated.

564 """

565 if self.initially is not None and not self.deferrable:

566 raise ValueError("INITIALLY clause can only be used if deferrable is True")

567 return self

568

569

570class CheckConstraint(Constraint):

571 """Table check constraint model."""

572

573 type: Literal["Check"] = Field("Check", alias="@type")

574 """Type of the constraint."""

575

576 expression: str

577 """Expression for the check constraint."""

578

579 @field_serializer("type")

580 def serialize_type(self, value: str) -> str:

581 """Ensure '@type' is included in serialized output.

582

583 Parameters

584 ----------

585 value

586 The value to serialize.

587

588 Returns

589 -------

590 `str`

591 The serialized value.

592 """

593 return value

594

595

596class UniqueConstraint(Constraint):

597 """Table unique constraint model."""

598

599 type: Literal["Unique"] = Field("Unique", alias="@type")

600 """Type of the constraint."""

601

602 columns: list[str]

603 """Columns in the unique constraint."""

604

605 @field_serializer("type")

606 def serialize_type(self, value: str) -> str:

607 """Ensure '@type' is included in serialized output.

608

609 Parameters

610 ----------

611 value

612 The value to serialize.

613

614 Returns

615 -------

616 `str`

617 The serialized value.

618 """

619 return value

620

621

622class ForeignKeyConstraint(Constraint):

623 """Table foreign key constraint model.

624

625 This constraint is used to define a foreign key relationship between two

626 tables in the schema. There must be at least one column in the

627 `columns` list, and at least one column in the `referenced_columns` list

628 or a validation error will be raised.

629

630 Notes

631 -----

632 These relationships will be reflected in the TAP_SCHEMA ``keys`` and

633 ``key_columns`` data.

634 """

635

636 type: Literal["ForeignKey"] = Field("ForeignKey", alias="@type")

637 """Type of the constraint."""

638

639 columns: list[str] = Field(min_length=1)

640 """The columns comprising the foreign key."""

641

642 referenced_columns: list[str] = Field(alias="referencedColumns", min_length=1)

643 """The columns referenced by the foreign key."""

644

645 on_delete: Literal["CASCADE", "SET NULL", "SET DEFAULT", "RESTRICT", "NO ACTION"] | None = None

646 """Action to take when the referenced row is deleted."""

647

648 on_update: Literal["CASCADE", "SET NULL", "SET DEFAULT", "RESTRICT", "NO ACTION"] | None = None

649 """Action to take when the referenced row is updated."""

650

651 @field_serializer("type")

652 def serialize_type(self, value: str) -> str:

653 """Ensure '@type' is included in serialized output.

654

655 Parameters

656 ----------

657 value

658 The value to serialize.

659

660 Returns

661 -------

662 `str`

663 The serialized value.

664 """

665 return value

666

667 @model_validator(mode="after")

668 def check_column_lengths(self) -> ForeignKeyConstraint:

669 """Check that the `columns` and `referenced_columns` lists have the

670 same length.

671

672 Returns

673 -------

674 `ForeignKeyConstraint`

675 The foreign key constraint being validated.

676

677 Raises

678 ------

679 ValueError

680 Raised if the `columns` and `referenced_columns` lists do not have

681 the same length.

682 """

683 if len(self.columns) != len(self.referenced_columns):

684 raise ValueError(

685 "Columns and referencedColumns must have the same length for a ForeignKey constraint"

686 )

687 return self

688

689

690_ConstraintType = Annotated[

691 CheckConstraint | ForeignKeyConstraint | UniqueConstraint, Field(discriminator="type")

692]

693"""Type alias for a constraint type."""

694

695

696class Index(BaseObject):

697 """Table index model.

698

699 An index can be defined on either columns or expressions, but not both.

700 """

701

702 columns: list[str] | None = None

703 """Columns in the index."""

704

705 expressions: list[str] | None = None

706 """Expressions in the index."""

707

708 @model_validator(mode="before")

709 @classmethod

710 def check_columns_or_expressions(cls, values: dict[str, Any]) -> dict[str, Any]:

711 """Check that columns or expressions are specified, but not both.

712

713 Parameters

714 ----------

715 values

716 Values of the index.

717

718 Returns

719 -------

720 `dict` [ `str`, `Any` ]

721 The values of the index.

722

723 Raises

724 ------

725 ValueError

726 Raised if both columns and expressions are specified, or if neither

727 are specified.

728 """

729 if "columns" in values and "expressions" in values:

730 raise ValueError("Defining columns and expressions is not valid")

731 elif "columns" not in values and "expressions" not in values:

732 raise ValueError("Must define columns or expressions")

733 return values

734

735

736ColumnRef: TypeAlias = str

737"""Type alias for a column reference."""

738

739

740class ColumnGroup(BaseObject):

741 """Column group model."""

742

743 columns: list[ColumnRef | Column] = Field(..., min_length=1)

744 """Columns in the group."""

745

746 ivoa_ucd: str | None = Field(None, alias="ivoa:ucd")

747 """IVOA UCD of the column."""

748

749 table: Table | None = Field(None, exclude=True)

750 """Reference to the parent table."""

751

752 @field_validator("ivoa_ucd")

753 @classmethod

754 def check_ivoa_ucd(cls, ivoa_ucd: str) -> str:

755 """Check that IVOA UCD values are valid.

756

757 Parameters

758 ----------

759 ivoa_ucd

760 IVOA UCD value to check.

761

762 Returns

763 -------

764 `str`

765 The IVOA UCD value if it is valid.

766 """

767 return validate_ivoa_ucd(ivoa_ucd)

768

769 @model_validator(mode="after")

770 def check_unique_columns(self) -> ColumnGroup:

771 """Check that the columns list contains unique items.

772

773 Returns

774 -------

775 `ColumnGroup`

776 The column group being validated.

777 """

778 column_ids = [col if isinstance(col, str) else col.id for col in self.columns]

779 if len(column_ids) != len(set(column_ids)):

780 raise ValueError("Columns in the group must be unique")

781 return self

782

783 def _dereference_columns(self) -> None:

784 """Dereference ColumnRef to Column objects."""

785 if self.table is None:

786 raise ValueError("ColumnGroup must have a reference to its parent table")

787

788 dereferenced_columns: list[ColumnRef | Column] = []

789 for col in self.columns:

790 if isinstance(col, str):

791 # Dereference ColumnRef to Column object

792 try:

793 col_obj = self.table._find_column_by_id(col)

794 except KeyError as e:

795 raise ValueError(f"Column '{col}' not found in table '{self.table.name}'") from e

796 dereferenced_columns.append(col_obj)

797 else:

798 dereferenced_columns.append(col)

799

800 self.columns = dereferenced_columns

801

802 @field_serializer("columns")

803 def serialize_columns(self, columns: list[ColumnRef | Column]) -> list[str]:

804 """Serialize columns as their IDs.

805

806 Parameters

807 ----------

808 columns

809 The columns to serialize.

810

811 Returns

812 -------

813 `list` [ `str` ]

814 The serialized column IDs.

815 """

816 return [col if isinstance(col, str) else col.id for col in columns]

817

818

819class ColumnOverrides(BaseModel):

820 """Allowed overrides for a referenced column.

821

822 All fields are optional; missing values mean "inherit from the base

823 column".

824 """

825

826 datatype: DataType | None = None

827 """New datatype for the column."""

828

829 length: int | None = None

830 """New length for the column."""

831

832 description: str | None = None

833 """New description for the column."""

834

835 nullable: bool = True

836 """New nullable flag for the column."""

837

838 tap_principal: int | None = Field(default=None, alias="tap:principal")

839 """Override for the TAP_SCHEMA 'principal' flag."""

840

841 tap_column_index: int | None = Field(default=None, alias="tap:column_index")

842 """Override for the TAP_SCHEMA column index."""

843

844 @field_serializer("datatype")

845 def serialize_datatype(self, value: DataType | None) -> str | None:

846 """Convert `DataType` to string when serializing to JSON/YAML.

847

848 Parameters

849 ----------

850 value

851 The `DataType` value to serialize, or None.

852

853 Returns

854 -------

855 `str` | None

856 The serialized `DataType` value, or None if the input was None.

857 """

858 if value is None:

859 return None

860 return str(value)

861

862 @field_validator("datatype", mode="before")

863 @classmethod

864 def deserialize_datatype(cls, value: str | None) -> DataType | None:

865 """Convert string back into `DataType` when loading from JSON/YAML.

866

867 Parameters

868 ----------

869 value

870 The string value to deserialize, or None.

871

872 Returns

873 -------

874 `DataType` | None

875 The deserialized `DataType` value, or None if the input was None.

876 """

877 if value is None:

878 return None

879 return DataType(value)

880

881

882class ColumnResourceRef(BaseModel):

883 """A column which is dervived from an external resource."""

884

885 ref_name: str | None = None

886 """Name of the referenced column in the resource

887 (if different from the key)."""

888

889 overrides: ColumnOverrides | None = None

890 """Optional overrides of the referenced column's attributes."""

891

892

893# Type aliases for the nested mapping structure of referenced columns

894ResourceColumnMap: TypeAlias = dict[str, ColumnResourceRef | None]

895ResourceTableMap: TypeAlias = dict[str, ResourceColumnMap]

896ResourceMap: TypeAlias = dict[str, ResourceTableMap]

897

898

899class Table(BaseObject):

900 """Table model."""

901

902 primary_key: str | list[str] | None = Field(None, alias="primaryKey")

903 """Primary key of the table."""

904

905 tap_table_index: int | None = Field(None, alias="tap:table_index")

906 """IVOA TAP_SCHEMA table index of the table."""

907

908 mysql_engine: str | None = Field("MyISAM", alias="mysql:engine")

909 """MySQL engine to use for the table."""

910

911 mysql_charset: str | None = Field(None, alias="mysql:charset")

912 """MySQL charset to use for the table."""

913

914 columns: list[Column] = Field(default_factory=list)

915 """Columns in the table."""

916

917 column_refs: ResourceMap = Field(default_factory=dict, alias="columnRefs")

918 """Referenced columns from external resources."""

919

920 column_groups: list[ColumnGroup] = Field(default_factory=list, alias="columnGroups")

921 """Column groups in the table."""

922

923 constraints: list[_ConstraintType] = Field(default_factory=list)

924 """Constraints on the table."""

925

926 indexes: list[Index] = Field(default_factory=list)

927 """Indexes on the table."""

928

929 @field_validator("columns", mode="after")

930 @classmethod

931 def check_unique_column_names(cls, columns: list[Column]) -> list[Column]:

932 """Check that column names are unique.

933

934 Parameters

935 ----------

936 columns

937 The columns to check.

938

939 Returns

940 -------

941 `list` [ `Column` ]

942 The columns if they are unique.

943

944 Raises

945 ------

946 ValueError

947 Raised if column names are not unique.

948 """

949 if len(columns) != len(set(column.name for column in columns)):

950 raise ValueError("Column names must be unique")

951 return columns

952

953 @model_validator(mode="after")

954 def check_tap_table_index(self, info: ValidationInfo) -> Table:

955 """Check that the table has a TAP table index.

956

957 Parameters

958 ----------

959 info

960 Validation context used to determine if the check is enabled.

961

962 Returns

963 -------

964 `Table`

965 The table being validated.

966

967 Raises

968 ------

969 ValueError

970 Raised If the table is missing a TAP table index.

971 """

972 context = info.context

973 if not context or not context.get("check_tap_table_indexes", False):

974 return self

975 if self.tap_table_index is None:

976 raise ValueError("Table is missing a TAP table index")

977 return self

978

979 @model_validator(mode="after")

980 def check_tap_principal(self, info: ValidationInfo) -> Table:

981 """Check that at least one column is flagged as 'principal' for TAP

982 purposes.

983

984 Parameters

985 ----------

986 info

987 Validation context used to determine if the check is enabled.

988

989 Returns

990 -------

991 `Table`

992 The table being validated.

993

994 Raises

995 ------

996 ValueError

997 Raised if the table is missing a column flagged as 'principal'.

998 """

999 context = info.context

1000 if not context or not context.get("check_tap_principal", False):

1001 return self

1002 for col in self.columns:

1003 if col.tap_principal == 1:

1004 return self

1005 raise ValueError(f"Table '{self.name}' is missing at least one column designated as 'tap:principal'")

1006

1007 def _find_column_by_id(self, id: str) -> Column:

1008 """Find a column by ID.

1009

1010 Parameters

1011 ----------

1012 id

1013 The ID of the column to find.

1014

1015 Returns

1016 -------

1017 `Column`

1018 The column with the given ID.

1019

1020 Raises

1021 ------

1022 ValueError

1023 Raised if the column is not found.

1024 """

1025 for column in self.columns:

1026 if column.id == id:

1027 return column

1028 raise KeyError(f"Column '{id}' not found in table '{self.name}'")

1029

1030 def _find_column_by_name(self, name: str) -> Column:

1031 for column in self.columns:

1032 if column.name == name:

1033 return column

1034 raise KeyError(f"Column '{name}' not found in table '{self.name}'")

1035

1036 @model_validator(mode="after")

1037 def dereference_column_groups(self: Table) -> Table:

1038 """Dereference columns in column groups.

1039

1040 Returns

1041 -------

1042 `Table`

1043 The table with dereferenced column groups.

1044 """

1045 for group in self.column_groups:

1046 group.table = self

1047 group._dereference_columns()

1048 return self

1049

1050 @field_serializer("columns")

1051 def _serialize_columns(self, columns: list[Column]) -> list[dict[str, Any]]:

1052 """Serialize only non-resource columns."""

1053 return [

1054 col.model_dump(

1055 by_alias=True,

1056 exclude_none=True,

1057 exclude_defaults=True,

1058 )

1059 for col in columns

1060 if not col._is_resource_ref

1061 ]

1062

1063

1064class SchemaVersion(BaseModel):

1065 """Schema version model."""

1066

1067 current: str

1068 """The current version of the schema."""

1069

1070 compatible: list[str] = Field(default_factory=list)

1071 """The compatible versions of the schema."""

1072

1073 read_compatible: list[str] = Field(default_factory=list)

1074 """The read compatible versions of the schema."""

1075

1076

1077class SchemaIdVisitor:

1078 """Visit a schema and build the map of IDs to objects.

1079

1080 Notes

1081 -----

1082 Duplicates are added to a set when they are encountered, which can be

1083 accessed via the ``duplicates`` attribute. The presence of duplicates will

1084 not throw an error. Only the first object with a given ID will be added to

1085 the map, but this should not matter, since a ``ValidationError`` will be

1086 thrown by the ``model_validator`` method if any duplicates are found in the

1087 schema.

1088 """

1089

1090 def __init__(self) -> None:

1091 """Create a new SchemaVisitor."""

1092 self.schema: Schema | None = None

1093 self.duplicates: set[str] = set()

1094

1095 def add(self, obj: BaseObject) -> None:

1096 """Add an object to the ID map.

1097

1098 Parameters

1099 ----------

1100 obj

1101 The object to add to the ID map.

1102 """

1103 if hasattr(obj, "id"):

1104 obj_id = getattr(obj, "id")

1105 if self.schema is not None:

1106 if obj_id in self.schema._id_map:

1107 self.duplicates.add(obj_id)

1108 else:

1109 self.schema._id_map[obj_id] = obj

1110

1111 def visit_schema(self, schema: Schema) -> None:

1112 """Visit the objects in a schema and build the ID map.

1113

1114 Parameters

1115 ----------

1116 schema

1117 The schema object to visit.

1118

1119 Notes

1120 -----

1121 This will set an internal variable pointing to the schema object.

1122 """

1123 self.schema = schema

1124 self.duplicates.clear()

1125 self.add(self.schema)

1126 for table in self.schema.tables:

1127 self.visit_table(table)

1128

1129 def visit_table(self, table: Table) -> None:

1130 """Visit a table object.

1131

1132 Parameters

1133 ----------

1134 table

1135 The table object to visit.

1136 """

1137 self.add(table)

1138 for column in table.columns:

1139 self.visit_column(column)

1140 for constraint in table.constraints:

1141 self.visit_constraint(constraint)

1142

1143 def visit_column(self, column: Column) -> None:

1144 """Visit a column object.

1145

1146 Parameters

1147 ----------

1148 column

1149 The column object to visit.

1150 """

1151 self.add(column)

1152

1153 def visit_constraint(self, constraint: Constraint) -> None:

1154 """Visit a constraint object.

1155

1156 Parameters

1157 ----------

1158 constraint

1159 The constraint object to visit.

1160 """

1161 self.add(constraint)

1162

1163

1164T = TypeVar("T", bound=BaseObject)

1165

1166

1167def _strip_ids(data: Any) -> Any:

1168 """Recursively strip '@id' fields from a dictionary or list.

1169

1170 Parameters

1171 ----------

1172 data

1173 The data to strip IDs from, which can be a dictionary, list, or any

1174 other type. Other types will be returned unchanged.

1175 """

1176 if isinstance(data, dict):

1177 data.pop("@id", None)

1178 for k, v in data.items():

1179 data[k] = _strip_ids(v)

1180 return data

1181 elif isinstance(data, list):

1182 return [_strip_ids(item) for item in data]

1183 else:

1184 return data

1185

1186

1187def _append_error(

1188 errors: list[InitErrorDetails],

1189 loc: tuple,

1190 input_value: Any,

1191 error_message: str,

1192 error_type: str = "value_error",

1193) -> None:

1194 """Append an error to the errors list.

1195

1196 Parameters

1197 ----------

1198 errors : list[InitErrorDetails]

1199 The list of errors to append to.

1200 loc : tuple

1201 The location of the error in the schema.

1202 input_value : Any

1203 The input value that caused the error.

1204 error_message : str

1205 The error message to include in the context.

1206 """

1207 errors.append(

1208 {

1209 "type": error_type,

1210 "loc": loc,

1211 "input": input_value,

1212 "ctx": {"error": error_message},

1213 }

1214 )

1215

1216

1217class Resource(BaseModel):

1218 """A resource definition referencing an external schema."""

1219

1220 uri: str = Field(..., description="Resource URI or path")

1221 """URI of the schema resource which may be a local path, ``resource://``,

1222 or remote URL."""

1223

1224

1225class Schema(BaseObject, Generic[T]):

1226 """Database schema model.

1227

1228 This represents a database schema, which contains one or more tables.

1229 """

1230

1231 version: SchemaVersion | str | None = None

1232 """The version of the schema."""

1233

1234 tables: Sequence[Table]

1235 """The tables in the schema."""

1236

1237 resources: dict[str, Resource] = Field(default_factory=dict)

1238 """External resources referenced by this schema."""

1239

1240 _id_map: dict[str, Any] = PrivateAttr(default_factory=dict)

1241 """Map of IDs to objects."""

1242

1243 _resource_map: dict[str, Schema] = PrivateAttr(default_factory=dict)

1244 """Map of resource names to loaded schemas."""

1245

1246 @model_validator(mode="after")

1247 def _load_resources(self: Schema, info: ValidationInfo) -> Schema:

1248 """Load external resources referenced by this schema into an internal

1249 mapping of resource names to their `Schema` objects.

1250

1251 Returns

1252 -------

1253 `Schema`

1254 The schema being validated.

1255

1256 Raises

1257 ------

1258 ValueError

1259 Raised if a resource cannot be loaded.

1260 """

1261 if info.context:

1262 context = info.context.copy()

1263 # Ignore this flag for loading the resources themselves

1264 context.pop("dereference_resources", None)

1265 else:

1266 context = {}

1267

1268 for resource_name, resource in self.resources.items():

1269 uri = resource.uri

1270 try:

1271 loaded_schema = Schema.from_uri(uri, context=context)

1272 self._resource_map[resource_name] = loaded_schema

1273 logger.debug(f"Loaded resource '{resource_name}' from URI '{uri}'")

1274 except Exception as e:

1275 raise ValueError(f"Failed to load resource '{resource_name}' from URI '{uri}': {e}") from e

1276 return self

1277

1278 def _find_table_by_name(self, name: str) -> Table:

1279 """Find a table by name.

1280

1281 Parameters

1282 ----------

1283 name

1284 The name of the table to find.

1285

1286 Returns

1287 -------

1288 `Table`

1289 The table with the given name.

1290

1291 Raises

1292 ------

1293 KeyError

1294 Raised if the table is not found.

1295 """

1296 for table in self.tables:

1297 if table.name == name:

1298 return table

1299 raise KeyError(f"Table '{name}' not found in schema '{self.name}'")

1300

1301 @model_validator(mode="after")

1302 def _dereference_resource_columns(self: Schema, info: ValidationInfo) -> Schema:

1303 """Dereference columns from external resources and add them to the

1304 tables in this schema.

1305 """

1306 context = info.context

1307 column_ref_index_increment: int | None = None

1308 dereference_resources = False

1309 if context is not None:

1310 dereference_resources = context.get("dereference_resources", False)

1311 column_ref_index_increment = context.get("column_ref_index_increment", None)

1312

1313 for table in self.tables:

1314 if column_refs := table.column_refs:

1315 for resource_name, tables in column_refs.items():

1316 resource_schema = self._resource_map.get(resource_name)

1317 if resource_schema is None:

1318 raise ValueError(f"Schema resource '{resource_name}' was not found in resources")

1319 self._process_column_refs(

1320 table,

1321 tables,

1322 resource_schema,

1323 dereference_resources,

1324 column_ref_index_increment,

1325 )

1326 if dereference_resources and len(table.column_refs) > 0:

1327 # Clear column refs in table if fully dereferencing

1328 logger.debug(

1329 f"Clearing columnRefs in table '{table.name}' after dereferencing resource columns"

1330 )

1331 table.column_refs = {}

1332 return self

1333

1334 def _process_column_refs(

1335 self,

1336 table: Table,

1337 ref_tables: ResourceTableMap,

1338 resource_schema: Schema,

1339 dereference_resources: bool = False,

1340 column_ref_index_increment: int | None = None,

1341 ) -> None:

1342 """Process column references from an external resource and add them

1343 to the given table.

1344 """

1345 current_column_index = column_ref_index_increment if column_ref_index_increment is not None else -1

1346

1347 for table_name, columns in ref_tables.items():

1348 try:

1349 resource_table = resource_schema._find_table_by_name(table_name)

1350 except KeyError as e:

1351 raise ValueError(

1352 f"Table '{table_name}' not found in resource '{resource_schema.name}'"

1353 ) from e

1354 for local_column_name, column_ref in columns.items():

1355 if column_ref is not None and column_ref.ref_name is not None:

1356 # Use specified ref_name

1357 ref_column_name = column_ref.ref_name

1358 else:

1359 # Use the local column name if no ref_name

1360 # specified

1361 ref_column_name = local_column_name

1362

1363 # Check if referenced column exists in resource

1364 try:

1365 base_column = resource_table._find_column_by_name(ref_column_name)

1366 except KeyError:

1367 # The ref_name is specified but column is not

1368 # found

1369 if column_ref is not None and column_ref.ref_name is not None:

1370 raise ValueError(

1371 f"Column '{ref_column_name}' not found in table '{table_name}' "

1372 f"from resource '{resource_schema.name}'"

1373 )

1374 # The ref_name is not specified and the local

1375 # column name is not found

1376 raise ValueError(

1377 f"Column '{local_column_name}' not found in table '{table_name}' "

1378 f"from resource '{resource_schema.name}' and no ref_name provided"

1379 )

1380

1381 # Create a copy of the base column and apply

1382 # overrides

1383 column_copy = base_column.model_copy()

1384

1385 # Set the local name (key from the mapping)

1386 column_copy.name = local_column_name

1387

1388 if not dereference_resources:

1389 # Flag the column as a resource reference so it will not be

1390 # written out during serialization

1391 column_copy._is_resource_ref = True

1392

1393 # Apply overrides to the original column definition

1394 overrides: ColumnOverrides | None = None

1395 if column_ref is not None and column_ref.overrides is not None:

1396 overrides = column_ref.overrides

1397 for field_name, override_value in overrides.model_dump().items():

1398 if override_value is not None:

1399 setattr(column_copy, field_name, override_value)

1400

1401 # Manually set the ID of the copied column as ID generation has

1402 # already occurred by now

1403 column_copy.id = f"{table.id}.{local_column_name}"

1404

1405 # Apply automatic assignment of 'tap:column_index', if enabled

1406 if column_ref_index_increment is not None:

1407 if (not overrides) or (overrides.tap_column_index is None):

1408 column_copy.tap_column_index = current_column_index

1409 current_column_index += column_ref_index_increment

1410 logger.debug(

1411 f"Automatically assigned 'tap:column_index' {column_copy.tap_column_index} to "

1412 f"column '{local_column_name}' in table '{table_name}' from resource "

1413 f"'{resource_schema.name}'"

1414 )

1415 else:

1416 # Skip automatic assignment of 'tap:column_index' if it

1417 # is already overridden

1418 logger.debug(

1419 f"Skipping automatic assignment of 'tap:column_index' for column "

1420 f"'{local_column_name}' in table '{table_name}' from resource "

1421 f"'{resource_schema.name}' as it is already overridden to "

1422 f"{column_copy.tap_column_index}"

1423 )

1424 table.columns.append(column_copy)

1425 logger.debug(

1426 f"Dereferenced column '{local_column_name}' from table '{table_name}' "

1427 f"in resource '{resource_schema.name}' into table '{table.name}'"

1428 )

1429

1430 @model_validator(mode="before")

1431 @classmethod

1432 def generate_ids(cls, values: dict[str, Any], info: ValidationInfo) -> dict[str, Any]:

1433 """Generate IDs for objects that do not have them.

1434

1435 Parameters

1436 ----------

1437 values

1438 The values of the schema.

1439 info

1440 Validation context used to determine if ID generation is enabled.

1441

1442 Returns

1443 -------

1444 `dict` [ `str`, `Any` ]

1445 The values of the schema with generated IDs.

1446 """

1447 context = info.context

1448 if not context or not context.get("id_generation", False):

1449 logger.debug("Skipping ID generation")

1450 return values

1451 schema_name = values["name"]

1452 if "@id" not in values:

1453 values["@id"] = f"#{schema_name}"

1454 logger.debug(f"Generated ID '{values['@id']}' for schema '{schema_name}'")

1455 if "tables" in values:

1456 for table in values["tables"]:

1457 if "@id" not in table:

1458 table["@id"] = f"#{table['name']}"

1459 logger.debug(f"Generated ID '{table['@id']}' for table '{table['name']}'")

1460 if "columns" in table:

1461 for column in table["columns"]:

1462 if "@id" not in column:

1463 column["@id"] = f"#{table['name']}.{column['name']}"

1464 logger.debug(f"Generated ID '{column['@id']}' for column '{column['name']}'")

1465 if "columnGroups" in table:

1466 for column_group in table["columnGroups"]:

1467 if "@id" not in column_group:

1468 column_group["@id"] = f"#{table['name']}.{column_group['name']}"

1469 logger.debug(

1470 f"Generated ID '{column_group['@id']}' for column group "

1471 f"'{column_group['name']}'"

1472 )

1473 if "constraints" in table:

1474 for constraint in table["constraints"]:

1475 if "@id" not in constraint:

1476 constraint["@id"] = f"#{constraint['name']}"

1477 logger.debug(

1478 f"Generated ID '{constraint['@id']}' for constraint '{constraint['name']}'"

1479 )

1480 if "indexes" in table:

1481 for index in table["indexes"]:

1482 if "@id" not in index:

1483 index["@id"] = f"#{index['name']}"

1484 logger.debug(f"Generated ID '{index['@id']}' for index '{index['name']}'")

1485 return values

1486

1487 @field_validator("tables", mode="after")

1488 @classmethod

1489 def check_unique_table_names(cls, tables: list[Table]) -> list[Table]:

1490 """Check that table names are unique.

1491

1492 Parameters

1493 ----------

1494 tables

1495 The tables to check.

1496

1497 Returns

1498 -------

1499 `list` [ `Table` ]

1500 The tables if they are unique.

1501

1502 Raises

1503 ------

1504 ValueError

1505 Raised if table names are not unique.

1506 """

1507 if len(tables) != len(set(table.name for table in tables)):

1508 raise ValueError("Table names must be unique")

1509 return tables

1510

1511 @model_validator(mode="after")

1512 def check_tap_table_indexes(self, info: ValidationInfo) -> Schema:

1513 """Check that the TAP table indexes are unique.

1514

1515 Parameters

1516 ----------

1517 info

1518 The validation context used to determine if the check is enabled.

1519

1520 Returns

1521 -------

1522 `Schema`

1523 The schema being validated.

1524 """

1525 context = info.context

1526 if not context or not context.get("check_tap_table_indexes", False):

1527 return self

1528 table_indicies = set()

1529 for table in self.tables:

1530 table_index = table.tap_table_index

1531 if table_index is not None:

1532 if table_index in table_indicies:

1533 raise ValueError(f"Duplicate 'tap:table_index' value {table_index} found in schema")

1534 table_indicies.add(table_index)

1535 return self

1536

1537 @model_validator(mode="after")

1538 def check_unique_constraint_names(self: Schema) -> Schema:

1539 """Check for duplicate constraint names in the schema.

1540

1541 Returns

1542 -------

1543 `Schema`

1544 The schema being validated.

1545

1546 Raises

1547 ------

1548 ValueError

1549 Raised if duplicate constraint names are found in the schema.

1550 """

1551 constraint_names = set()

1552 duplicate_names = []

1553

1554 for table in self.tables:

1555 for constraint in table.constraints:

1556 constraint_name = constraint.name

1557 if constraint_name in constraint_names:

1558 duplicate_names.append(constraint_name)

1559 else:

1560 constraint_names.add(constraint_name)

1561

1562 if duplicate_names:

1563 raise ValueError(f"Duplicate constraint names found in schema: {duplicate_names}")

1564

1565 return self

1566

1567 @model_validator(mode="after")

1568 def check_unique_index_names(self: Schema) -> Schema:

1569 """Check for duplicate index names in the schema.

1570

1571 Returns

1572 -------

1573 `Schema`

1574 The schema being validated.

1575

1576 Raises

1577 ------

1578 ValueError

1579 Raised if duplicate index names are found in the schema.

1580 """

1581 index_names = set()

1582 duplicate_names = []

1583

1584 for table in self.tables:

1585 for index in table.indexes:

1586 index_name = index.name

1587 if index_name in index_names:

1588 duplicate_names.append(index_name)

1589 else:

1590 index_names.add(index_name)

1591

1592 if duplicate_names:

1593 raise ValueError(f"Duplicate index names found in schema: {duplicate_names}")

1594

1595 return self

1596

1597 @model_validator(mode="after")

1598 def create_id_map(self: Schema) -> Schema:

1599 """Create a map of IDs to objects.

1600

1601 Returns

1602 -------

1603 `Schema`

1604 The schema with the ID map created.

1605

1606 Raises

1607 ------

1608 ValueError

1609 Raised if duplicate identifiers are found in the schema.

1610 """

1611 if self._id_map:

1612 logger.debug("Ignoring call to create_id_map() - ID map was already populated")

1613 return self

1614 visitor: SchemaIdVisitor = SchemaIdVisitor()

1615 visitor.visit_schema(self)

1616 if len(visitor.duplicates):

1617 raise ValueError(

1618 "Duplicate IDs found in schema:\n " + "\n ".join(visitor.duplicates) + "\n"

1619 )

1620 logger.debug("Created ID map with %d entries", len(self._id_map))

1621 return self

1622

1623 def _validate_column_id(

1624 self: Schema,

1625 column_id: str,

1626 loc: tuple,

1627 errors: list[InitErrorDetails],

1628 ) -> None:

1629 """Validate a column ID from a constraint and append errors if invalid.

1630

1631 Parameters

1632 ----------

1633 schema : Schema

1634 The schema being validated.

1635 column_id : str

1636 The column ID to validate.

1637 loc : tuple

1638 The location of the error in the schema.

1639 errors : list[InitErrorDetails]

1640 The list of errors to append to.

1641 """

1642 if column_id not in self:

1643 _append_error(

1644 errors,

1645 loc,

1646 column_id,

1647 f"Column ID '{column_id}' not found in schema",

1648 )

1649 elif not isinstance(self[column_id], Column):

1650 _append_error(

1651 errors,

1652 loc,

1653 column_id,

1654 f"ID '{column_id}' does not refer to a Column object",

1655 )

1656

1657 def _validate_foreign_key_column(

1658 self: Schema,

1659 column_id: str,

1660 table: Table,

1661 loc: tuple,

1662 errors: list[InitErrorDetails],

1663 ) -> None:

1664 """Validate a foreign key column ID from a constraint and append errors

1665 if invalid.

1666

1667 Parameters

1668 ----------

1669 schema : Schema

1670 The schema being validated.

1671 column_id : str

1672 The foreign key column ID to validate.

1673 loc : tuple

1674 The location of the error in the schema.

1675 errors : list[InitErrorDetails]

1676 The list of errors to append to.

1677 """

1678 try:

1679 table._find_column_by_id(column_id)

1680 except KeyError:

1681 _append_error(

1682 errors,

1683 loc,

1684 column_id,

1685 f"Column '{column_id}' not found in table '{table.name}'",

1686 )

1687

1688 @model_validator(mode="after")

1689 def check_constraints(self: Schema) -> Schema:

1690 """Check constraint objects for validity. This needs to be deferred

1691 until after the schema is fully loaded and the ID map is created.

1692

1693 Raises

1694 ------

1695 pydantic.ValidationError

1696 Raised if any constraints are invalid.

1697

1698 Returns

1699 -------

1700 `Schema`

1701 The schema being validated.

1702 """

1703 errors: list[InitErrorDetails] = []

1704

1705 for table_index, table in enumerate(self.tables):

1706 for constraint_index, constraint in enumerate(table.constraints):

1707 column_ids: list[str] = []

1708 referenced_column_ids: list[str] = []

1709

1710 if isinstance(constraint, ForeignKeyConstraint):

1711 column_ids += constraint.columns

1712 referenced_column_ids += constraint.referenced_columns

1713 elif isinstance(constraint, UniqueConstraint):

1714 column_ids += constraint.columns

1715 # No extra checks are required on CheckConstraint objects.

1716

1717 # Validate the foreign key columns

1718 for column_id in column_ids:

1719 self._validate_column_id(

1720 column_id,

1721 (

1722 "tables",

1723 table_index,

1724 "constraints",

1725 constraint_index,

1726 "columns",

1727 column_id,

1728 ),

1729 errors,

1730 )

1731 # Check that the foreign key column is within the source

1732 # table.

1733 self._validate_foreign_key_column(

1734 column_id,

1735 table,

1736 (

1737 "tables",

1738 table_index,

1739 "constraints",

1740 constraint_index,

1741 "columns",

1742 column_id,

1743 ),

1744 errors,

1745 )

1746

1747 # Validate the primary key (reference) columns

1748 for referenced_column_id in referenced_column_ids:

1749 self._validate_column_id(

1750 referenced_column_id,

1751 (

1752 "tables",

1753 table_index,

1754 "constraints",

1755 constraint_index,

1756 "referenced_columns",

1757 referenced_column_id,

1758 ),

1759 errors,

1760 )

1761

1762 if errors:

1763 raise ValidationError.from_exception_data("Schema validation failed", errors)

1764

1765 return self

1766

1767 def __getitem__(self, id: str) -> BaseObject:

1768 """Get an object by its ID.

1769

1770 Parameters

1771 ----------

1772 id

1773 The ID of the object to get.

1774

1775 Raises

1776 ------

1777 KeyError

1778 Raised if the object with the given ID is not found in the schema.

1779 """

1780 if id not in self:

1781 raise KeyError(f"Object with ID '{id}' not found in schema")

1782 return self._id_map[id]

1783

1784 def __contains__(self, id: str) -> bool:

1785 """Check if an object with the given ID is in the schema.

1786

1787 Parameters

1788 ----------

1789 id

1790 The ID of the object to check.

1791 """

1792 return id in self._id_map

1793

1794 def find_object_by_id(self, id: str, obj_type: type[T]) -> T:

1795 """Find an object with the given type by its ID.

1796

1797 Parameters

1798 ----------

1799 id

1800 The ID of the object to find.

1801 obj_type

1802 The type of the object to find.

1803

1804 Returns

1805 -------

1806 BaseObject

1807 The object with the given ID and type.

1808

1809 Raises

1810 ------

1811 KeyError

1812 If the object with the given ID is not found in the schema.

1813 TypeError

1814 If the object that is found does not have the right type.

1815

1816 Notes

1817 -----

1818 The actual return type is the user-specified argument ``T``, which is

1819 expected to be a subclass of `BaseObject`.

1820 """

1821 obj = self[id]

1822 if not isinstance(obj, obj_type):

1823 raise TypeError(f"Object with ID '{id}' is not of type '{obj_type.__name__}'")

1824 return obj

1825

1826 def get_table_by_column(self, column: Column) -> Table:

1827 """Find the table that contains a column.

1828

1829 Parameters

1830 ----------

1831 column

1832 The column to find.

1833

1834 Returns

1835 -------

1836 `Table`

1837 The table that contains the column.

1838

1839 Raises

1840 ------

1841 ValueError

1842 If the column is not found in any table.

1843 """

1844 for table in self.tables:

1845 if column in table.columns:

1846 return table

1847 raise ValueError(f"Column '{column.name}' not found in any table")

1848

1849 @classmethod

1850 def from_uri(cls, resource_path: ResourcePathExpression, context: dict[str, Any] = {}) -> Schema:

1851 """Load a `Schema` from a string representing a ``ResourcePath``.

1852

1853 Parameters

1854 ----------

1855 resource_path

1856 The ``ResourcePath`` pointing to a YAML file.

1857 context

1858 Pydantic context to be used in validation.

1859

1860 Returns

1861 -------

1862 `str`

1863 The ID of the object.

1864

1865 Raises

1866 ------

1867 yaml.YAMLError

1868 Raised if there is an error loading the YAML data.

1869 ValueError

1870 Raised if there is an error reading the resource.

1871 pydantic.ValidationError

1872 Raised if the schema fails validation.

1873 """

1874 try:

1875 rp_stream = ResourcePath(resource_path).read()

1876 except Exception as e:

1877 raise ValueError(f"Error reading resource from '{resource_path}' : {e}") from e

1878 yaml_data = yaml.safe_load(rp_stream)

1879 return Schema.model_validate(yaml_data, context=context)

1880

1881 @classmethod

1882 def from_stream(cls, source: IO[str], context: dict[str, Any] = {}) -> Schema:

1883 """Load a `Schema` from a file stream which should contain YAML data.

1884

1885 Parameters

1886 ----------

1887 source

1888 The file stream to read from.

1889 context

1890 Pydantic context to be used in validation.

1891

1892 Returns

1893 -------

1894 `Schema`

1895 The Felis schema loaded from the stream.

1896

1897 Raises

1898 ------

1899 yaml.YAMLError

1900 Raised if there is an error loading the YAML file.

1901 pydantic.ValidationError

1902 Raised if the schema fails validation.

1903 """

1904 logger.debug("Loading schema from: '%s'", source)

1905 yaml_data = yaml.safe_load(source)

1906 return Schema.model_validate(yaml_data, context=context)

1907

1908 def _model_dump(self, strip_ids: bool = False) -> dict[str, Any]:

1909 """Dump the schema as a dictionary with some default arguments

1910 applied.

1911

1912 Parameters

1913 ----------

1914 strip_ids

1915 Whether to strip the IDs from the dumped data. Defaults to `False`.

1916

1917 Returns

1918 -------

1919 `dict` [ `str`, `Any` ]

1920 The dumped schema data as a dictionary.

1921 """

1922 data = self.model_dump(by_alias=True, exclude_none=True, exclude_defaults=True)

1923 if strip_ids:

1924 data = _strip_ids(data)

1925 return data

1926

1927 def dump_yaml(self, stream: IO[str] = sys.stdout, strip_ids: bool = False) -> None:

1928 """Pretty print the schema as YAML.

1929

1930 Parameters

1931 ----------

1932 stream

1933 The stream to write the YAML data to.

1934 strip_ids

1935 Whether to strip the IDs from the dumped data. Defaults to `False`.

1936 """

1937 data = self._model_dump(strip_ids=strip_ids)

1938 yaml.safe_dump(

1939 data,

1940 stream,

1941 default_flow_style=False,

1942 sort_keys=False,

1943 )

1944

1945 def dump_json(self, stream: IO[str] = sys.stdout, strip_ids: bool = False) -> None:

1946 """Pretty print the schema as JSON.

1947

1948 Parameters

1949 ----------

1950 stream

1951 The stream to write the JSON data to.

1952 strip_ids

1953 Whether to strip the IDs from the dumped data. Defaults to `False`.

1954 """

1955 data = self._model_dump(strip_ids=strip_ids)

1956 json.dump(

1957 data,

1958 stream,

1959 indent=4,

1960 sort_keys=False,

1961 )

Coverage for python / felis / datamodel.py: 31%

705 statements