Coverage for pygeoapi/process/manager/tinydb_.py: 63%

1# ================================================================= 

2# 

3# Authors: Tom Kralidis <tomkralidis@gmail.com> 

4# 

5# Copyright (c) 2026 Tom Kralidis 

6# 

7# Permission is hereby granted, free of charge, to any person 

8# obtaining a copy of this software and associated documentation 

9# files (the "Software"), to deal in the Software without 

10# restriction, including without limitation the rights to use, 

11# copy, modify, merge, publish, distribute, sublicense, and/or sell 

12# copies of the Software, and to permit persons to whom the 

13# Software is furnished to do so, subject to the following 

14# conditions: 

15# 

16# The above copyright notice and this permission notice shall be 

17# included in all copies or substantial portions of the Software. 

18# 

19# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, 

20# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES 

21# OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND 

22# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT 

23# HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, 

24# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING 

25# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR 

26# OTHER DEALINGS IN THE SOFTWARE. 

27# 

28# ================================================================= 

29 

30 

31from contextlib import contextmanager 

32import json 

33import logging 

34from pathlib import Path 

35from typing import Any, Tuple 

36 

37import tinydb 

38from filelock import FileLock 

39 

40from pygeoapi.formats import FORMAT_TYPES, F_JSON, F_JSONLD 

41from pygeoapi.process.base import ( 

42 JobNotFoundError, 

43 JobResultNotFoundError, 

44) 

45from pygeoapi.process.manager.base import BaseManager 

46from pygeoapi.util import JobStatus 

47 

48LOGGER = logging.getLogger(__name__) 

49 

50 

51class TinyDBManager(BaseManager): 

52 """TinyDB Manager""" 

53 

54 def __init__(self, manager_def: dict): 

55 """ 

56 Initialize object 

57 

58 :param manager_def: manager definition 

59 

60 :returns: `pygeoapi.process.manager.base.BaseManager` 

61 """ 

62 

63 super().__init__(manager_def) 

64 self.is_async = True 

65 self.supports_subscribing = True 

66 

67 @contextmanager 

68 def _db(self): 

69 self.lock = FileLock(f"{self.connection}.lock") 

70 with self.lock: 

71 with tinydb.TinyDB(self.connection) as db: 

72 yield db 

73 

74 def destroy(self) -> bool: 

75 """ 

76 Destroy manager 

77 

78 :returns: `bool` status of result 

79 """ 

80 

81 with self._db as db: 

82 db.purge() 

83 

84 return True 

85 

86 def get_jobs(self, status: JobStatus = None, limit=None, offset=None 

87 ) -> dict: 

88 """ 

89 Get jobs 

90 

91 :param status: job status (accepted, running, successful, 

92 failed, results) (default is all) 

93 :param limit: number of jobs to return 

94 :param offset: pagination offset 

95 

96 :returns: dict of list of jobs (identifier, status, process identifier) 

97 and numberMatched 

98 """ 

99 

100 with self._db() as db: 

101 jobs_list = db.all() 

102 

103 number_matched = len(jobs_list) 

104 

105 if offset: 105 ↛ 106line 105 didn't jump to line 106 because the condition on line 105 was never true

106 jobs_list = jobs_list[offset:] 

107 

108 if limit: 108 ↛ 111line 108 didn't jump to line 111 because the condition on line 108 was always true

109 jobs_list = jobs_list[:limit] 

110 

111 return { 

112 'jobs': jobs_list, 

113 'numberMatched': number_matched 

114 } 

115 

116 def add_job(self, job_metadata: dict) -> str: 

117 """ 

118 Add a job 

119 

120 :param job_metadata: `dict` of job metadata 

121 

122 :returns: identifier of added job 

123 """ 

124 

125 with self._db() as db: 

126 doc_id = db.insert(job_metadata) 

127 

128 return doc_id # noqa 

129 

130 def update_job(self, job_id: str, update_dict: dict) -> bool: 

131 """ 

132 Updates a job 

133 

134 :param job_id: job identifier 

135 :param update_dict: `dict` of property updates 

136 

137 :returns: `bool` of status result 

138 """ 

139 

140 with self._db() as db: 

141 db.update(update_dict, tinydb.where('identifier') == job_id) 

142 

143 return True 

144 

145 def delete_job(self, job_id: str) -> bool: 

146 """ 

147 Deletes a job 

148 

149 :param job_id: job identifier 

150 

151 :raises JobNotFoundError: if the job_id does not correspond to a 

152 known job 

153 :return `bool` of status result 

154 """ 

155 # delete result file if present 

156 job_result = self.get_job(job_id) 

157 location = job_result.get('location') 

158 if location and self.output_dir is not None: 158 ↛ 159line 158 didn't jump to line 159 because the condition on line 158 was never true

159 Path(location).unlink() 

160 

161 with self._db() as db: 

162 removed = bool(db.remove(tinydb.where('identifier') == job_id)) 

163 

164 return removed 

165 

166 def get_job(self, job_id: str) -> dict: 

167 """ 

168 Get a single job 

169 

170 :param job_id: job identifier 

171 

172 :raises JobNotFoundError: if the job_id does not correspond to a 

173 known job 

174 :returns: `dict` # `pygeoapi.process.manager.Job` 

175 """ 

176 

177 query = tinydb.Query() 

178 with self._db() as db: 

179 found = db.search(query.identifier == job_id) 

180 if found is not None: 180 ↛ 186line 180 didn't jump to line 186 because the condition on line 180 was always true

181 try: 

182 return found[0] 

183 except IndexError: 

184 raise JobNotFoundError() 

185 else: 

186 raise JobNotFoundError() 

187 

188 def get_job_result(self, job_id: str) -> Tuple[str, Any]: 

189 """ 

190 Get a job's status, and actual output of executing the process 

191 

192 :param job_id: job identifier 

193 

194 :raises JobNotFoundError: if the job_id does not correspond to a 

195 known job 

196 :raises JobResultNotFoundError: if the job-related result cannot 

197 be returned 

198 :returns: `tuple` of mimetype and raw output 

199 """ 

200 

201 job_result = self.get_job(job_id) 

202 location = job_result.get('location') 

203 mimetype = job_result.get('mimetype') 

204 job_status = JobStatus[job_result['status']] 

205 

206 if not job_status == JobStatus.successful: 

207 # Job is incomplete 

208 return (None,) 

209 if not location: 

210 LOGGER.warning(f'job {job_id!r} - unknown result location') 

211 raise JobResultNotFoundError() 

212 else: 

213 try: 

214 location = Path(location) 

215 if mimetype in (None, FORMAT_TYPES[F_JSON], 

216 FORMAT_TYPES[F_JSONLD]): 

217 with location.open('r', encoding='utf-8') as filehandler: 

218 result = json.load(filehandler) 

219 else: 

220 with location.open('rb') as filehandler: 

221 result = filehandler.read() 

222 except (TypeError, FileNotFoundError, json.JSONDecodeError): 

223 raise JobResultNotFoundError() 

224 else: 

225 return mimetype, result 

226 

227 def __repr__(self): 

228 return f'<TinyDBManager> {self.name}'