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

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.begin(): 

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.begin() 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.begin() 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 return feature.insertedPrimaryKey 

223 

224 def update_feature(self, feature, mc): 

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

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

227 

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

229 

230 with self.db.begin() as conn: 

231 for fld in self.fields: 

232 fld.before_update(feature, mc) 

233 

234 if not feature.record.attributes: 

235 return feature.uid() 

236 

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

238 conn.execute(sql) 

239 

240 for fld in self.fields: 

241 fld.after_update(feature, mc) 

242 

243 return feature.uid() 

244 

245 def delete_feature(self, feature, mc): 

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

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

248 

249 with self.db.begin() as conn: 

250 for fld in self.fields: 

251 fld.before_delete(feature, mc) 

252 

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

254 

255 conn.execute(sql) 

256 

257 for fld in self.fields: 

258 fld.after_delete(feature, mc) 

259 

260 return feature.uid()