Coverage for python / lsst / dax / apdb / apdbUpdateRecord.py: 62%

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__ = [ 

25 "ApdbCloseDiaObjectValidityRecord", 

26 "ApdbReassignDiaSourceToDiaObjectRecord", 

27 "ApdbReassignDiaSourceToSSObjectRecord", 

28 "ApdbUpdateNDiaSourcesRecord", 

29 "ApdbUpdateRecord", 

30 "ApdbWithdrawDiaForcedSourceRecord", 

31 "ApdbWithdrawDiaSourceRecord", 

32] 

33 

34import dataclasses 

35import json 

36from abc import ABC 

37from collections.abc import Mapping 

38from dataclasses import dataclass 

39from typing import Any, ClassVar 

40 

41from .apdb import ApdbTables 

42from .recordIds import DiaForcedSourceId, DiaObjectId, DiaSourceId 

43 

44 

45@dataclass(kw_only=True) 

46class ApdbUpdateRecord(ABC): 

47 """Abstract base class representing all types of update records saved to 

48 replica table. 

49 """ 

50 

51 update_time_ns: int 

52 """Time in nanoseconds since epoch when update happened.""" 

53 

54 update_order: int 

55 """Record order in the update.""" 

56 

57 update_type: ClassVar[str] 

58 """Class variable defining type of the update, must be defined in all 

59 concrete subclasses and be unique. 

60 """ 

61 

62 apdb_table: ClassVar[ApdbTables] 

63 """Class variable defining APDB table that this update applies to, must be 

64 defined in all concrete subclasses. 

65 """ 

66 

67 _update_types: ClassVar[dict[str, type[ApdbUpdateRecord]]] = {} 

68 """Class variable for mapping the type of the update to corresponding 

69 record class. 

70 """ 

71 

72 def __init_subclass__(cls, /, update_type: str, **kwargs: Any) -> None: 

73 super().__init_subclass__(**kwargs) 

74 cls.update_type = update_type 

75 cls._update_types[update_type] = cls 

76 

77 def __lt__(self, other: Any) -> bool: 

78 if isinstance(other, ApdbUpdateRecord): 

79 return (self.update_time_ns, self.update_order) < (other.update_time_ns, other.update_order) 

80 raise self._type_error(other) 

81 

82 def __le__(self, other: Any) -> bool: 

83 if isinstance(other, ApdbUpdateRecord): 

84 return (self.update_time_ns, self.update_order) <= (other.update_time_ns, other.update_order) 

85 raise self._type_error(other) 

86 

87 def __gt__(self, other: Any) -> bool: 

88 if isinstance(other, ApdbUpdateRecord): 

89 return (self.update_time_ns, self.update_order) > (other.update_time_ns, other.update_order) 

90 raise self._type_error(other) 

91 

92 def __ge__(self, other: Any) -> bool: 

93 if isinstance(other, ApdbUpdateRecord): 

94 return (self.update_time_ns, self.update_order) >= (other.update_time_ns, other.update_order) 

95 raise self._type_error(other) 

96 

97 def _type_error(self, other: Any) -> TypeError: 

98 return TypeError( 

99 "ordering is not supported between instances of " 

100 f"'{self.__class__.__name__}' and '{other.__class__.__name__}'" 

101 ) 

102 

103 @classmethod 

104 def from_json(cls, update_time_ns: int, update_order: int, json_str: str) -> ApdbUpdateRecord: 

105 json_obj = json.loads(json_str) 

106 if not isinstance(json_obj, Mapping): 

107 raise TypeError("String must contain JSON object.") 

108 kw = dict(json_obj) 

109 if (update_type := kw.pop("update_type", None)) is None: 

110 raise LookupError("`update_type` key is not in JSON object.") 

111 if (klass := cls._update_types.get(update_type)) is not None: 

112 return klass(update_time_ns=update_time_ns, update_order=update_order, **kw) 

113 else: 

114 raise ValueError(f"Unknown update type: {update_type}") 

115 

116 def to_json(self) -> str: 

117 data = dataclasses.asdict(self) 

118 # These fields are stored separately. 

119 data.pop("update_time_ns") 

120 data.pop("update_order") 

121 data["update_type"] = self.update_type 

122 return json.dumps(data) 

123 

124 

125@dataclass(kw_only=True) 

126class ApdbReassignDiaSourceToDiaObjectRecord( 

127 ApdbUpdateRecord, DiaSourceId, update_type="reassign_diasource_to_diaobject" 

128): 

129 """Update record representing re-assignment of DIASource to a different 

130 DIAObject. 

131 """ 

132 

133 diaObjectId: int 

134 """ID of a new associated DIAObject record.""" 

135 

136 apdb_table: ClassVar[ApdbTables] = ApdbTables.DiaSource 

137 

138 

139@dataclass(kw_only=True) 

140class ApdbReassignDiaSourceToSSObjectRecord( 

141 ApdbUpdateRecord, DiaSourceId, update_type="reassign_diasource_to_ssobject" 

142): 

143 """Update record representing re-assignment of DIASource to SSObject.""" 

144 

145 ssObjectId: int 

146 """ID of SSObject to re-associate to.""" 

147 

148 ssObjectReassocTimeMjdTai: float 

149 """Time when DIASource was re-associated from DIAObject to SSObject.""" 

150 

151 apdb_table: ClassVar[ApdbTables] = ApdbTables.DiaSource 

152 

153 

154@dataclass(kw_only=True) 

155class ApdbWithdrawDiaSourceRecord(ApdbUpdateRecord, DiaSourceId, update_type="withdraw_diasource"): 

156 """Update record representing withdrawal of DIASource.""" 

157 

158 timeWithdrawnMjdTai: float 

159 """Time when this record was marked invalid.""" 

160 

161 apdb_table: ClassVar[ApdbTables] = ApdbTables.DiaSource 

162 

163 

164@dataclass(kw_only=True) 

165class ApdbWithdrawDiaForcedSourceRecord( 

166 ApdbUpdateRecord, DiaForcedSourceId, update_type="withdraw_diaforcedsource" 

167): 

168 """Update record representing withdrawal of DIAForcedSource.""" 

169 

170 timeWithdrawnMjdTai: float 

171 """Time when this record was marked invalid.""" 

172 

173 apdb_table: ClassVar[ApdbTables] = ApdbTables.DiaForcedSource 

174 

175 

176@dataclass(kw_only=True) 

177class ApdbCloseDiaObjectValidityRecord(ApdbUpdateRecord, DiaObjectId, update_type="close_diaobject_validity"): 

178 """Record representing closing of the validity interval of DIAObject.""" 

179 

180 validityEndMjdTai: float 

181 """Time to set validityEnd to.""" 

182 

183 nDiaSources: int | None 

184 """New value for nDiaSources column for updated record, or None if 

185 nDiaSources does not change. 

186 """ 

187 

188 apdb_table: ClassVar[ApdbTables] = ApdbTables.DiaObject 

189 

190 

191@dataclass(kw_only=True) 

192class ApdbUpdateNDiaSourcesRecord(ApdbUpdateRecord, DiaObjectId, update_type="update_n_dia_sources"): 

193 """Record representing change in the number of associated sources of 

194 DIAObject. 

195 """ 

196 

197 nDiaSources: int 

198 """New value for nDiaSources column for updated record.""" 

199 

200 apdb_table: ClassVar[ApdbTables] = ApdbTables.DiaObject