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

1"""Base database model.""" 

2 

3from typing import Optional, Iterable 

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 """UID of the database provider.""" 

17 tableName: Optional[str] 

18 """Database table of the model.""" 

19 sqlFilter: Optional[str] 

20 """SQL condition added to every query of the model.""" 

21 

22 

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

24 pass 

25 

26 

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

28 """Base model for a database table. 

29 

30 Reads features from one table with SELECT queries built from the model 

31 fields, and creates, updates and deletes table rows. Provides subclasses 

32 with the database provider, access to the table and its columns, and the 

33 SELECT builder. 

34 """ 

35 

36 def configure(self): 

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

38 if not self.tableName: 

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

40 

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

42 self.configure_model() 

43 

44 def configure_provider(self): 

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

46 

47 ## 

48 

49 def describe(self): 

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

51 

52 def table(self): 

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

54 

55 def column(self, column_name): 

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

57 

58 def uid_column(self): 

59 if not self.uidName: 

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

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

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

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

64 

65 def uid_equals(self, uid): 

66 col = self.uid_column() 

67 if isinstance(uid, (str, bytes)) or not isinstance(uid, Iterable): 

68 return col == sa.bindparam(None, uid, type_=col.type) 

69 return col.in_(sa.bindparam(None, list(uid), expanding=True, type_=col.type)) 

70 

71 ## 

72 

73 def find_features(self, search, mc): 

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

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

76 

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

78 mc.search = search 

79 mc.dbSelect = gws.ModelSelectBuild( 

80 columns=[], 

81 geometryWhere=[], 

82 keywordWhere=[], 

83 order=[], 

84 where=[], 

85 ) 

86 

87 with self.db.connect(): 

88 for fld in self.fields: 

89 fld.before_select(mc) 

90 

91 sql = self.build_select(mc) 

92 if sql is None: 

93 return [] 

94 

95 features = self.fetch_features(sql) 

96 

97 for fld in self.fields: 

98 fld.after_select(features, mc) 

99 

100 return features 

101 

102 def fetch_features(self, select): 

103 features = [] 

104 

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

106 for row in conn.fetch_all(select): 

107 features.append( 

108 gws.base.feature.new( 

109 model=self, 

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

111 ) 

112 ) 

113 

114 return features 

115 

116 def build_select(self, mc): 

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

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

119 for s in sorts: 

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

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

122 

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

124 

125 if mc.search.uids: 

126 if not self.uidName: 

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

128 return 

129 sel = sel.where(self.uid_equals(mc.search.uids)) 

130 

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

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

133 return 

134 if mc.dbSelect.keywordWhere: 

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

136 

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

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

139 return 

140 if mc.dbSelect.geometryWhere: 

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

142 

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

144 if mc.search.extraWhere: 

145 for w in mc.search.extraWhere: 

146 sel = sel.where(w) 

147 

148 if self.sqlFilter: 

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

150 

151 cols = [] 

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

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

154 continue 

155 cols.append(col) 

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

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

158 continue 

159 cols.append(col) 

160 

161 sel = sel.add_columns(*cols) 

162 

163 if mc.dbSelect.order: 

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

165 

166 if mc.search.limit: 

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

168 

169 return sel 

170 

171 ## 

172 

173 def init_feature(self, feature, mc): 

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

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

176 

177 for fld in self.fields: 

178 fld.do_init(feature, mc) 

179 

180 for rf in feature.createWithFeatures: 

181 for fld in rf.model.fields: 

182 fld.do_init_related(feature, mc) 

183 

184 feature.isNew = True 

185 

186 def create_feature(self, feature, mc): 

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

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

189 

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

191 

192 related_models = [] 

193 for from_feature in feature.createWithFeatures: 

194 if from_feature.model not in related_models: 

195 related_models.append(from_feature.model) 

196 

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

198 for m in related_models: 

199 for fld in m.fields: 

200 fld.before_create_related(feature, mc) 

201 

202 for fld in self.fields: 

203 fld.before_create(feature, mc) 

204 

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

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

207 pk = rs.inserted_primary_key 

208 if not pk: 

209 feature.insertedPrimaryKey = None 

210 elif len(pk) == 1: 

211 feature.insertedPrimaryKey = pk[0] 

212 else: 

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

214 

215 for fld in self.fields: 

216 fld.after_create(feature, mc) 

217 

218 for m in related_models: 

219 for fld in m.fields: 

220 fld.after_create_related(feature, mc) 

221 

222 conn.commit() 

223 

224 return feature.insertedPrimaryKey 

225 

226 def update_feature(self, feature, mc): 

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

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

229 

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

231 

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

233 for fld in self.fields: 

234 fld.before_update(feature, mc) 

235 

236 if not feature.record.attributes: 

237 return feature.uid() 

238 

239 sql = self.table().update().where(self.uid_equals(feature.uid())).values(feature.record.attributes) 

240 conn.execute(sql) 

241 

242 for fld in self.fields: 

243 fld.after_update(feature, mc) 

244 

245 conn.commit() 

246 

247 return feature.uid() 

248 

249 def delete_feature(self, feature, mc): 

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

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

252 

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

254 for fld in self.fields: 

255 fld.before_delete(feature, mc) 

256 

257 sql = sa.delete(self.table()).where(self.uid_equals(feature.uid())) 

258 

259 conn.execute(sql) 

260 

261 for fld in self.fields: 

262 fld.after_delete(feature, mc) 

263 

264 conn.commit() 

265 

266 return feature.uid()