Coverage for .tox / coverage / lib / python3.11 / site-packages / wuttaweb / progress.py: 100%

1# -*- coding: utf-8; -*- 

2################################################################################ 

3# 

4# wuttaweb -- Web App for Wutta Framework 

5# Copyright © 2024-2025 Lance Edgar 

6# 

7# This file is part of Wutta Framework. 

8# 

9# Wutta Framework is free software: you can redistribute it and/or modify it 

10# under the terms of the GNU General Public License as published by the Free 

11# Software Foundation, either version 3 of the License, or (at your option) any 

12# later version. 

13# 

14# Wutta Framework is distributed in the hope that it will be useful, but 

15# WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or 

16# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for 

17# more details. 

18# 

19# You should have received a copy of the GNU General Public License along with 

20# Wutta Framework. If not, see <http://www.gnu.org/licenses/>. 

21# 

22################################################################################ 

23""" 

24Progress Indicators 

25""" 

26 

27from wuttjamaican.progress import ProgressBase 

28 

29from beaker.session import Session as BeakerSession 

30 

31 

32def get_basic_session(request, **kwargs): 

33 """ 

34 Create/get a "basic" Beaker session object. 

35 """ 

36 kwargs["use_cookies"] = False 

37 return BeakerSession(request, **kwargs) 

38 

39 

40def get_progress_session(request, key, **kwargs): 

41 """ 

42 Create/get a Beaker session object, to be used for progress. 

43 """ 

44 kwargs["id"] = f"{request.session.id}.progress.{key}" 

45 return get_basic_session(request, **kwargs) 

46 

47 

48class SessionProgress(ProgressBase): # pylint: disable=too-many-instance-attributes 

49 """ 

50 Progress indicator which uses Beaker session storage to track 

51 current status. 

52 

53 This is a subclass of 

54 :class:`wuttjamaican:wuttjamaican.progress.ProgressBase`. 

55 

56 A view callable can create one of these, and then pass it into 

57 :meth:`~wuttjamaican.app.AppHandler.progress_loop()` or similar. 

58 

59 As the loop updates progress along the way, this indicator will 

60 update the Beaker session to match. 

61 

62 Separately then, the client side can send requests for the 

63 :func:`~wuttaweb.views.progress.progress()` view, to fetch current 

64 status out of the Beaker session. 

65 

66 :param request: Current :term:`request` object. 

67 

68 :param key: Unique key for this progress indicator. Used to 

69 distinguish progress indicators in the Beaker session. 

70 

71 Note that in addition to 

72 :meth:`~wuttjamaican:wuttjamaican.progress.ProgressBase.update()` 

73 and 

74 :meth:`~wuttjamaican:wuttjamaican.progress.ProgressBase.finish()` 

75 this progres class has some extra attributes and methods: 

76 

77 .. attribute:: success_msg 

78 

79 Optional message to display to the user (via session flash) 

80 when the operation completes successfully. 

81 

82 .. attribute:: success_url 

83 

84 URL to which user should be redirected, once the operation 

85 completes. 

86 

87 .. attribute:: error_url 

88 

89 URL to which user should be redirected, if the operation 

90 encounters an error. If not specified, will fall back to 

91 :attr:`success_url`. 

92 """ 

93 

94 def __init__( # pylint: disable=too-many-arguments,too-many-positional-arguments,super-init-not-called 

95 self, request, key, success_msg=None, success_url=None, error_url=None 

96 ): 

97 self.request = request 

98 self.config = self.request.wutta_config 

99 self.app = self.config.get_app() 

100 self.key = key 

101 self.success_msg = success_msg 

102 self.success_url = success_url 

103 self.error_url = error_url or self.success_url 

104 self.session = get_progress_session(request, key) 

105 self.clear() 

106 

107 def __call__(self, message, maximum): 

108 self.clear() 

109 self.session["message"] = message 

110 self.session["maximum"] = maximum 

111 self.session["maximum_display"] = f"{maximum:,d}" 

112 self.session["value"] = 0 

113 self.session.save() 

114 return self 

115 

116 def clear(self): # pylint: disable=empty-docstring 

117 """ """ 

118 self.session.clear() 

119 self.session["complete"] = False 

120 self.session["error"] = False 

121 self.session.save() 

122 

123 def update(self, value): # pylint: disable=empty-docstring 

124 """ """ 

125 self.session.load() 

126 self.session["value"] = value 

127 self.session.save() 

128 

129 def handle_error(self, error, error_url=None): 

130 """ 

131 This should be called by the view code, within a try/catch 

132 block upon error. 

133 

134 The session storage will be updated to reflect details of the 

135 error. Next time client requests the progress status it will 

136 learn of the error and redirect the user. 

137 

138 :param error: :class:`python:Exception` instance. 

139 

140 :param error_url: Optional redirect URL; if not specified 

141 :attr:`error_url` is used. 

142 """ 

143 self.session.load() 

144 self.session["error"] = True 

145 self.session["error_msg"] = self.app.render_error(error) 

146 self.session["error_url"] = error_url or self.error_url 

147 self.session.save() 

148 

149 def handle_success(self, success_msg=None, success_url=None): 

150 """ 

151 This should be called by the view code, when the long-running 

152 operation completes. 

153 

154 The session storage will be updated to reflect the completed 

155 status. Next time client requests the progress status it will 

156 discover it has completed, and redirect the user. 

157 

158 :param success_msg: Optional message to display to the user 

159 (via session flash) when the operation completes 

160 successfully. If not specified :attr:`success_msg` (or 

161 nothing) is used 

162 

163 :param success_url: Optional redirect URL; if not specified 

164 :attr:`success_url` is used. 

165 """ 

166 self.session.load() 

167 self.session["complete"] = True 

168 self.session["success_msg"] = success_msg or self.success_msg 

169 self.session["success_url"] = success_url or self.success_url 

170 self.session.save()