Coverage for pygeoapi/plugin.py: 71%

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"""Plugin loader""" 

30 

31import importlib 

32import logging 

33from typing import Any 

34 

35LOGGER = logging.getLogger(__name__) 

36 

37#: Loads provider plugins to be used by pygeoapi,\ 

38#: formatters and processes available 

39PLUGINS = { 

40 'provider': { 

41 'AzureBlobStorage': 'pygeoapi.provider.azure_.AzureBlobStorageProvider', # noqa 

42 'CSV': 'pygeoapi.provider.csv_.CSVProvider', 

43 'CSWFacade': 'pygeoapi.provider.csw_facade.CSWFacadeProvider', 

44 'Elasticsearch': 'pygeoapi.provider.elasticsearch_.ElasticsearchProvider', # noqa 

45 'ElasticsearchCatalogue': 'pygeoapi.provider.elasticsearch_.ElasticsearchCatalogueProvider', # noqa 

46 'ERDDAPTabledap': 'pygeoapi.provider.erddap.TabledapProvider', 

47 'ESRI': 'pygeoapi.provider.esri.ESRIServiceProvider', 

48 'FileSystem': 'pygeoapi.provider.filesystem.FileSystemProvider', 

49 'GeoJSON': 'pygeoapi.provider.geojson.GeoJSONProvider', 

50 'Hateoas': 'pygeoapi.provider.hateoas.HateoasProvider', 

51 'MapScript': 'pygeoapi.provider.mapscript_.MapScriptProvider', 

52 'MongoDB': 'pygeoapi.provider.mongo.MongoProvider', 

53 'MVT-tippecanoe': 'pygeoapi.provider.mvt_tippecanoe.MVTTippecanoeProvider', # noqa: E501 

54 'MVT-elastic': 'pygeoapi.provider.mvt_elastic.MVTElasticProvider', 

55 'MVT-proxy': 'pygeoapi.provider.mvt_proxy.MVTProxyProvider', 

56 'MySQL': 'pygeoapi.provider.sql.MySQLProvider', 

57 'MVT-postgresql': 'pygeoapi.provider.mvt_postgresql.MVTPostgreSQLProvider', # noqa: E501 

58 'OracleDB': 'pygeoapi.provider.oracle.OracleProvider', 

59 'OGR': 'pygeoapi.provider.ogr.OGRProvider', 

60 'OpenSearch': 'pygeoapi.provider.opensearch_.OpenSearchProvider', 

61 'Parquet': 'pygeoapi.provider.parquet.ParquetProvider', 

62 'PostgreSQL': 'pygeoapi.provider.sql.PostgreSQLProvider', 

63 'rasterio': 'pygeoapi.provider.rasterio_.RasterioProvider', 

64 'SensorThings': 'pygeoapi.provider.sensorthings.SensorThingsProvider', 

65 'SensorThingsEDR': 'pygeoapi.provider.sensorthings_edr.SensorThingsEDRProvider', # noqa: E501 

66 'SQLiteGPKG': 'pygeoapi.provider.sqlite.SQLiteGPKGProvider', 

67 'Socrata': 'pygeoapi.provider.socrata.SODAServiceProvider', 

68 'TinyDB': 'pygeoapi.provider.tinydb_.TinyDBProvider', 

69 'TinyDBCatalogue': 'pygeoapi.provider.tinydb_.TinyDBCatalogueProvider', 

70 'WMSFacade': 'pygeoapi.provider.wms_facade.WMSFacadeProvider', 

71 'WMTSFacade': 'pygeoapi.provider.wmts_facade.WMTSFacadeProvider', 

72 'xarray': 'pygeoapi.provider.xarray_.XarrayProvider', 

73 'xarray-edr': 'pygeoapi.provider.xarray_edr.XarrayEDRProvider' 

74 }, 

75 'formatter': { 

76 'CSV': 'pygeoapi.formatter.csv_.CSVFormatter' 

77 }, 

78 'process': { 

79 'HelloWorld': 'pygeoapi.process.hello_world.HelloWorldProcessor', 

80 'ShapelyFunctions': 'pygeoapi.process.shapely_functions.ShapelyFunctionsProcessor', # noqa: E501 

81 'Echo': 'pygeoapi.process.echo.EchoProcessor' 

82 }, 

83 'process_manager': { 

84 'Dummy': 'pygeoapi.process.manager.dummy.DummyManager', 

85 'MongoDB': 'pygeoapi.process.manager.mongodb_.MongoDBManager', 

86 'TinyDB': 'pygeoapi.process.manager.tinydb_.TinyDBManager', 

87 'PostgreSQL': 'pygeoapi.process.manager.postgresql.PostgreSQLManager' 

88 }, 

89 'pubsub': { 

90 'HTTP': 'pygeoapi.pubsub.http.HTTPPubSubClient', 

91 'Kafka': 'pygeoapi.pubsub.kafka.KafkaPubSubClient', 

92 'MQTT': 'pygeoapi.pubsub.mqtt.MQTTPubSubClient' 

93 } 

94} 

95 

96 

97def load_plugin(plugin_type: str, plugin_def: dict) -> Any: 

98 """ 

99 loads plugin by name 

100 

101 :param plugin_type: type of plugin (provider, formatter) 

102 :param plugin_def: plugin definition 

103 

104 :returns: plugin object 

105 """ 

106 

107 name = plugin_def['name'] 

108 

109 if plugin_type not in PLUGINS.keys(): 109 ↛ 110line 109 didn't jump to line 110 because the condition on line 109 was never true

110 msg = f'Plugin type {plugin_type} not found' 

111 LOGGER.exception(msg) 

112 raise InvalidPluginError(msg) 

113 

114 plugin_list = PLUGINS[plugin_type] 

115 

116 LOGGER.debug(f'Plugins: {plugin_list}') 

117 

118 if '.' not in name and name not in plugin_list.keys(): 118 ↛ 119line 118 didn't jump to line 119 because the condition on line 118 was never true

119 msg = f'Plugin {name} not found' 

120 LOGGER.exception(msg) 

121 raise InvalidPluginError(msg) 

122 

123 if '.' in name: # dotted path 123 ↛ 124line 123 didn't jump to line 124 because the condition on line 123 was never true

124 packagename, classname = name.rsplit('.', 1) 

125 else: # core formatter 

126 packagename, classname = plugin_list[name].rsplit('.', 1) 

127 

128 LOGGER.debug(f'package name: {packagename}') 

129 LOGGER.debug(f'class name: {classname}') 

130 

131 module = importlib.import_module(packagename) 

132 class_ = getattr(module, classname) 

133 plugin = class_(plugin_def) 

134 

135 return plugin 

136 

137 

138class InvalidPluginError(Exception): 

139 """Invalid plugin""" 

140 pass