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

1"""Exporter action.""" 

2 

3from typing import cast 

4 

5import gws 

6import gws.base.action 

7import gws.lib.jsonx 

8import gws.lib.mime 

9 

10 

11@gws.ext.config.action('exporter') 

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

13 """Feature export in the client, run as background jobs.""" 

14 

15 pass 

16 

17 

18@gws.ext.props.action('exporter') 

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

20 pass 

21 

22 

23class CliParams(gws.CliParams): 

24 """Parameters of the ``gws exporter export`` command.""" 

25 

26 request: str 

27 """Path to a JSON file with a ``gws.ExportRequest``.""" 

28 output: str 

29 """Path to copy the export result to.""" 

30 

31 

32@gws.ext.object.action('exporter') 

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

34 """Exporter action.""" 

35 

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

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

38 """Start a background export job.""" 

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

40 

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

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

43 """Return the status of an export job.""" 

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

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

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

47 res.output = { 

48 'numFiles': er.numFiles, 

49 'numFeaturesTotal': er.numFeaturesTotal, 

50 'numFeaturesExported': er.numFeaturesExported, 

51 } 

52 if er.path: 

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

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

55 

56 return res 

57 

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

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

60 """Cancel an export job.""" 

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

62 

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

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

65 """Return the result file of a completed export job.""" 

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

67 if not er.path: 

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

69 return gws.ContentResponse( 

70 contentPath=er.path, 

71 mimeType=er.mimeType, 

72 ) 

73 

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

75 def do_export(self, p: CliParams): 

76 """Run an export from the command line.""" 

77 

78 root = gws.load_root() 

79 request = root.specs.read( 

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

81 'gws.ExportRequest', 

82 path=p.request, 

83 ) 

84 

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