Coverage for gws-app/gws/base/database/model.py: 85%

1"""Database-based models.""" 

2 

3from typing import Optional 

4 

5import gws 

6import gws.base.feature 

7import gws.base.model 

8import gws.config.util 

9import gws.lib.sa as sa 

10 

11 

12class Config(gws.base.model.Config): 

13 """Configuration for the database model.""" 

14 

15 dbUid: Optional[str] 

16 """Database provider uid.""" 

17 tableName: Optional[str] 

18 """Table name for the model.""" 

19 sqlFilter: Optional[str] 

20 """Extra SQL filter.""" 

21 

22 

23class Props(gws.base.model.Props): 

24 pass 

25 

26 

27class Object(gws.base.model.Object, gws.DatabaseModel): 

28 def configure(self): 

29 self.tableName = self.cfg('tableName') or self.cfg('_defaultTableName') 

30 if not self.tableName: 

31 raise gws.ConfigurationError(f'table name missing in model {self!r}') 

32 

33 self.sqlFilter = self.cfg('sqlFilter') 

34 self.configure_model() 

35 

36 def configure_provider(self): 

37 return gws.config.util.configure_database_provider_for(self) 

38 

39 ## 

40 

41 def describe(self): 

42 return self.db.describe(self.tableName) 

43 

44 def table(self): 

45 return self.db.table(self.tableName) 

46 

47 def column(self, column_name): 

48 return self.db.column(self.table(), column_name) 

49 

50 def uid_column(self): 

51 if not self.uidName: 

52 raise gws.Error(f'no primary key found for table {self.tableName!r}') 

53 if not self.db.has_column(self.table(), self.uidName): 

54 raise gws.Error(f'invalid primary key {self.uidName!r} for table {self.tableName!r}') 

55 return self.db.column(self.table(), self.uidName) 

56 

57 ## 

58 

59 def find_features(self, search, mc): 

60 if not mc.user.can_read(self): 

61 raise gws.ForbiddenError(f'model {self.uid!r} can_read=False') 

62 

63 mc = gws.base.model.copy_context(mc) 

64 mc.search = search 

65 mc.dbSelect = gws.ModelSelectBuild( 

66 columns=[], 

67 geometryWhere=[], 

68 keywordWhere=[], 

69 order=[], 

70 where=[], 

71 ) 

72 

73 with self.db.connect(): 

74 for fld in self.fields: 

75 fld.before_select(mc) 

76 

77 sql = self.build_select(mc) 

78 if sql is None: 

79 return [] 

80 

81 features = self.fetch_features(sql) 

82 

83 for fld in self.fields: 

84 fld.after_select(features, mc) 

85 

86 return features 

87 

88 def fetch_features(self, select): 

89 features = [] 

90 

91 with self.db.connect() as conn: 

92 for row in conn.fetch_all(select): 

93 features.append( 

94 gws.base.feature.new( 

95 model=self, 

96 record=gws.FeatureRecord(attributes=row), 

97 ) 

98 ) 

99 

100 return features 

101 

102 def build_select(self, mc): 

103 # @TODO sorting should be handled on the field level 

104 sorts = mc.search.sort or self.defaultSort or [] 

105 for s in sorts: 

106 fn = sa.desc if s.reverse else sa.asc 

107 mc.dbSelect.order.append(fn(self.column(s.fieldName))) 

108 

109 sel = sa.select().select_from(self.table()) 

110 

111 if mc.search.uids: 

112 if not self.uidName: 

113 gws.log.debug(f'build_select: {self}: no primary key for {self.tableName=}') 

114 return 

115 sel = sel.where(self.uid_column().in_(mc.search.uids)) 

116 

117 if mc.search.keyword and not mc.dbSelect.keywordWhere: 

118 gws.log.debug(f'build_select: {self}: no keyword where') 

119 return 

120 if mc.dbSelect.keywordWhere: 

121 sel = sel.where(sa.or_(*mc.dbSelect.keywordWhere)) 

122 

123 if mc.search.shape and not mc.dbSelect.geometryWhere: 

124 gws.log.debug(f'build_select: {self}: no geometry where') 

125 return 

126 if mc.dbSelect.geometryWhere: 

127 sel = sel.where(sa.or_(*mc.dbSelect.geometryWhere)) 

128 

129 sel = sel.where(*mc.dbSelect.where) 

130 if mc.search.extraWhere: 

131 for w in mc.search.extraWhere: 

132 sel = sel.where(w) 

133 

134 if self.sqlFilter: 

135 sel = sel.where(sa.text('(' + self.sqlFilter + ')')) 

136 

137 cols = [] 

138 for col in mc.dbSelect.columns or []: 

139 if any(col is c for c in cols): 

140 continue 

141 cols.append(col) 

142 for col in mc.search.extraColumns or []: 

143 if any(col is c for c in cols): 

144 continue 

145 cols.append(col) 

146 

147 sel = sel.add_columns(*cols) 

148 

149 if mc.dbSelect.order: 

150 sel = sel.order_by(*mc.dbSelect.order) 

151 

152 if mc.search.limit: 

153 sel = sel.limit(mc.search.limit) 

154 

155 return sel 

156 

157 ## 

158 

159 def init_feature(self, feature, mc): 

160 if not mc.user.can_create(self): 

161 raise gws.ForbiddenError(f'model {self.uid!r} can_create=False') 

162 

163 for fld in self.fields: 

164 fld.do_init(feature, mc) 

165 

166 for rf in feature.createWithFeatures: 

167 for fld in rf.model.fields: 

168 fld.do_init_related(feature, mc) 

169 

170 feature.isNew = True 

171 

172 def create_feature(self, feature, mc): 

173 if not mc.user.can_create(self): 

174 raise gws.ForbiddenError(f'model {self.uid!r} can_create=False') 

175 

176 feature.record = gws.FeatureRecord(attributes={}, meta={}) 

177 

178 related_models = [] 

179 for from_feature in feature.createWithFeatures: 

180 if from_feature.model not in related_models: 

181 related_models.append(from_feature.model) 

182 

183 with self.db.connect() as conn: 

184 for m in related_models: 

185 for fld in m.fields: 

186 fld.before_create_related(feature, mc) 

187 

188 for fld in self.fields: 

189 fld.before_create(feature, mc) 

190 

191 sql = sa.insert(self.table()) 

192 rs = conn.execute(sql, feature.record.attributes) 

193 pk = rs.inserted_primary_key 

194 if not pk: 

195 feature.insertedPrimaryKey = None 

196 elif len(pk) == 1: 

197 feature.insertedPrimaryKey = pk[0] 

198 else: 

199 raise gws.Error(f'composite primary keys not supported for {self.tableName!r}') 

200 

201 for fld in self.fields: 

202 fld.after_create(feature, mc) 

203 

204 for m in related_models: 

205 for fld in m.fields: 

206 fld.after_create_related(feature, mc) 

207 

208 conn.commit() 

209 

210 return feature.insertedPrimaryKey 

211 

212 def update_feature(self, feature, mc): 

213 if not mc.user.can_write(self): 

214 raise gws.ForbiddenError(f'model {self.uid!r} can_write=False') 

215 

216 feature.record = gws.FeatureRecord(attributes={}, meta={}) 

217 

218 with self.db.connect() as conn: 

219 for fld in self.fields: 

220 fld.before_update(feature, mc) 

221 

222 if not feature.record.attributes: 

223 return feature.uid() 

224 

225 sql = self.table().update().where(self.uid_column().__eq__(feature.uid())).values(feature.record.attributes) 

226 conn.execute(sql) 

227 

228 for fld in self.fields: 

229 fld.after_update(feature, mc) 

230 

231 conn.commit() 

232 

233 return feature.uid() 

234 

235 def delete_feature(self, feature, mc): 

236 if not mc.user.can_delete(self): 

237 raise gws.ForbiddenError(f'model {self.uid!r} can_delete=False') 

238 

239 with self.db.connect() as conn: 

240 for fld in self.fields: 

241 fld.before_delete(feature, mc) 

242 

243 sql = sa.delete(self.table()).where(self.uid_column().__eq__(feature.uid())) 

244 

245 conn.execute(sql) 

246 

247 for fld in self.fields: 

248 fld.after_delete(feature, mc) 

249 

250 conn.commit() 

251 

252 return feature.uid()