Coverage for gws-app/gws/base/model/related_field.py: 48%

1"""Base class for related fields.""" 

2 

3from typing import Optional, cast 

4 

5import gws 

6import gws.base.model 

7import gws.lib.sa as sa 

8 

9from . import field 

10 

11 

12class Config(field.Config): 

13 pass 

14 

15 

16class Props(field.Props): 

17 pass 

18 

19 

20class Link(gws.Data): 

21 """Link table of a many-to-many relationship.""" 

22 

23 tableName: str 

24 """The link table name.""" 

25 srcKeyName: str 

26 """Column in the link table that refers to the source model.""" 

27 dstKeyName: str 

28 """Column in the link table that refers to the related model.""" 

29 

30 

31class RelRef(gws.Data): 

32 """One side of a relationship.""" 

33 

34 model: gws.DatabaseModel 

35 """The model.""" 

36 keyName: str 

37 """The column that takes part in the relationship.""" 

38 

39 

40class Relationship(gws.Data): 

41 """Relationship between the model of a field and related models.""" 

42 

43 src: RelRef 

44 """The side of the field's own model.""" 

45 dst: RelRef 

46 """The related side, for relationships with a single related model.""" 

47 dstList: list[RelRef] 

48 """The related sides.""" 

49 link: Link 

50 """The link table, for many-to-many relationships.""" 

51 deleteCascade: bool = False 

52 """When unlinking related features, delete them instead of clearing their key.""" 

53 

54 

55class Object(field.Object): 

56 """Base related field. 

57 

58 Links features of the field's model to features of other database models. 

59 Provides the relationship description, conversion of related features to and from 

60 props, and database helpers to read and update the keys of related features. 

61 Subclasses describe the link in `configure_relationship`. 

62 """ 

63 

64 model: gws.DatabaseModel 

65 """The model of this field.""" 

66 rel: Relationship 

67 """The relationship of this field.""" 

68 

69 def post_configure(self): 

70 self.configure_relationship() 

71 

72 def configure_relationship(self): 

73 """Set up the relationship ``rel``. 

74 

75 Called after configuration. The base implementation does nothing. 

76 """ 

77 pass 

78 

79 def key_column(self, ref: RelRef) -> sa.Column: 

80 """Return the key column of a relationship side. 

81 

82 Args: 

83 ref: The relationship side. 

84 

85 Returns: 

86 The column ``ref.keyName`` of ``ref.model``. 

87 """ 

88 return ref.model.column(ref.keyName) 

89 

90 def link_table(self) -> sa.Table: 

91 """Return the link table of the relationship.""" 

92 return self.model.db.table(self.rel.link.tableName) 

93 

94 def link_column(self, name: str) -> sa.Column: 

95 """Return a column of the link table. 

96 

97 Args: 

98 name: The column name. 

99 

100 Returns: 

101 The column. 

102 """ 

103 return self.model.db.column(self.link_table(), name) 

104 

105 def configure_widget(self): 

106 if not super().configure_widget(): 

107 if self.attributeType == gws.AttributeType.feature: 

108 self.widget = self.root.create_shared(gws.ext.object.modelWidget, type='featureSelect') 

109 return True 

110 if self.attributeType == gws.AttributeType.featurelist: 

111 self.widget = self.root.create_shared(gws.ext.object.modelWidget, type='featureList') 

112 return True 

113 

114 def get_model(self, uid: str) -> gws.DatabaseModel: 

115 """Return a model by uid. 

116 

117 Args: 

118 uid: The model uid. 

119 

120 Returns: 

121 The model. 

122 

123 Raises: 

124 gws.ConfigurationError: If the model is not found. 

125 """ 

126 mod = self.root.get(uid) 

127 if not mod: 

128 raise gws.ConfigurationError(f'model {uid!r} not found') 

129 return cast(gws.DatabaseModel, mod) 

130 

131 def find_relatable_features(self, search, mc): 

132 return [f for dst in self.rel.dstList for f in dst.model.find_features(search, mc)] 

133 

134 def related_field(self, dst: RelRef) -> Optional[gws.ModelField]: 

135 """Find the field of a related model that describes the reverse relationship. 

136 

137 Args: 

138 dst: The related side of the relationship. 

139 

140 Returns: 

141 The field of ``dst.model`` whose relationship starts at ``dst.model`` and ``dst.keyName``, 

142 or None if there is no such field. 

143 """ 

144 for fld in dst.model.fields: 

145 rel2 = cast(Relationship, getattr(fld, 'rel', None)) 

146 if not rel2: 

147 continue 

148 if rel2.src.model == dst.model and rel2.src.keyName == dst.keyName: 

149 return fld 

150 

151 def do_init_related(self, dst_feature, mc): 

152 our_features = [f for f in dst_feature.createWithFeatures if f.model == self.model] 

153 if not our_features: 

154 return 

155 

156 for dst in self.rel.dstList: 

157 if dst.model == dst_feature.model: 

158 fld = self.related_field(dst) 

159 if fld: 

160 if fld.attributeType == gws.AttributeType.feature: 

161 dst_feature.attributes[fld.name] = our_features[0] 

162 if fld.attributeType == gws.AttributeType.featurelist: 

163 dst_feature.attributes.setdefault(fld.name, []).extend(our_features) 

164 

165 def related_models(self): 

166 return [dst.model for dst in self.rel.dstList] 

167 

168 ## 

169 

170 def to_props(self, feature, mc): 

171 if not mc.user.can_read(self) or mc.relDepth >= mc.maxDepth: 

172 return 

173 

174 value = feature.get(self.name) 

175 if not value: 

176 return 

177 if not isinstance(value, list): 

178 value = [value] 

179 

180 mc2 = gws.base.model.secondary_context(mc) 

181 res = [] 

182 

183 for v in value: 

184 related = cast(gws.Feature, v) 

185 if related: 

186 p = related.model.feature_to_props(related, mc2) 

187 if p: 

188 res.append(p) 

189 

190 if self.attributeType == gws.AttributeType.featurelist: 

191 feature.props.attributes[self.name] = res 

192 elif res: 

193 feature.props.attributes[self.name] = res[0] 

194 

195 def from_props(self, feature, mc): 

196 if mc.relDepth >= mc.maxDepth: 

197 return 

198 

199 value = feature.props.attributes.get(self.name) 

200 if not value: 

201 return 

202 if not isinstance(value, list): 

203 value = [value] 

204 

205 mc2 = gws.base.model.secondary_context(mc) 

206 res = [] 

207 dst_model_map: dict[str, gws.Model] = {dst.model.uid: dst.model for dst in self.rel.dstList} 

208 

209 for v in value: 

210 rel_props = cast(gws.FeatureProps, gws.u.to_data_object(v)) 

211 if rel_props: 

212 dst_model = dst_model_map.get(rel_props.modelUid) 

213 if dst_model: 

214 related = dst_model.feature_from_props(rel_props, mc2) 

215 if related: 

216 res.append(related) 

217 

218 if self.attributeType == gws.AttributeType.featurelist: 

219 feature.set(self.name, res) 

220 elif res: 

221 feature.set(self.name, res[0]) 

222 

223 ## 

224 

225 def key_for_uid( 

226 self, 

227 model: gws.DatabaseModel, 

228 key_column: sa.Column, 

229 uid: gws.FeatureUid, 

230 mc: gws.ModelContext, 

231 ): 

232 """Return the key value of the feature with the given uid. 

233 

234 Args: 

235 model: The model to query. 

236 key_column: The key column. 

237 uid: The feature uid. 

238 mc: The model context. 

239 

240 Returns: 

241 The key value, or None if the feature is not found. 

242 """ 

243 sql = sa.select(key_column).where(model.uid_equals(uid)) 

244 with model.db.connect() as conn: 

245 rs = list(conn.execute(sql)) 

246 return rs[0][0] if rs else None