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

1"""Exporter action.""" 

2 

3from typing import Optional, 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 project: Optional[str] 

27 """Project uid. Not used, the project is taken from the request.""" 

28 request: str 

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

30 output: str 

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

32 

33 

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

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

36 """Exporter action.""" 

37 

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

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

40 """Start a background export job.""" 

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

42 

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

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

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

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

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

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

49 res.output = { 

50 'numFiles': er.numFiles, 

51 'numFeaturesTotal': er.numFeaturesTotal, 

52 'numFeaturesExported': er.numFeaturesExported, 

53 } 

54 if er.path: 

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

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

57 

58 return res 

59 

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

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

62 """Cancel an export job.""" 

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

64 

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

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

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

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

69 if not er.path: 

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

71 return gws.ContentResponse( 

72 contentPath=er.path, 

73 mimeType=er.mimeType, 

74 ) 

75 

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

77 def do_export(self, p: CliParams): 

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

79 

80 root = gws.load_root() 

81 request = root.specs.read( 

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

83 'gws.ExportRequest', 

84 path=p.request, 

85 ) 

86 

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