Coverage for python/felis/browser/schemas.py: 100%

1"""Lookups in schemas shared by the browser's pages.""" 

2 

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/>. 

23 

24from __future__ import annotations 

25 

26import re 

27from collections.abc import Iterable 

28from typing import TypeVar 

29 

30from lsst.utils.iteration import ensure_iterable 

31 

32from felis.datamodel import BaseObject, Column, ForeignKeyConstraint, Schema, SchemaVersion, Table 

33 

34_Named = TypeVar("_Named", bound=BaseObject) 

35"""A schema object of any kind, sorted by name in the sidebar data.""" 

36 

37 

38def _resolve_column(schema: Schema, ref: str) -> Column: 

39 """Return the column a primary key, index or constraint refers to. 

40 

41 Raises `ValueError` if ``ref`` isn't a column of the schema. This is the 

42 only place the browser looks up columns by ID; switch it to the name-based 

43 lookup from DM-51789 once that is merged, as ID lookups may be removed. 

44 """ 

45 try: 

46 return schema.find_object_by_id(ref, Column) 

47 except (KeyError, TypeError) as e: 

48 raise ValueError( 

49 f"Column reference '{ref}' in schema '{schema.name}' is not a column of the schema" 

50 ) from e 

51 

52 

53def _primary_key_columns(schema: Schema, table: Table) -> list[Column]: 

54 """Return the columns of a table's primary key, one reference or a list.""" 

55 return [_resolve_column(schema, ref) for ref in ensure_iterable(table.primary_key or [])] 

56 

57 

58def _foreign_keys(table: Table) -> list[ForeignKeyConstraint]: 

59 return [constraint for constraint in table.constraints if isinstance(constraint, ForeignKeyConstraint)] 

60 

61 

62def _schema_version(schema: Schema) -> str | None: 

63 if isinstance(schema.version, SchemaVersion): 

64 return schema.version.current 

65 return schema.version 

66 

67 

68def _version_order(version: str | None) -> str: 

69 """Return a sort key for a version: numbers zero-padded, so 3.8.1 sorts 

70 before 10.0.0. 

71 """ 

72 if version is None: 

73 return "" 

74 # Splitting on a captured group keeps the numbers as separate parts. 

75 parts = re.split(r"(\d+)", version) 

76 return "".join(part.zfill(10) if part.isdigit() else part for part in parts) 

77 

78 

79def _name_sort_key(obj: BaseObject) -> str: 

80 return obj.name.lower() 

81 

82 

83def _sorted_by_name(objects: Iterable[_Named]) -> list[_Named]: 

84 return sorted(objects, key=_name_sort_key)