Coverage for pygeoapi/provider/sqlite.py: 76%

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

2# 

3# Authors: Jorge Samuel Mendes de Jesus <jorge.dejesus@protonmail.net> 

4# Tom Kralidis <tomkralidis@gmail.com> 

5# Francesco Bartoli <xbartolone@gmail.com> 

6# 

7# Copyright (c) 2018 Jorge Samuel Mendes de Jesus 

8# Copyright (c) 2023 Tom Kralidis 

9# Copyright (c) 2025 Francesco Bartoli 

10# 

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

12# obtaining a copy of this software and associated documentation 

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

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

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

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

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

18# conditions: 

19# 

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

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

22# 

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

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

25# OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND 

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

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

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

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

30# OTHER DEALINGS IN THE SOFTWARE. 

31# 

32# ================================================================= 

33 

34import sqlite3 

35import logging 

36import os 

37import json 

38 

39from pygeoapi.crs import crs_transform 

40from pygeoapi.plugin import InvalidPluginError 

41from pygeoapi.provider.base import (BaseProvider, ProviderConnectionError, 

42 ProviderItemNotFoundError) 

43 

44LOGGER = logging.getLogger(__name__) 

45 

46 

47SPATIALITE_EXTENSION = os.getenv('SPATIALITE_LIBRARY_PATH', 

48 'mod_spatialite.so') 

49 

50 

51class SQLiteGPKGProvider(BaseProvider): 

52 """Generic provider for SQLITE and GPKG using sqlite3 module. 

53 This module requires install of libsqlite3-mod-spatialite 

54 TODO: DELETE, UPDATE, CREATE 

55 """ 

56 

57 def __init__(self, provider_def): 

58 """ 

59 SQLiteGPKGProvider Class constructor 

60 

61 :param provider_def: provider definitions from yml pygeoapi-config. 

62 data,id_field, name set in parent class 

63 

64 :returns: pygeoapi.provider.base.SQLiteProvider 

65 """ 

66 super().__init__(provider_def) 

67 

68 self.table = provider_def['table'] 

69 self.application_id = None 

70 self.geom_col = None 

71 

72 LOGGER.debug('Setting SQLite properties:') 

73 LOGGER.debug(f'Data source: {self.data}') 

74 LOGGER.debug(f'Name: {self.name}') 

75 LOGGER.debug(f'ID_field: {self.id_field}') 

76 LOGGER.debug(f'Table: {self.table}') 

77 

78 self.cursor = self.__load() 

79 

80 LOGGER.debug('Got cursor from DB') 

81 LOGGER.debug('Get available fields/properties') 

82 

83 self.get_fields() 

84 

85 def get_fields(self): 

86 """ 

87 Get fields from sqlite table (columns are field) 

88 

89 :returns: dict of fields 

90 """ 

91 

92 if not self._fields: 92 ↛ 106line 92 didn't jump to line 106 because the condition on line 92 was always true

93 results = self.cursor.execute( 

94 f'PRAGMA table_info({self.table})').fetchall() 

95 for item in results: 

96 json_type = None 

97 

98 if item['type'] in ['INTEGER', 'REAL']: 

99 json_type = 'number' 

100 elif item['type'].startswith('TEXT') or item['type'] == 'BLOB': 100 ↛ 101line 100 didn't jump to line 101 because the condition on line 100 was never true

101 json_type = 'string' 

102 

103 if json_type is not None: 

104 self._fields[item['name']] = {'type': json_type} 

105 

106 return self._fields 

107 

108 def __get_where_clauses(self, properties=[], bbox=[]): 

109 """ 

110 Generarates WHERE conditions to be implemented in query. 

111 Private method mainly associated with query method. 

112 

113 Method returns part of the SQL query, plus tuple to be used 

114 in the sqlite query method 

115 

116 :param properties: list of tuples (name, value) 

117 :param bbox: bounding box [minx,miny,maxx,maxy] 

118 

119 :returns: str, tuple 

120 """ 

121 

122 where_values = tuple() 

123 where_clause = " WHERE " if (properties or bbox) else "" 

124 if not where_clause: 

125 return where_clause, where_values 

126 

127 if properties: 

128 where_clause += " AND ".join( 

129 [f"{k}=?" for k, v in properties]) 

130 where_values += where_values + tuple((v for k, v in properties)) 

131 

132 if bbox: 132 ↛ 138line 132 didn't jump to line 138 because the condition on line 132 was always true

133 if properties: 

134 where_clause += " AND " 

135 where_clause += f" Intersects({self.geom_col}, BuildMbr(?,?,?,?)) " 

136 where_values += tuple(bbox) 

137 # WHERE continent=? <class 'tuple'>: ('Europe',) 

138 return where_clause, where_values 

139 

140 def __response_feature(self, row_data, skip_geometry=False): 

141 """ 

142 Assembles GeoJSON output from DB query 

143 

144 :param row_data: DB row result 

145 :param skip_geometry: whether to skip geometry (default False) 

146 

147 :returns: `dict` of GeoJSON Feature 

148 """ 

149 

150 if row_data: 

151 rd = dict(row_data) # sqlite3.Row is doesnt support pop 

152 feature = { 

153 'type': 'Feature', 

154 'geometry': None 

155 } 

156 

157 try: 

158 if not skip_geometry: 

159 feature['geometry'] = json.loads( 

160 rd.pop(f'AsGeoJSON({self.geom_col})') 

161 ) 

162 except TypeError: 

163 LOGGER.warning('Missing geometry') 

164 

165 feature['properties'] = rd 

166 feature['id'] = feature['properties'][self.id_field] 

167 

168 return feature 

169 else: 

170 return None 

171 

172 def __response_feature_hits(self, hits): 

173 """Assembles GeoJSON/Feature number 

174 

175 :returns: GeoJSON FeaturesCollection 

176 """ 

177 

178 feature_collection = {"features": [], 

179 "type": "FeatureCollection"} 

180 feature_collection['numberMatched'] = hits 

181 

182 return feature_collection 

183 

184 def __load(self): 

185 """ 

186 Private method for loading spatiallite, 

187 get the table structure and dump geometry 

188 

189 :returns: sqlite3.Cursor 

190 """ 

191 

192 if (os.path.exists(self.data)): 192 ↛ 195line 192 didn't jump to line 195 because the condition on line 192 was always true

193 conn = sqlite3.connect(self.data) 

194 else: 

195 LOGGER.error('Path to sqlite does not exist') 

196 raise InvalidPluginError() 

197 

198 try: 

199 conn.enable_load_extension(True) 

200 except AttributeError as err: 

201 LOGGER.error(f'Extension loading not enabled: {err}') 

202 raise ProviderConnectionError() 

203 

204 conn.row_factory = sqlite3.Row 

205 conn.enable_load_extension(True) 

206 # conn.set_trace_callback(LOGGER.debug) 

207 cursor = conn.cursor() 

208 try: 

209 cursor.execute(f"SELECT load_extension('{SPATIALITE_EXTENSION}')") 

210 except sqlite3.OperationalError as err: 

211 LOGGER.error(f'Extension loading error: {err}') 

212 raise ProviderConnectionError() 

213 result = cursor.fetchall() 

214 

215 # Checking for geopackage 

216 cursor.execute("PRAGMA application_id") 

217 result = cursor.fetchone() 

218 

219 self.application_id = result["application_id"] 

220 if self.application_id == 1196444487: 220 ↛ 221line 220 didn't jump to line 221 because the condition on line 220 was never true

221 LOGGER.info("Detected GPKG 1.2 and greater") 

222 elif self.application_id == 1196437808: 222 ↛ 223line 222 didn't jump to line 223 because the condition on line 222 was never true

223 LOGGER.info("Detected GPKG 1.0 or 1.1") 

224 else: 

225 LOGGER.info("No GPKG detected assuming spatial sqlite3") 

226 self.application_id = 0 

227 

228 if self.application_id: 228 ↛ 229line 228 didn't jump to line 229 because the condition on line 228 was never true

229 geometry_columns_table = 'gpkg_geometry_columns' 

230 geometry_columns_table_name = 'table_name' 

231 geometry_columns_column_name = 'column_name' 

232 cursor.execute("SELECT AutoGPKGStart()") 

233 result = cursor.fetchall() 

234 if result[0][0] >= 1: 

235 LOGGER.info("Loaded Geopackage support") 

236 else: 

237 LOGGER.info("SELECT AutoGPKGStart() returned 0." + 

238 "Detected GPKG but couldn't load support") 

239 raise InvalidPluginError() 

240 else: 

241 geometry_columns_table = 'geometry_columns' 

242 geometry_columns_column_name = 'f_geometry_column' 

243 geometry_columns_table_name = 'f_table_name' 

244 

245 try: 

246 cursor.execute(f'PRAGMA table_info({self.table})') 

247 result = cursor.fetchall() 

248 except sqlite3.OperationalError: 

249 LOGGER.error(f'Could not find table: {self.table}') 

250 raise ProviderConnectionError() 

251 

252 LOGGER.debug('Determining name of geometry column') 

253 cursor.execute(f"SELECT {geometry_columns_column_name} FROM {geometry_columns_table} WHERE {geometry_columns_table_name} = '{self.table}'") # noqa 

254 geometry_column = cursor.fetchall() 

255 

256 if geometry_column: 256 ↛ 260line 256 didn't jump to line 260 because the condition on line 256 was always true

257 LOGGER.debug("Found geometry column") 

258 self.geom_col = geometry_column[0][0] 

259 else: 

260 msg = 'No geometry column found' 

261 LOGGER.error(msg) 

262 raise ProviderConnectionError(msg) 

263 

264 try: 

265 assert len(result), 'Table not found' 

266 assert len([item for item in result 

267 if self.id_field in item]), 'id_field not present' 

268 except AssertionError: 

269 raise InvalidPluginError() 

270 

271 self.columns = [item[1] for item in result if item[1] 

272 not in [self.geom_col, self.geom_col.upper()]] 

273 self.columns = ','.join(self.columns)+f',AsGeoJSON({self.geom_col})' 

274 

275 if self.application_id: 275 ↛ 276line 275 didn't jump to line 276 because the condition on line 275 was never true

276 self.table = f"vgpkg_{self.table}" 

277 

278 return cursor 

279 

280 @crs_transform 

281 def query(self, offset=0, limit=10, resulttype='results', 

282 bbox=[], datetime_=None, properties=[], sortby=[], 

283 select_properties=[], skip_geometry=False, q=None, **kwargs): 

284 """ 

285 Query SQLite/GPKG for all the content. 

286 e,g: http://localhost:5000/collections/countries/items? 

287 limit=5&offset=2&resulttype=results&continent=Europe&admin=Albania&bbox=29.3373,-3.4099,29.3761,-3.3924 

288 http://localhost:5000/collections/countries/items?continent=Africa&bbox=29.3373,-3.4099,29.3761,-3.3924 

289 

290 :param offset: starting record to return (default 0) 

291 :param limit: number of records to return (default 10) 

292 :param resulttype: return results or hit limit (default results) 

293 :param bbox: bounding box [minx,miny,maxx,maxy] 

294 :param datetime_: temporal (datestamp or extent) 

295 :param properties: list of tuples (name, value) 

296 :param sortby: list of dicts (property, order) 

297 :param select_properties: list of property names 

298 :param skip_geometry: bool of whether to skip geometry (default False) 

299 :param q: full-text search term(s) 

300 

301 :returns: GeoJSON FeaturesCollection 

302 """ 

303 LOGGER.debug('Querying SQLite/GPKG') 

304 

305 where_clause, where_values = self.__get_where_clauses( 

306 properties=properties, bbox=bbox) 

307 

308 if resulttype == 'hits': 308 ↛ 310line 308 didn't jump to line 310 because the condition on line 308 was never true

309 

310 sql_query = f"SELECT COUNT(*) as hits FROM {self.table} {where_clause} " # noqa 

311 

312 res = self.cursor.execute(sql_query, where_values) 

313 

314 hits = res.fetchone()["hits"] 

315 return self.__response_feature_hits(hits) 

316 

317 sql_query = f"SELECT DISTINCT {self.columns} from \ 

318 {self.table} {where_clause} limit ? offset ?" 

319 

320 end_index = offset + limit 

321 

322 LOGGER.debug(f'SQL Query: {sql_query}') 

323 LOGGER.debug(f'Start Index: {offset}') 

324 LOGGER.debug(f'End Index: {end_index}') 

325 

326 row_data = self.cursor.execute( 

327 sql_query, where_values + (limit, offset)) 

328 

329 feature_collection = { 

330 'type': 'FeatureCollection', 

331 'features': [] 

332 } 

333 

334 for rd in row_data: 

335 feature_collection['features'].append( 

336 self.__response_feature(rd, skip_geometry=skip_geometry)) 

337 

338 return feature_collection 

339 

340 @crs_transform 

341 def get(self, identifier, **kwargs): 

342 """ 

343 Query the provider for a specific 

344 feature id e.g: /collections/countries/items/1 

345 

346 :param identifier: feature id 

347 

348 :returns: dict of single GeoJSON feature 

349 """ 

350 

351 LOGGER.debug('Get item from SQLite/GPKG') 

352 

353 sql_query = f'SELECT {self.columns} FROM \ 

354 {self.table} WHERE {self.id_field}==?;' 

355 

356 LOGGER.debug(f'SQL Query: {sql_query}') 

357 LOGGER.debug(f'Identifier: {identifier}') 

358 

359 row_data = self.cursor.execute(sql_query, (identifier, )).fetchone() 

360 

361 feature = self.__response_feature(row_data) 

362 if feature: 

363 return feature 

364 else: 

365 err = f'item {identifier} not found' 

366 LOGGER.error(err) 

367 raise ProviderItemNotFoundError(err) 

368 

369 def __repr__(self): 

370 return f'<SQLiteGPKGProvider> {self.data}, {self.table}'