Coverage for gws-app/gws/base/exporter/action.py: 0%

1"""Provides the printing API.""" 

2 

3from typing import Optional, cast 

4 

5import gws 

6import gws.base.action 

7import gws.config 

8import gws.lib.jsonx 

9import gws.lib.mime 

10 

11gws.ext.new.action('exporter') 

12 

13 

14class Config(gws.base.action.Config): 

15 """Configuration for the exporter action. (added in 8.4)""" 

16 

17 pass 

18 

19 

20class Props(gws.base.action.Props): 

21 pass 

22 

23 

24class CliParams(gws.CliParams): 

25 project: Optional[str] 

26 """project uid""" 

27 request: str 

28 """path to request.json""" 

29 output: str 

30 """output path""" 

31 

32 

33class Object(gws.base.action.Object): 

34 @gws.ext.command.api('exporterStart') 

35 def exporter_start(self, req: gws.WebRequester, p: gws.ExportRequest) -> gws.JobStatusResponse: 

36 """Start a background export job""" 

37 return self.root.app.exporterMgr.start_export_job(p, req.user) 

38 

39 @gws.ext.command.api('exporterStatus') 

40 def exporter_status(self, req: gws.WebRequester, p: gws.JobRequest) -> gws.JobStatusResponse: 

41 res = self.root.app.jobMgr.handle_status_request(req, p) 

42 if res.state == gws.JobState.complete: 

43 er = gws.ExportResult(self.root.app.jobMgr.require_result(req, p)) 

44 res.output = { 

45 'numFiles': er.numFiles, 

46 'numFeaturesTotal': er.numFeaturesTotal, 

47 'numFeaturesExported': er.numFeaturesExported, 

48 } 

49 if er.path: 

50 ext = gws.lib.mime.extension_for(er.mime) or 'bin' 

51 res.output['url'] = gws.u.action_url_path('exporterOutput', projectUid=p.projectUid, jobUid=res.jobUid) + f'/gws.{ext}' 

52 

53 return res 

54 

55 @gws.ext.command.api('exporterCancel') 

56 def exporter_cancel(self, req: gws.WebRequester, p: gws.JobRequest) -> gws.JobStatusResponse: 

57 return self.root.app.jobMgr.handle_cancel_request(req, p) 

58 

59 @gws.ext.command.get('exporterOutput') 

60 def exporter_output(self, req: gws.WebRequester, p: gws.JobRequest) -> gws.ContentResponse: 

61 er = gws.ExportResult(self.root.app.jobMgr.require_result(req, p)) 

62 if not er.path: 

63 raise gws.NotFoundError('export result not found') 

64 return gws.ContentResponse( 

65 contentPath=er.path, 

66 mime=er.mime, 

67 ) 

68 

69 @gws.ext.command.cli('exporterExport') 

70 def do_export(self, p: CliParams): 

71 """Export using the specified params""" 

72 

73 root = gws.config.load() 

74 request = root.specs.read( 

75 gws.lib.jsonx.from_path(p.request), 

76 'gws.ExportRequest', 

77 path=p.request, 

78 ) 

79 

80 root.app.exporterMgr.exec_export(cast(gws.ExportRequest, request), p.output)