Coverage for gws-app/gws/plugin/model_field/file/__init__.py: 86%

1"""File field.""" 

2 

3from typing import Optional, cast 

4 

5import gws 

6import gws.base.database.model 

7import gws.base.model.field 

8import gws.lib.image 

9import gws.lib.mime 

10import gws.lib.sa as sa 

11 

12gws.ext.new.modelField('file') 

13 

14 

15class Config(gws.base.model.field.Config): 

16 """Configuration for the file field.""" 

17 

18 contentColumn: str = '' 

19 """Column name for the file content, if stored in the database.""" 

20 pathColumn: str = '' 

21 """Column name for the file path, if stored in the filesystem.""" 

22 nameColumn: str = '' 

23 """Column name for the file name, if stored in the database or filesystem.""" 

24 

25 

26class Props(gws.base.model.field.Props): 

27 pass 

28 

29 

30class FileInputProps(gws.Data): 

31 content: bytes 

32 name: str 

33 

34 

35class ServerFileProps(gws.Data): 

36 downloadUrl: str 

37 extension: str 

38 label: str 

39 previewUrl: str 

40 size: int 

41 

42 

43class ClientFileProps(gws.Data): 

44 name: str 

45 content: bytes 

46 

47 

48class FileValue(gws.Data): 

49 content: bytes 

50 name: str 

51 path: str 

52 size: int 

53 

54 

55_PREVIEW_SIZE = 120, 120 

56_PREVIEW_MIME = gws.lib.mime.PNG 

57_PREVIEW_CACHE_LIFE_TIME = 24 * 3600 

58_PREVIEW_MAX_PIXELS = 40_000_000 

59_PREVIEW_BIG_FILE_SIZE = 1024 * 1024 

60 

61 

62class Object(gws.base.model.field.Object): 

63 model: gws.DatabaseModel 

64 

65 attributeType = gws.AttributeType.file 

66 

67 contentColumn: Optional[sa.Column] = None 

68 pathColumn: Optional[sa.Column] = None 

69 nameColumn: Optional[sa.Column] = None 

70 

71 def __getstate__(self): 

72 return gws.u.omit(vars(self), 'cols') 

73 

74 def post_configure(self): 

75 self.configure_columns() 

76 

77 def activate(self): 

78 self.configure_columns() 

79 

80 def configure_columns(self): 

81 model = cast(gws.base.database.model.Object, self.model) 

82 

83 p = self.cfg('contentColumn') 

84 self.contentColumn = model.column(p) if p else None 

85 

86 p = self.cfg('pathColumn') 

87 self.pathColumn = model.column(p) if p else None 

88 

89 p = self.cfg('nameColumn') 

90 self.nameColumn = model.column(p) if p else None 

91 

92 if self.contentColumn is None and self.pathColumn is None: 

93 raise gws.ConfigurationError('contentColumn or pathColumn must be set') 

94 

95 if not self.model.uidName: 

96 raise gws.ConfigurationError('file fields require a primary key') 

97 

98 def configure_widget(self): 

99 if not super().configure_widget(): 

100 self.widget = self.root.create_shared(gws.ext.object.modelWidget, type='file') 

101 return True 

102 

103 ## 

104 

105 def before_select(self, mc): 

106 mc.dbSelect.columns.extend(self.select_columns(mc)) 

107 

108 def after_select(self, features, mc): 

109 for feature in features: 

110 self.from_record(feature, mc) 

111 

112 def before_create(self, feature, mc): 

113 self.to_record(feature, mc) 

114 

115 def before_update(self, feature, mc): 

116 self.to_record(feature, mc) 

117 

118 def from_record(self, feature, mc): 

119 feature.set(self.name, self.load_value(feature.record.attributes, mc)) 

120 

121 def to_record(self, feature, mc): 

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

123 return 

124 

125 # @TODO store in the filesystem 

126 

127 fv = cast(FileValue, feature.get(self.name)) 

128 if not fv: 

129 return 

130 if self.contentColumn is not None: 

131 feature.record.attributes[self.contentColumn.name] = fv.content 

132 if self.nameColumn is not None: 

133 feature.record.attributes[self.nameColumn.name] = fv.name 

134 

135 # @TODO merge with scalar_field? 

136 

137 def from_props(self, feature, mc): 

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

139 if value is not None: 

140 value = self.prop_to_python(feature, value, mc) 

141 if value is not None: 

142 feature.set(self.name, value) 

143 

144 def to_props(self, feature, mc): 

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

146 return 

147 value = feature.get(self.name) 

148 if value is not None: 

149 value = self.python_to_prop(feature, value, mc) 

150 if value is not None: 

151 feature.props.attributes[self.name] = value 

152 

153 ## 

154 

155 def can_preview(self, mime) -> bool: 

156 return mime.startswith('image/') and mime != gws.lib.mime.SVG 

157 

158 def prop_to_python(self, feature, value, mc) -> FileValue: 

159 try: 

160 return FileValue( 

161 content=gws.u.get(value, 'content'), 

162 name=gws.u.get(value, 'name'), 

163 ) 

164 except ValueError: 

165 return gws.ErrorValue 

166 

167 def python_to_prop(self, feature, value, mc) -> ServerFileProps: 

168 fv = cast(FileValue, value) 

169 

170 mime = self.get_mime_type(fv) 

171 ext = gws.lib.mime.extension_for(mime) 

172 

173 p = ServerFileProps( 

174 # @TODO use a template 

175 label=fv.name or '', 

176 extension=ext, 

177 size=fv.size or 0, 

178 previewUrl='', 

179 downloadUrl='', 

180 ) 

181 

182 name = fv.name or f'gws.{ext}' 

183 

184 url_args = dict( 

185 projectUid=mc.project.uid, 

186 modelUid=self.model.uid, 

187 fieldName=self.name, 

188 featureUid=feature.uid(), 

189 ) 

190 

191 if self.can_preview(mime): 

192 p.previewUrl = gws.u.action_url_path('webFile', preview=1, **url_args) + '/' + name 

193 

194 p.downloadUrl = gws.u.action_url_path('webFile', **url_args) + '/' + name 

195 

196 return p 

197 

198 ## 

199 

200 def get_mime_type(self, fv: FileValue) -> str: 

201 if fv.path: 

202 return gws.lib.mime.for_path(fv.path) 

203 if fv.name: 

204 return gws.lib.mime.for_path(fv.name) 

205 # @TODO guess mime from content? 

206 return gws.lib.mime.BIN 

207 

208 def handle_web_file_request(self, feature_uid: str, preview: bool, mc: gws.ModelContext) -> Optional[gws.ContentResponse]: 

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

210 return 

211 

212 if self.contentColumn is None: 

213 # @TODO serve files stored in the filesystem 

214 return 

215 

216 search = gws.SearchQuery(uids=[feature_uid]) 

217 if preview: 

218 # for small files, fetch content md5 and content, for big files only md5 

219 search.extraColumns = [ 

220 sa.func.md5(self.contentColumn).label(f'{self.name}_preview_md5'), 

221 sa.case( 

222 (sa.func.length(self.contentColumn) < _PREVIEW_BIG_FILE_SIZE, self.contentColumn), 

223 else_=sa.null(), 

224 ).label(f'{self.name}_preview_content'), 

225 ] 

226 else: 

227 search.extraColumns = [self.contentColumn] 

228 

229 features = self.model.find_features(search, mc) 

230 if not features: 

231 return 

232 

233 feature = features[0] 

234 

235 fv = cast(FileValue, feature.get(self.name)) 

236 if not fv: 

237 return 

238 

239 if not preview: 

240 # download complete file content 

241 mime = self.get_mime_type(fv) 

242 return gws.ContentResponse( 

243 content=fv.content, 

244 contentFilename=fv.name or f'gws.{gws.lib.mime.extension_for(mime)}', 

245 mime=mime, 

246 ) 

247 

248 # preview 

249 

250 if not self.can_preview(self.get_mime_type(fv)): 

251 return 

252 

253 md5 = feature.record.attributes.get(f'{self.name}_preview_md5') 

254 if not md5: 

255 return 

256 

257 cache_key = gws.u.sha256( 

258 [ 

259 self.model.uid, 

260 self.name, 

261 feature.uid(), 

262 md5, 

263 _PREVIEW_SIZE, 

264 _PREVIEW_MIME, 

265 ] 

266 ) 

267 

268 def make_preview(): 

269 content = feature.record.attributes.get(f'{self.name}_preview_content') 

270 

271 if content is None: 

272 # big file 

273 search = gws.SearchQuery(uids=[feature.uid()]) 

274 search.extraColumns = [self.contentColumn] 

275 features = self.model.find_features(search, mc) 

276 if not features: 

277 raise gws.NotFoundError(f'file preview: no feature {feature.uid()!r}') 

278 fv = cast(FileValue, features[0].get(self.name)) 

279 if not fv or fv.content is None: 

280 raise gws.NotFoundError(f'file preview: no content for {feature.uid()!r}') 

281 content = fv.content 

282 

283 try: 

284 return gws.lib.image.thumbnail( 

285 content, 

286 _PREVIEW_SIZE, 

287 max_pixels=_PREVIEW_MAX_PIXELS, 

288 mime=_PREVIEW_MIME, 

289 ) 

290 except gws.lib.image.Error as exc: 

291 raise gws.NotFoundError(f'file preview: {exc}') from exc 

292 

293 cache_dir = gws.u.ensure_dir(gws.c.CACHE_DIR + '/preview') 

294 cache_path = cache_dir + f'/{cache_key}.{gws.lib.mime.extension_for(_PREVIEW_MIME)}' 

295 

296 return gws.ContentResponse( 

297 contentPath=gws.u.get_cached_file(cache_path, _PREVIEW_CACHE_LIFE_TIME, make_preview), 

298 mime=_PREVIEW_MIME, 

299 ) 

300 

301 ## 

302 

303 def select_columns(self, mc): 

304 cs = [] 

305 

306 if self.contentColumn is not None: 

307 cs.append(sa.func.length(self.contentColumn).label(f'{self.name}_length')) 

308 if self.pathColumn is not None: 

309 cs.append(self.pathColumn) 

310 if self.nameColumn is not None: 

311 cs.append(self.nameColumn) 

312 

313 return cs 

314 

315 def load_value(self, attributes: dict, mc) -> Optional[FileValue]: 

316 d = {} 

317 

318 if self.contentColumn is not None: 

319 d['size'] = attributes.get(f'{self.name}_length') 

320 d['content'] = attributes.get(self.contentColumn.name) 

321 if self.pathColumn is not None: 

322 d['path'] = attributes.get(self.pathColumn.name) 

323 if self.nameColumn is not None: 

324 d['name'] = attributes.get(self.nameColumn.name) 

325 

326 if d: 

327 return FileValue(**d)