Coverage for python / lsst / dax / apdb / apdbConfigFreezer.py: 23%

1# This file is part of dax_apdb. 

2# 

3# Developed for the LSST Data Management System. 

4# This product includes software developed by the LSST Project 

5# (http://www.lsst.org). 

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

7# for details of code ownership. 

8# 

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

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

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

12# (at your option) any later version. 

13# 

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

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

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

17# GNU General Public License for more details. 

18# 

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

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

21 

22from __future__ import annotations 

23 

24__all__ = ["ApdbConfigFreezer"] 

25 

26import json 

27from collections.abc import Iterable 

28from operator import attrgetter 

29from typing import Generic, TypeVar 

30 

31import pydantic 

32 

33from .apdb import ApdbConfig 

34 

35_Config = TypeVar("_Config", bound=ApdbConfig) 

36 

37 

38class ApdbConfigFreezer(Generic[_Config]): 

39 """Class that handles freezing of the configuration parameters, this is 

40 an implementation detail for use in Apdb subclasses. 

41 

42 Parameters 

43 ---------- 

44 field_names : `~collections.abc.Iterable` [`str`] 

45 Names of configuration fields to be frozen, they can be hierarchical 

46 (dot-separated). 

47 """ 

48 

49 def __init__(self, field_names: Iterable[str]): 

50 self._field_names = list(field_names) 

51 self._getters = {name.split(".")[-1]: attrgetter(name) for name in self._field_names} 

52 self._attr_parents = {} 

53 for name in self._field_names: 

54 path = name.split(".") 

55 self._attr_parents[path[-1]] = path[:-1] 

56 

57 def to_json(self, config: ApdbConfig) -> str: 

58 """Convert part of the configuration object to JSON string. 

59 

60 Parameters 

61 ---------- 

62 config : `ApdbConfig` 

63 Configuration object. 

64 

65 Returns 

66 ------- 

67 json_str : `str` 

68 JSON representation of the frozen part of the config. For 

69 hierarchical dot-separated names only that last part of the name is 

70 used in the returned JSON mapping. 

71 """ 

72 data = {name: getter(config) for name, getter in self._getters.items()} 

73 return json.dumps(data) 

74 

75 def update(self, config: _Config, json_str: str) -> _Config: 

76 """Update configuration field values from a JSON string. 

77 

78 Parameters 

79 ---------- 

80 config : `ApdbConfig` 

81 Configuration object. 

82 json_str : str 

83 String containing JSON representation of configuration. 

84 

85 Returns 

86 ------- 

87 updated : `ApdbConfig` 

88 Copy of the ``config`` with some fields updated from JSON object. 

89 

90 Raises 

91 ------ 

92 TypeError 

93 Raised if JSON string does not represent JSON object. 

94 ValueError 

95 Raised if JSON object contains key which is not present in 

96 ``field_names``. 

97 """ 

98 data = json.loads(json_str) 

99 if not isinstance(data, dict): 

100 raise TypeError(f"JSON string must be convertible to object: {json_str!r}") 

101 

102 # Older config used different parameter name for it. 

103 if "use_insert_id" in data: 

104 data["enable_replica"] = data.pop("use_insert_id") 

105 if "use_insert_id_skips_diaobjects" in data: 

106 data["replica_skips_diaobjects"] = data.pop("use_insert_id_skips_diaobjects") 

107 

108 # We want to update some fields and re-validate the model, the easiest 

109 # way to do it is to convert model to a dict first, update the dict and 

110 # convert back to model. 

111 model_data = config.model_dump() 

112 for attr, value in data.items(): 

113 parent_path = self._attr_parents.get(attr) 

114 if parent_path is None: 

115 raise ValueError(f"Frozen configuration contains unexpected attribute {attr}={value}") 

116 obj = model_data 

117 for parent_attr in parent_path: 

118 obj = obj[parent_attr] 

119 obj[attr] = value 

120 

121 try: 

122 new_config = type(config).model_validate(model_data) 

123 return new_config 

124 except pydantic.ValidationError as exc: 

125 raise ValueError("Validation error for frozen config") from exc