diff --git a/geoservercloud/geoservercloud.py b/geoservercloud/geoservercloud.py index b4317da..9223a63 100644 --- a/geoservercloud/geoservercloud.py +++ b/geoservercloud/geoservercloud.py @@ -12,12 +12,14 @@ from geoservercloud.models.coveragestore import CoverageStore from geoservercloud.models.datastore import DataStore from geoservercloud.models.featuretype import FeatureType +from geoservercloud.models.gwclayer import GridSubset, GwcLayer, ParameterFilter from geoservercloud.models.layer import Layer from geoservercloud.models.layergroup import LayerGroup from geoservercloud.models.style import Style from geoservercloud.models.wmslayer import WmsLayer from geoservercloud.models.wmssettings import WmsSettings from geoservercloud.models.wmsstore import WmsStore +from geoservercloud.models.wmtsstore import WmtsStore from geoservercloud.models.workspace import Workspace from geoservercloud.services import OwsService, RestService @@ -71,6 +73,12 @@ def get_version(self) -> tuple[dict[str, dict[str, list]] | str, int]: return self.rest_service.get_version() def create_wms(self, workspace: str | None = None) -> None: + """ + Initialize a WMS OWSLib client scoped to the given workspace + + :param workspace: Name of the workspace, or None to use the default workspace + :type workspace: str, optional + """ if workspace: self.wms = self.ows_service.create_wms(workspace) elif self.default_workspace: @@ -79,6 +87,12 @@ def create_wms(self, workspace: str | None = None) -> None: self.wms = self.ows_service.create_wms() def create_wmts(self, workspace_name: str | None = None) -> None: + """ + Initialize a WMTS OWSLib client scoped to the given workspace + + :param workspace_name: Name of the workspace, or None to use the default workspace + :type workspace_name: str, optional + """ if workspace_name: self.wmts = self.ows_service.create_wmts(workspace_name) elif self.default_workspace: @@ -98,6 +112,9 @@ def cleanup(self): def get_workspaces(self) -> tuple[list[dict[str, str]] | str, int]: """ Get all GeoServer workspaces + + :return: Tuple of (workspaces, status_code) + :rtype: tuple """ workspaces, status_code = self.rest_service.get_workspaces() if isinstance(workspaces, str): @@ -107,6 +124,11 @@ def get_workspaces(self) -> tuple[list[dict[str, str]] | str, int]: def get_workspace(self, workspace_name: str) -> tuple[dict[str, str] | str, int]: """ Get a workspace by name + + :param workspace_name: Name of the workspace + :type workspace_name: str + :return: Tuple of (workspace, status_code) + :rtype: tuple """ workspace, status_code = self.rest_service.get_workspace(workspace_name) if isinstance(workspace, str): @@ -121,7 +143,16 @@ def create_workspace( ) -> tuple[str, int]: """ Create a workspace in GeoServer, if it does not already exist. - It if exists, update it + If it exists, update it. + + :param workspace_name: Name of the workspace + :type workspace_name: str + :param isolated: Whether the workspace should be isolated (default: False) + :type isolated: bool, optional + :param set_default_workspace: Whether to set as the default workspace (default: False) + :type set_default_workspace: bool, optional + :return: Tuple of (content, status_code) + :rtype: tuple """ workspace = Workspace(workspace_name, isolated) content, status_code = self.rest_service.create_workspace(workspace) @@ -132,6 +163,11 @@ def create_workspace( def delete_workspace(self, workspace_name: str) -> tuple[str, int]: """ Delete a GeoServer workspace (recursively) + + :param workspace_name: Name of the workspace + :type workspace_name: str + :return: Tuple of (content, status_code) + :rtype: tuple """ content, status_code = self.rest_service.delete_workspace( Workspace(workspace_name) @@ -150,6 +186,15 @@ def recreate_workspace( ) -> tuple[str, int]: """ Create a workspace in GeoServer, and first delete it if it already exists. + + :param workspace_name: Name of the workspace + :type workspace_name: str + :param isolated: Whether the workspace should be isolated (default: False) + :type isolated: bool, optional + :param set_default_workspace: Whether to set as the default workspace (default: False) + :type set_default_workspace: bool, optional + :return: Tuple of (content, status_code) + :rtype: tuple """ self.delete_workspace(workspace_name) return self.create_workspace( @@ -163,6 +208,11 @@ def get_workspace_wms_settings( ) -> tuple[dict[str, Any] | str, int]: """ Get the WMS settings for a given workspace + + :param workspace_name: Name of the workspace + :type workspace_name: str + :return: Tuple of (wms_settings, status_code) + :rtype: tuple """ wms_settings, status_code = self.rest_service.get_workspace_wms_settings( workspace_name @@ -207,6 +257,55 @@ def publish_workspace( ) -> tuple[str, int]: """ Publish the WMS service for a given workspace + + :param workspace_name: Name of the workspace + :type workspace_name: str + :param versions: WMS versions to enable (default: ["1.1.1", "1.3.0"]) + :type versions: list of str, optional + :param cite_compliant: Whether the service should be CITE compliant (default: False) + :type cite_compliant: bool, optional + :param schema_base_url: Base URL for OGC schemas (default: "http://schemas.opengis.net") + :type schema_base_url: str, optional + :param verbose: Enable verbose XML output (default: False) + :type verbose: bool, optional + :param bbox_for_each_crs: Whether to compute the bounding box for each supported CRS (default: False) + :type bbox_for_each_crs: bool, optional + :param watermark: Watermark configuration (default: disabled) + :type watermark: dict, optional + :param interpolation: Default interpolation method (default: "Nearest") + :type interpolation: str, optional + :param get_feature_info_mime_type_checking_enabled: Restrict allowed MIME types for GetFeatureInfo (default: False) + :type get_feature_info_mime_type_checking_enabled: bool, optional + :param get_map_mime_type_checking_enabled: Restrict allowed MIME types for GetMap (default: False) + :type get_map_mime_type_checking_enabled: bool, optional + :param dynamic_styling_disabled: Disable the SLD_BODY parameter in requests (default: False) + :type dynamic_styling_disabled: bool, optional + :param features_reprojection_disabled: Disable on-the-fly feature reprojection (default: False) + :type features_reprojection_disabled: bool, optional + :param max_buffer: Maximum buffer size in pixels for rendering, 0 for unlimited (default: 0) + :type max_buffer: int, optional + :param max_request_memory: Maximum memory in KB usable per request, 0 for unlimited (default: 0) + :type max_request_memory: int, optional + :param max_rendering_time: Maximum rendering time in seconds, 0 for unlimited (default: 0) + :type max_rendering_time: int, optional + :param max_rendering_errors: Maximum number of rendering errors tolerated, 0 for unlimited (default: 0) + :type max_rendering_errors: int, optional + :param max_requested_dimension_values: Maximum number of dimension values that can be requested (default: 100) + :type max_requested_dimension_values: int, optional + :param cache_configuration: GetMap caching configuration (default: disabled) + :type cache_configuration: dict, optional + :param remote_style_max_request_time: Maximum time in ms allowed to fetch a remote style (default: 60000) + :type remote_style_max_request_time: int, optional + :param remote_style_timeout: Timeout in ms for fetching a remote style (default: 30000) + :type remote_style_timeout: int, optional + :param default_group_style_enabled: Whether a default style is generated for layer groups (default: True) + :type default_group_style_enabled: bool, optional + :param transform_feature_info_disabled: Disable XSLT transformation of GetFeatureInfo output (default: False) + :type transform_feature_info_disabled: bool, optional + :param auto_escape_template_values: Automatically escape template values in GetFeatureInfo output (default: False) + :type auto_escape_template_values: bool, optional + :return: Tuple of (content, status_code) + :rtype: tuple """ wms_settings = WmsSettings( workspace_name=workspace_name, @@ -248,6 +347,13 @@ def set_default_locale_for_service( ) -> tuple[str, int]: """ Set a default language for localized WMS requests + + :param workspace_name: Name of the workspace + :type workspace_name: str + :param locale: Locale code to set as default (e.g. "en"), or None to unset + :type locale: str, optional + :return: Tuple of (content, status_code) + :rtype: tuple """ wms_settings = WmsSettings(default_locale=locale) return self.rest_service.put_workspace_wms_settings( @@ -257,6 +363,11 @@ def set_default_locale_for_service( def unset_default_locale_for_service(self, workspace_name) -> tuple[str, int]: """ Remove the default language for localized WMS requests + + :param workspace_name: Name of the workspace + :type workspace_name: str + :return: Tuple of (content, status_code) + :rtype: tuple """ return self.set_default_locale_for_service(workspace_name, None) @@ -265,6 +376,11 @@ def get_datastores( ) -> tuple[list[dict[str, str]] | str, int]: """ Get all datastores for a given workspace + + :param workspace_name: Name of the workspace + :type workspace_name: str + :return: Tuple of (datastores, status_code) + :rtype: tuple """ datastores, status_code = self.rest_service.get_datastores(workspace_name) if isinstance(datastores, str): @@ -276,6 +392,13 @@ def get_datastore( ) -> tuple[dict[str, Any] | str, int]: """ Get a datastore by workspace and name + + :param workspace_name: Name of the workspace + :type workspace_name: str + :param datastore_name: Name of the datastore + :type datastore_name: str + :return: Tuple of (datastore, status_code) + :rtype: tuple """ datastore, status_code = self.rest_service.get_datastore( workspace_name, datastore_name @@ -289,6 +412,13 @@ def get_pg_datastore( ) -> tuple[dict[str, Any] | str, int]: """ Get a datastore by workspace and name + + :param workspace_name: Name of the workspace + :type workspace_name: str + :param datastore_name: Name of the datastore + :type datastore_name: str + :return: Tuple of (datastore, status_code) + :rtype: tuple """ return self.get_datastore(workspace_name, datastore_name) @@ -373,7 +503,31 @@ def create_pg_datastore( set_default_datastore: bool = False, ) -> tuple[str, int]: """ - Create a PostGIS datastore from the DB connection parameters, or update it if it already exist. + Create a PostGIS datastore from the DB connection parameters, or update it if it already exists. + + :param workspace_name: Name of the workspace + :type workspace_name: str + :param datastore_name: Name for the datastore + :type datastore_name: str + :param pg_host: PostgreSQL host + :type pg_host: str + :param pg_port: PostgreSQL port + :type pg_port: int + :param pg_db: PostgreSQL database name + :type pg_db: str + :param pg_user: PostgreSQL user + :type pg_user: str + :param pg_password: PostgreSQL password + :type pg_password: str + :param pg_schema: PostgreSQL schema (default: "public") + :type pg_schema: str, optional + :param description: Optional description + :type description: str, optional + :param set_default_datastore: Whether to set as default datastore (default: False) + :type set_default_datastore: bool, optional + + :return: Tuple of (datastore_name, status_code) + :rtype: tuple """ datastore = DataStore( workspace_name, @@ -411,7 +565,23 @@ def create_jndi_datastore( set_default_datastore: bool = False, ) -> tuple[str, int]: """ - Create a PostGIS datastore from JNDI resource, or update it if it already exist. + Create a PostGIS datastore from a JNDI resource, or update it if it already exists. + + :param workspace_name: Name of the workspace + :type workspace_name: str + :param datastore_name: Name for the datastore + :type datastore_name: str + :param jndi_reference: JNDI resource reference name + :type jndi_reference: str + :param pg_schema: PostgreSQL schema (default: "public") + :type pg_schema: str, optional + :param description: Optional description + :type description: str, optional + :param set_default_datastore: Whether to set as default datastore (default: False) + :type set_default_datastore: bool, optional + + :return: Tuple of (datastore_name, status_code) + :rtype: tuple """ datastore = DataStore( workspace_name, @@ -529,6 +699,13 @@ def delete_datastore( ) -> tuple[str, int]: """ Delete a datastore recursively + + :param workspace_name: Name of the workspace + :type workspace_name: str + :param datastore_name: Name of the datastore + :type datastore_name: str + :return: Tuple of (content, status_code) + :rtype: tuple """ return self.rest_service.delete_datastore(workspace_name, datastore_name) @@ -536,7 +713,14 @@ def get_wms_store( self, workspace_name: str, datastore_name: str ) -> tuple[dict[str, Any] | str, int]: """ - Get a WMS by workspace and name + Get a WMS store by workspace and name + + :param workspace_name: Name of the workspace + :type workspace_name: str + :param datastore_name: Name of the WMS store + :type datastore_name: str + :return: Tuple of (wms_store, status_code) + :rtype: tuple """ wms_store, status_code = self.rest_service.get_wms_store( workspace_name, datastore_name @@ -552,7 +736,16 @@ def create_wms_store( capabilities_url: str, ) -> tuple[str, int]: """ - Create a cascaded WMS store, or update it if it already exist. + Create a cascaded WMS store, or update it if it already exists. + + :param workspace_name: Name of the workspace + :type workspace_name: str + :param wms_store_name: Name for the WMS store + :type wms_store_name: str + :param capabilities_url: URL of the remote WMS GetCapabilities document + :type capabilities_url: str + :return: Tuple of (content, status_code) + :rtype: tuple """ wms_store = WmsStore( workspace_name, @@ -566,6 +759,13 @@ def delete_wms_store( ) -> tuple[str, int]: """ Delete a WMS store recursively + + :param workspace_name: Name of the workspace + :type workspace_name: str + :param wms_store_name: Name of the WMS store + :type wms_store_name: str + :return: Tuple of (content, status_code) + :rtype: tuple """ return self.rest_service.delete_wms_store(workspace_name, wms_store_name) @@ -574,6 +774,15 @@ def get_wms_layer( ) -> tuple[dict[str, Any] | str, int]: """ Get a WMS layer by workspace, store and name + + :param workspace_name: Name of the workspace + :type workspace_name: str + :param wms_store_name: Name of the WMS store + :type wms_store_name: str + :param wms_layer_name: Name of the WMS layer + :type wms_layer_name: str + :return: Tuple of (wms_layer, status_code) + :rtype: tuple """ wms_layer, status_code = self.rest_service.get_wms_layer( workspace_name, wms_store_name, wms_layer_name @@ -590,8 +799,19 @@ def create_wms_layer( published_layer_name: str | None = None, ) -> tuple[str, int]: """ - Publish a remote WMS layer + Publish a remote WMS layer. If it already exists, delete and recreate it (update is not supported by GeoServer) + + :param workspace_name: Name of the workspace + :type workspace_name: str + :param wms_store_name: Name of the WMS store + :type wms_store_name: str + :param native_layer_name: Name of the layer on the remote WMS server + :type native_layer_name: str + :param published_layer_name: Name for the published layer (default: same as native_layer_name) + :type published_layer_name: str, optional + :return: Tuple of (content, status_code) + :rtype: tuple """ published_layer_name = published_layer_name or native_layer_name wms_layer = WmsLayer( @@ -609,6 +829,15 @@ def delete_wms_layer( ) -> tuple[str, int]: """ Delete a WMS layer + + :param workspace_name: Name of the workspace + :type workspace_name: str + :param wms_store_name: Name of the WMS store + :type wms_store_name: str + :param wms_layer_name: Name of the WMS layer + :type wms_layer_name: str + :return: Tuple of (content, status_code) + :rtype: tuple """ return self.rest_service.delete_wms_layer( workspace_name, wms_store_name, wms_layer_name @@ -619,17 +848,66 @@ def create_wmts_store( workspace_name: str, name: str, capabilities: str, + enabled: bool = True, + default: bool | None = None, + disable_on_conn_failure: bool | None = None, + use_connection_pooling: bool | None = True, + max_connections: int | None = None, + read_timeout: int | None = None, + connect_timeout: int | None = None, ) -> tuple[str, int]: """ - Create a cascaded WMTS store, or update it if it already exist. + Create a cascaded WMTS store, or update it if it already exists. + + :param workspace_name: Name of the workspace + :type workspace_name: str + :param name: Name for the WMTS store + :type name: str + :param capabilities: URL of the remote WMTS GetCapabilities document + :type capabilities: str + :param enabled: Whether the store should be enabled (default: True) + :type enabled: bool, optional + :param default: Whether this is the default WMTS store + :type default: bool, optional + :param disable_on_conn_failure: Disable the store on connection failure + :type disable_on_conn_failure: bool, optional + :param use_connection_pooling: Whether to use connection pooling (default: True) + :type use_connection_pooling: bool, optional + :param max_connections: Maximum number of connections + :type max_connections: int, optional + :param read_timeout: Read timeout in seconds + :type read_timeout: int, optional + :param connect_timeout: Connect timeout in seconds + :type connect_timeout: int, optional + :return: Tuple of (content, status_code) + :rtype: tuple """ - return self.rest_service.create_wmts_store(workspace_name, name, capabilities) + wmts_store = WmtsStore( + workspace_name=workspace_name, + name=name, + capabilities_url=capabilities, + enabled=enabled, + default=default, + disable_on_conn_failure=disable_on_conn_failure, + use_connection_pooling=use_connection_pooling, + max_connections=max_connections, + read_timeout=read_timeout, + connect_timeout=connect_timeout, + ) + return self.rest_service.create_wmts_store(workspace_name, wmts_store) def delete_wmts_store( self, workspace_name: str, wmts_store_name: str ) -> tuple[str, int]: """ Delete a WMTS store recursively + + :param workspace_name: Name of the workspace + :type workspace_name: str + :param wmts_store_name: Name of the WMTS store + :type wmts_store_name: str + :return: Tuple of (content, status_code) + :rtype: tuple """ return self.rest_service.delete_wmts_store(workspace_name, wmts_store_name) @@ -638,6 +916,13 @@ def get_feature_types( ) -> tuple[list[dict[str, Any]] | str, int]: """ Get all feature types for a given workspace and datastore + + :param workspace_name: Name of the workspace + :type workspace_name: str + :param datastore_name: Name of the datastore + :type datastore_name: str + :return: Tuple of (feature_types, status_code) + :rtype: tuple """ feature_types, status_code = self.rest_service.get_feature_types( workspace_name, datastore_name @@ -651,6 +936,15 @@ def get_feature_type( ) -> tuple[dict[str, Any] | str, int]: """ Get a feature type by workspace, datastore and name + + :param workspace_name: Name of the workspace + :type workspace_name: str + :param datastore_name: Name of the datastore + :type datastore_name: str + :param feature_type_name: Name of the feature type + :type feature_type_name: str + :return: Tuple of (feature_type, status_code) + :rtype: tuple """ content, code = self.rest_service.get_feature_type( workspace_name, datastore_name, feature_type_name @@ -664,6 +958,13 @@ def get_coverages( ) -> tuple[list[dict[str, str]] | str, int]: """ Get all coverages for a given workspace and coverage store + + :param workspace_name: Name of the workspace + :type workspace_name: str + :param coveragestore_name: Name of the coverage store + :type coveragestore_name: str + :return: Tuple of (coverages, status_code) + :rtype: tuple """ coverages, status_code = self.rest_service.get_coverages( workspace_name, coveragestore_name @@ -677,6 +978,15 @@ def get_coverage( ) -> tuple[dict[str, object] | str, int]: """ Get a single coverage for a given workspace, coverage store, and coverage name + + :param workspace_name: Name of the workspace + :type workspace_name: str + :param coveragestore_name: Name of the coverage store + :type coveragestore_name: str + :param coverage_name: Name of the coverage + :type coverage_name: str + :return: Tuple of (coverage, status_code) + :rtype: tuple """ coverage, status_code = self.rest_service.get_coverage( workspace_name, coveragestore_name, coverage_name @@ -695,6 +1005,19 @@ def create_coverage( ) -> tuple[str, int]: """ Publish a coverage layer from a given coverage store + + :param workspace_name: Name of the workspace + :type workspace_name: str + :param coveragestore_name: Name of the coverage store + :type coveragestore_name: str + :param coverage_name: Name for the coverage + :type coverage_name: str + :param title: Optional title for the coverage (default: same as coverage_name) + :type title: str, optional + :param native_name: Native name of the coverage (default: same as coverage_name) + :type native_name: str, optional + :return: Tuple of (content, status_code) + :rtype: tuple """ coverage = Coverage( workspace_name=workspace_name, @@ -710,6 +1033,13 @@ def get_coverage_store( ) -> tuple[dict[str, Any] | str, int]: """ Get a coverage store by workspace and name + + :param workspace_name: Name of the workspace + :type workspace_name: str + :param coveragestore_name: Name of the coverage store + :type coveragestore_name: str + :return: Tuple of (coverage_store, status_code) + :rtype: tuple """ coverage_store, status_code = self.rest_service.get_coverage_store( workspace_name, coveragestore_name @@ -731,11 +1061,19 @@ def create_coverage_store( Create a coverage store from a store definition. When using a directory path as URL, coverages will be auto-discovered :param workspace_name: Name of the workspace + :type workspace_name: str :param coveragestore_name: Name of the coverage store + :type coveragestore_name: str :param url: Directory path on the server or URL of the granules (raster images) - :param type: Type of the coverage store, e.g. ImageMosaic, GeoTIFF (default: ImageMosaic) + :type url: str + :param type: Type of the coverage store, e.g. ImageMosaic, GeoTIFF (default: "ImageMosaic") + :type type: str, optional :param enabled: Whether the coverage store is enabled (default: True) + :type enabled: bool, optional :param metadata: Optional metadata dictionary (e.g. {"cogSettings": {"rangeReaderSettings": "HTTP"}}) + :type metadata: dict, optional + :return: Tuple of (content, status_code) + :rtype: tuple """ return self.rest_service.create_coverage_store( CoverageStore( @@ -755,6 +1093,15 @@ def create_imagemosaic_store_from_directory( Create an ImageMosaic coverage store from a directory on the server which contains granules (raster images). Granules and coverages will be auto-discovered. Similar to creating a store from the WebUI. Calls /workspaces/{workspace_name}/coveragestores/{coveragestore_name}/external.imagemosaic + + :param workspace_name: Name of the workspace + :type workspace_name: str + :param coveragestore_name: Name of the coverage store + :type coveragestore_name: str + :param directory_path: Directory path on the server containing the granules + :type directory_path: str + :return: Tuple of (content, status_code) + :rtype: tuple """ return self.rest_service.create_imagemosaic_store_from_directory( workspace_name, coveragestore_name, directory_path @@ -767,6 +1114,15 @@ def create_imagemosaic_store_from_properties_zip( Upload an ImageMosaic coverage store configuration as ZIP to create an empty coverage store. The ZIP contains two files: indexer.properties and datastore.properties Calls /workspaces/{workspace_name}/coveragestores/{coveragestore_name}/file.imagemosaic?configure=none + + :param workspace_name: Name of the workspace + :type workspace_name: str + :param coveragestore_name: Name of the coverage store + :type coveragestore_name: str + :param properties_zip: ZIP archive content containing indexer.properties and datastore.properties + :type properties_zip: bytes + :return: Tuple of (content, status_code) + :rtype: tuple """ return self.rest_service.create_imagemosaic_store_from_properties_zip( workspace_name, coveragestore_name, properties_zip @@ -784,9 +1140,15 @@ def publish_granule_to_coverage_store( The granule is an existing file stored either on the server or remotely. :param workspace_name: Name of the workspace + :type workspace_name: str :param coveragestore_name: Name of the coverage store + :type coveragestore_name: str :param method: "external" for a file on the server, "remote" for a remote file + :type method: str :param granule_path: file path (for external granules) or URL (for remote granules) + :type granule_path: str + :return: Tuple of (content, status_code) + :rtype: tuple """ if method not in ["external", "remote"]: raise ValueError( @@ -801,6 +1163,15 @@ def harvest_granules_to_coverage_store( ) -> tuple[str, int]: """ Harvest granules (raster files) from a server directory into an existing ImageMosaic coverage store + + :param workspace_name: Name of the workspace + :type workspace_name: str + :param coveragestore_name: Name of the coverage store + :type coveragestore_name: str + :param directory_path: Directory path on the server containing the granules + :type directory_path: str + :return: Tuple of (content, status_code) + :rtype: tuple """ return self.rest_service.harvest_granules_to_coverage_store( workspace_name, coveragestore_name, directory_path @@ -811,6 +1182,13 @@ def delete_coverage_store( ) -> tuple[str, int]: """ Delete a coverage store recursively + + :param workspace_name: Name of the workspace + :type workspace_name: str + :param coveragestore_name: Name of the coverage store + :type coveragestore_name: str + :return: Tuple of (content, status_code) + :rtype: tuple """ return self.rest_service.delete_coverage_store( workspace_name, coveragestore_name @@ -935,6 +1313,15 @@ def delete_feature_type( ) -> tuple[str, int]: """ Delete a feature type and associated layer + + :param workspace_name: Name of the workspace + :type workspace_name: str + :param datastore_name: Name of the datastore + :type datastore_name: str + :param layer_name: Name of the feature type / layer + :type layer_name: str + :return: Tuple of (content, status_code) + :rtype: tuple """ return self.rest_service.delete_feature_type( workspace_name, datastore_name, layer_name @@ -945,6 +1332,11 @@ def get_layer_groups( ) -> tuple[list[dict[str, str]] | str, int]: """ Get all layer groups for a given workspace + + :param workspace_name: Name of the workspace + :type workspace_name: str + :return: Tuple of (layer_groups, status_code) + :rtype: tuple """ layer_groups, status_code = self.rest_service.get_layer_groups(workspace_name) if isinstance(layer_groups, str): @@ -956,6 +1348,13 @@ def get_layer_group( ) -> tuple[dict[str, Any] | str, int]: """ Get a layer group by name + + :param workspace_name: Name of the workspace + :type workspace_name: str + :param layer_group_name: Name of the layer group + :type layer_group_name: str + :return: Tuple of (layer_group, status_code) + :rtype: tuple """ layer_group, status_code = self.rest_service.get_layer_group( workspace_name, layer_group_name @@ -979,7 +1378,32 @@ def create_layer_group( global_styles: bool = False, ) -> tuple[str, int]: """ - Create a layer group or update it if it already exists. + Create a layer group, or update it if it already exists. + + :param group: Name for the layer group + :type group: str + :param workspace_name: Name of the workspace + :type workspace_name: str, optional + :param layers: List of layer names to include in the group + :type layers: list of str, optional + :param styles: List of style names associated with each layer + :type styles: list of str, optional + :param title: Title for the layer group (can be internationalized as dict) + :type title: str or dict, optional + :param abstract: Abstract for the layer group (can be internationalized as dict) + :type abstract: str or dict, optional + :param epsg: EPSG code used to compute the layer group bounds (default: 4326) + :type epsg: int, optional + :param mode: Layer group mode, e.g. "SINGLE", "NAMED", "CONTAINER", "EO" (default: "SINGLE") + :type mode: str, optional + :param enabled: Whether the layer group is enabled (default: True) + :type enabled: bool, optional + :param advertised: Whether the layer group is advertised (default: True) + :type advertised: bool, optional + :param global_styles: Whether the provided styles are global styles rather than workspace styles (default: False) + :type global_styles: bool, optional + :return: Tuple of (content, status_code) + :rtype: tuple """ workspace_name = workspace_name or self.default_workspace if not workspace_name: @@ -1031,6 +1455,13 @@ def delete_layer_group( ) -> tuple[str, int]: """ Delete a layer group + + :param workspace_name: Name of the workspace + :type workspace_name: str + :param layer_group_name: Name of the layer group + :type layer_group_name: str + :return: Tuple of (content, status_code) + :rtype: tuple """ return self.rest_service.delete_layer_group(workspace_name, layer_group_name) @@ -1046,6 +1477,23 @@ def create_wmts_layer( ) -> tuple[str, int]: """ Publish a remote WMTS layer (first delete it if it already exists) + + :param workspace_name: Name of the workspace + :type workspace_name: str + :param wmts_store: Name of the WMTS store + :type wmts_store: str + :param native_layer: Name of the layer on the remote WMTS server + :type native_layer: str + :param published_layer: Name for the published layer (default: same as native_layer) + :type published_layer: str, optional + :param epsg: EPSG code for the layer SRS (default: 4326) + :type epsg: int, optional + :param international_title: Internationalized title, e.g. {"en": "English Title"} + :type international_title: dict, optional + :param international_abstract: Internationalized abstract, e.g. {"en": "English Abstract"} + :type international_abstract: dict, optional + :return: Tuple of (content, status_code) + :rtype: tuple """ if not published_layer: published_layer = native_layer @@ -1062,14 +1510,93 @@ def create_wmts_layer( def get_gwc_layer( self, workspace_name: str, layer: str ) -> tuple[dict[str, Any] | str, int]: + """ + Get a GeoWebCache layer by workspace and layer name + + :param workspace_name: Name of the workspace + :type workspace_name: str + :param layer: Name of the layer + :type layer: str + :return: Tuple of (content, status_code) + :rtype: tuple + """ return self.rest_service.get_gwc_layer(workspace_name, layer) def publish_gwc_layer( - self, workspace_name: str, layer: str, epsg: int = 4326 + self, + workspace_name: str, + layer: str, + epsg: int = 4326, + id: str | None = None, + enabled: bool = True, + grid_subsets: list[GridSubset] | None = None, + mime_formats: list[str] | None = None, + parameter_filters: list[ParameterFilter] | None = None, + meta_width_height: list[int] | None = None, + gutter: int | None = None, + expire_cache: int | None = None, + expire_clients: int | None = None, + cache_warning_skips: list[Any] | None = None, ) -> tuple[str, int]: - return self.rest_service.publish_gwc_layer(workspace_name, layer, epsg) + """ + Publish a GeoWebCache layer, or update it if it already exists. + + :param workspace_name: Name of the workspace + :type workspace_name: str + :param layer: Name of the layer to cache + :type layer: str + :param epsg: EPSG code used to derive the default grid subset (default: 4326) + :type epsg: int, optional + :param id: Optional GeoWebCache layer id + :type id: str, optional + :param enabled: Whether the cached layer is enabled (default: True) + :type enabled: bool, optional + :param grid_subsets: List of grid subsets to cache (default: derived from epsg) + :type grid_subsets: list of GridSubset, optional + :param mime_formats: List of MIME types to cache, e.g. ["image/png"] + :type mime_formats: list of str, optional + :param parameter_filters: List of parameter filters applied to the cache + :type parameter_filters: list of ParameterFilter, optional + :param meta_width_height: Meta-tiling factors as [width, height] + :type meta_width_height: list of int, optional + :param gutter: Gutter size in pixels + :type gutter: int, optional + :param expire_cache: Cache expiration time in seconds + :type expire_cache: int, optional + :param expire_clients: Client cache expiration time in seconds + :type expire_clients: int, optional + :param cache_warning_skips: List of warnings to skip when seeding the cache + :type cache_warning_skips: list, optional + :return: Tuple of (content, status_code) + :rtype: tuple + """ + gwc_layer = GwcLayer( + workspace_name=workspace_name, + layer_name=layer, + id=id, + enabled=enabled, + grid_subsets=grid_subsets or [GridSubset(grid_set_name=f"EPSG:{epsg}")], + mime_formats=mime_formats, + parameter_filters=parameter_filters, + meta_width_height=meta_width_height, + gutter=gutter, + expire_cache=expire_cache, + expire_clients=expire_clients, + cache_warning_skips=cache_warning_skips, + ) + return self.rest_service.publish_gwc_layer(gwc_layer) def delete_gwc_layer(self, workspace_name: str, layer: str) -> tuple[str, int]: + """ + Delete a GeoWebCache layer + + :param workspace_name: Name of the workspace + :type workspace_name: str + :param layer: Name of the layer + :type layer: str + :return: Tuple of (content, status_code) + :rtype: tuple + """ return self.rest_service.delete_gwc_layer(workspace_name, layer) def get_styles( @@ -1077,6 +1604,11 @@ def get_styles( ) -> tuple[list[dict[str, str]] | str, int]: """ Get all styles for a given workspace. If no workspace is provided, get all global styles + + :param workspace_name: Name of the workspace, or None for global styles + :type workspace_name: str, optional + :return: Tuple of (styles, status_code) + :rtype: tuple """ content, code = self.rest_service.get_styles(workspace_name) if isinstance(content, str): @@ -1088,6 +1620,13 @@ def get_style_definition( ) -> tuple[dict[str, Any] | str, int]: """ Get a style definition by name + + :param style: Name of the style + :type style: str + :param workspace_name: Name of the workspace, or None for a global style + :type workspace_name: str, optional + :return: Tuple of (style_definition, status_code) + :rtype: tuple """ content, code = self.rest_service.get_style_definition(style, workspace_name) if isinstance(content, str): @@ -1101,7 +1640,20 @@ def create_style_definition( workspace_name: str | None = None, format: str = "sld", ) -> tuple[str, int]: - """Create a style definition""" + """ + Create a style definition, or update it if it already exists. + + :param style_name: Name for the style + :type style_name: str + :param filename: Filename of the style resource, e.g. "mystyle.sld" + :type filename: str + :param workspace_name: Name of the workspace, or None for a global style + :type workspace_name: str, optional + :param format: Style format, e.g. "sld" or "mbstyle" (default: "sld") + :type format: str, optional + :return: Tuple of (content, status_code) + :rtype: tuple + """ style = Style( name=style_name, filename=filename, @@ -1118,7 +1670,18 @@ def create_style_from_string( style_string: str, workspace_name: str | None = None, ) -> tuple[str, int]: - """Create a style (SLD) from its definition as a string or update it if it already exists.""" + """ + Create a style (SLD) from its definition as a string, or update it if it already exists. + + :param style_name: Name for the style + :type style_name: str + :param style_string: SLD style definition as a string + :type style_string: str + :param workspace_name: Name of the workspace, or None for a global style + :type workspace_name: str, optional + :return: Tuple of (content, status_code) + :rtype: tuple + """ content, code = self.create_style_definition( style_name, f"{style_name}.sld", workspace_name ) @@ -1134,8 +1697,19 @@ def create_style_from_file( file: str, workspace_name: str | None = None, ) -> tuple[str, int]: - """Create a style from a file, or update it if it already exists. - Supported file extensions are .sld, .zip and .mbstyle.""" + """ + Create a style from a file, or update it if it already exists. + Supported file extensions are .sld, .zip and .mbstyle. + + :param style_name: Name for the style + :type style_name: str + :param file: Path to the style file (.sld, .zip or .mbstyle) + :type file: str + :param workspace_name: Name of the workspace, or None for a global style + :type workspace_name: str, optional + :return: Tuple of (content, status_code) + :rtype: tuple + """ file_ext = Path(file).suffix.lower() if file_ext == ".sld": style_format = "sld" @@ -1168,16 +1742,45 @@ def create_style_from_file( def set_default_layer_style( self, layer_name: str, workspace_name: str, style: str ) -> tuple[str, int]: - """Set the default style for a layer""" + """ + Set the default style for a layer + + :param layer_name: Name of the layer + :type layer_name: str + :param workspace_name: Name of the workspace + :type workspace_name: str + :param style: Name of the style to set as default + :type style: str + :return: Tuple of (content, status_code) + :rtype: tuple + """ layer = Layer(layer_name, default_style_name=style) return self.rest_service.update_layer(layer, workspace_name) def get_wms_layers( self, workspace_name: str, accept_languages: str | None = None ) -> Any | dict[str, Any]: + """ + Get the capabilities of all WMS layers for a given workspace + + :param workspace_name: Name of the workspace + :type workspace_name: str + :param accept_languages: Optional comma-separated list of preferred languages for localized content + :type accept_languages: str, optional + :return: Parsed WMS capabilities document + :rtype: Any or dict + """ return self.ows_service.get_wms_layers(workspace_name, accept_languages) def get_wfs_layers(self, workspace_name: str) -> Any | dict[str, Any]: + """ + Get the capabilities of all WFS layers for a given workspace + + :param workspace_name: Name of the workspace + :type workspace_name: str + :return: Parsed WFS capabilities document + :rtype: Any or dict + """ return self.ows_service.get_wfs_layers(workspace_name) def get_map( @@ -1194,6 +1797,27 @@ def get_map( ) -> ResponseWrapper | None: """ WMS GetMap request + + :param layers: List of layer names to render + :type layers: list of str + :param bbox: Bounding box as (minx, miny, maxx, maxy) + :type bbox: tuple of float + :param size: Image size as (width, height) in pixels + :type size: tuple of int + :param srs: Spatial reference system (default: "EPSG:2056") + :type srs: str, optional + :param format: Image format (default: "image/png") + :type format: str, optional + :param transparent: Whether the background should be transparent (default: True) + :type transparent: bool, optional + :param styles: List of style names to apply to the layers + :type styles: list of str, optional + :param language: Optional language code for localized content + :type language: str, optional + :param time: Optional time value for time-enabled layers + :type time: str, optional + :return: owslib.util.ResponseWrapper with the map image, or None + :rtype: ResponseWrapper, optional """ if not self.wms: self.create_wms() @@ -1231,6 +1855,29 @@ def get_feature_info( ) -> ResponseWrapper | None: """ WMS GetFeatureInfo request + + :param layers: List of layer names to query + :type layers: list of str + :param bbox: Bounding box as (minx, miny, maxx, maxy) + :type bbox: tuple of float + :param size: Image size as (width, height) in pixels + :type size: tuple of int + :param srs: Spatial reference system (default: "EPSG:2056") + :type srs: str, optional + :param info_format: Format of the returned feature info (default: "application/json") + :type info_format: str, optional + :param transparent: Whether the background should be transparent (default: True) + :type transparent: bool, optional + :param styles: List of style names to apply to the layers + :type styles: list of str, optional + :param xy: Pixel coordinates (x, y) to query (default: [0, 0]) + :type xy: list of float, optional + :param time: Optional time value for time-enabled layers + :type time: str, optional + :param workspace_name: Optional workspace name + :type workspace_name: str, optional + :return: owslib.util.ResponseWrapper with the feature info, or None + :rtype: ResponseWrapper, optional """ if not self.wms: self.create_wms(workspace_name) @@ -1261,6 +1908,19 @@ def get_legend_graphic( ) -> Response: """ WMS GetLegendGraphic request + + :param layer: Name of the layer(s) to get a legend for + :type layer: str or list of str + :param format: Image format (default: "image/png") + :type format: str, optional + :param language: Optional language code for localized content + :type language: str, optional + :param style: Optional style name to use for the legend + :type style: str, optional + :param workspace_name: Optional workspace name + :type workspace_name: str, optional + :return: HTTP response with the legend image + :rtype: requests.Response """ return self.ows_service.get_legend_graphic( layer, format, language, style, workspace_name @@ -1309,8 +1969,22 @@ def get_feature( max_feature: int | None = None, format: str = "application/json", ) -> dict[str, Any] | str: - """WFS GetFeature request + """ + WFS GetFeature request Return the feature(s) as dict if found, otherwise return the response content as string + + :param workspace_name: Name of the workspace + :type workspace_name: str + :param type_name: Name of the feature type + :type type_name: str + :param feature_id: Optional feature id to fetch a single feature + :type feature_id: int, optional + :param max_feature: Maximum number of features to return + :type max_feature: int, optional + :param format: Response format (default: "application/json") + :type format: str, optional + :return: Feature(s) as a dict, or the response content as a string + :rtype: dict or str """ # FIXME: we should consider also the global wfs endpoint return self.ows_service.get_feature( @@ -1323,8 +1997,18 @@ def describe_feature_type( type_name: str | None = None, format: str = "application/json", ) -> dict[str, Any] | str: - """WFS DescribeFeatureType request + """ + WFS DescribeFeatureType request Return the feature type(s) as dict if found, otherwise return the response content as string + + :param workspace_name: Optional workspace name + :type workspace_name: str, optional + :param type_name: Optional name of the feature type + :type type_name: str, optional + :param format: Response format (default: "application/json") + :type format: str, optional + :return: Feature type(s) as a dict, or the response content as a string + :rtype: dict or str """ return self.ows_service.describe_feature_type(workspace_name, type_name, format) @@ -1334,9 +2018,19 @@ def get_property_value( type_name: str, property: str, ) -> dict | list | str: - """WFS GetPropertyValue request + """ + WFS GetPropertyValue request Return the properties as dict (if one feature was found), a list (if multiple features were found), an empty dict if no feature was found or the response content as string + + :param workspace_name: Name of the workspace + :type workspace_name: str + :param type_name: Name of the feature type + :type type_name: str + :param property: Name of the property to fetch + :type property: str + :return: Property value(s) as a dict or list, or the response content as a string + :rtype: dict or list or str """ # FIXME: we should consider also the global wfs endpoint return self.ows_service.get_property_value(workspace_name, type_name, property) @@ -1346,6 +2040,15 @@ def create_user( ) -> tuple[str, int]: """ Create a GeoServer user + + :param user: Username + :type user: str + :param password: Password for the user + :type password: str + :param enabled: Whether the user should be enabled (default: True) + :type enabled: bool, optional + :return: Tuple of (content, status_code) + :rtype: tuple """ return self.rest_service.create_user(user, password, enabled) @@ -1354,42 +2057,85 @@ def update_user( ) -> tuple[str, int]: """ Update a GeoServer user + + :param user: Username + :type user: str + :param password: New password for the user + :type password: str, optional + :param enabled: Whether the user should be enabled + :type enabled: bool, optional + :return: Tuple of (content, status_code) + :rtype: tuple """ return self.rest_service.update_user(user, password, enabled) def delete_user(self, user: str) -> tuple[str, int]: """ Delete a GeoServer user + + :param user: Username + :type user: str + :return: Tuple of (content, status_code) + :rtype: tuple """ return self.rest_service.delete_user(user) def create_role(self, role_name: str) -> tuple[str, int]: """ Create a GeoServer role if it does not already exist + + :param role_name: Name of the role + :type role_name: str + :return: Tuple of (content, status_code) + :rtype: tuple """ return self.rest_service.create_role_if_not_exists(role_name) def delete_role(self, role_name: str) -> tuple[str, int]: """ Delete a GeoServer role + + :param role_name: Name of the role + :type role_name: str + :return: Tuple of (content, status_code) + :rtype: tuple """ return self.rest_service.delete_role(role_name) def get_user_roles(self, user: str) -> tuple[list[str] | str, int]: """ Get all roles assigned to a GeoServer user + + :param user: Username + :type user: str + :return: Tuple of (roles, status_code) + :rtype: tuple """ return self.rest_service.get_user_roles(user) def assign_role_to_user(self, user: str, role: str) -> tuple[str, int]: """ Assign a role to a GeoServer user + + :param user: Username + :type user: str + :param role: Name of the role + :type role: str + :return: Tuple of (content, status_code) + :rtype: tuple """ return self.rest_service.assign_role_to_user(user, role) def remove_role_from_user(self, user: str, role: str) -> tuple[str, int]: """ Remove a role from a GeoServer user + + :param user: Username + :type user: str + :param role: Name of the role + :type role: str + :return: Tuple of (content, status_code) + :rtype: tuple """ return self.rest_service.remove_role_from_user(user, role) @@ -1403,6 +2149,19 @@ def create_acl_admin_rule( ) -> tuple[dict | str, int]: """ Create a GeoServer ACL admin rule + + :param priority: Rule priority (default: 0) + :type priority: int, optional + :param access: Access level, e.g. "ADMIN" (default: "ADMIN") + :type access: str, optional + :param role: Optional role the rule applies to + :type role: str, optional + :param user: Optional user the rule applies to + :type user: str, optional + :param workspace_name: Optional workspace the rule applies to + :type workspace_name: str, optional + :return: Tuple of (content, status_code) + :rtype: tuple """ return self.rest_service.create_acl_admin_rule( priority, access, role, user, workspace_name @@ -1411,18 +2170,29 @@ def create_acl_admin_rule( def delete_acl_admin_rule(self, id: int | str) -> tuple[str, int]: """ Delete a GeoServer ACL admin rule by id + + :param id: Id of the ACL admin rule + :type id: int or str + :return: Tuple of (content, status_code) + :rtype: tuple """ return self.rest_service.delete_acl_admin_rule(str(id)) def delete_all_acl_admin_rules(self) -> tuple[str, int]: """ Delete all existing GeoServer ACL admin rules + + :return: Tuple of (content, status_code) + :rtype: tuple """ return self.rest_service.delete_all_acl_admin_rules() def get_acl_rules(self) -> tuple[dict[str, Any] | str, int]: """ Return all GeoServer ACL data rules + + :return: Tuple of (rules, status_code) + :rtype: tuple """ return self.rest_service.get_acl_rules() @@ -1436,7 +2206,22 @@ def create_acl_rules_for_requests( workspace_name: str | None = None, ) -> list[tuple[dict | str, int]]: """ - Create ACL rules for multiple type of OGC requests + Create ACL rules for multiple types of OGC requests + + :param requests: List of request types, e.g. ["GetMap", "GetFeatureInfo"] + :type requests: list of str + :param priority: Base priority for the rules (default: 0) + :type priority: int, optional + :param access: Access level, e.g. "DENY" (default: "DENY") + :type access: str, optional + :param role: Optional role the rules apply to + :type role: str, optional + :param service: Optional OGC service the rules apply to, e.g. "WMS" + :type service: str, optional + :param workspace_name: Optional workspace the rules apply to + :type workspace_name: str, optional + :return: List of tuples of (content, status_code), one per created rule + :rtype: list of tuple """ return self.rest_service.create_acl_rules_for_requests( requests, priority, access, role, service, workspace_name @@ -1454,6 +2239,23 @@ def create_acl_rule( ) -> tuple[dict | str, int]: """ Create a GeoServer ACL data rule + + :param priority: Rule priority (default: 0) + :type priority: int, optional + :param access: Access level, e.g. "DENY" (default: "DENY") + :type access: str, optional + :param role: Optional role the rule applies to + :type role: str, optional + :param user: Optional user the rule applies to + :type user: str, optional + :param service: Optional OGC service the rule applies to, e.g. "WMS" + :type service: str, optional + :param request: Optional request type the rule applies to, e.g. "GetMap" + :type request: str, optional + :param workspace_name: Optional workspace the rule applies to + :type workspace_name: str, optional + :return: Tuple of (content, status_code) + :rtype: tuple """ return self.rest_service.create_acl_rule( priority, access, role, user, service, request, workspace_name @@ -1462,6 +2264,9 @@ def create_acl_rule( def delete_all_acl_rules(self) -> tuple[str, int]: """ Delete all existing GeoServer ACL data rules + + :return: Tuple of (content, status_code) + :rtype: tuple """ return self.rest_service.delete_all_acl_rules() @@ -1469,5 +2274,10 @@ def create_gridset(self, epsg: int) -> tuple[str, int]: """ Create a gridset for GeoWebCache for a given projection Supported EPSG codes are 2056, 21781 and 3857 + + :param epsg: EPSG code for the gridset + :type epsg: int + :return: Tuple of (content, status_code) + :rtype: tuple """ return self.rest_service.create_gridset(epsg) diff --git a/geoservercloud/models/gwclayer.py b/geoservercloud/models/gwclayer.py new file mode 100644 index 0000000..f0b3ca6 --- /dev/null +++ b/geoservercloud/models/gwclayer.py @@ -0,0 +1,161 @@ +from typing import Any + +from geoservercloud.models.common import BaseModel, EntityModel + + +class GridSubsetExtent(BaseModel): + def __init__(self, coords: list[float]): + self.coords: list[float] = coords + + @classmethod + def from_get_response_payload(cls, content: dict): + return cls(coords=content["coords"]) + + def asdict(self) -> dict[str, Any]: + return {"coords": self.coords} + + +class GridSubset(BaseModel): + def __init__( + self, + grid_set_name: str, + extent: GridSubsetExtent | None = None, + ): + self.grid_set_name: str = grid_set_name + self.extent: GridSubsetExtent | None = extent + + @classmethod + def from_get_response_payload(cls, content: dict): + extent = content.get("extent") + return cls( + grid_set_name=content["gridSetName"], + extent=( + GridSubsetExtent.from_get_response_payload(extent) if extent else None + ), + ) + + def asdict(self) -> dict[str, Any]: + content: dict[str, Any] = {"gridSetName": self.grid_set_name} + if self.extent is not None: + content["extent"] = self.extent.asdict() + return content + + +class ParameterFilter(BaseModel): + def __init__(self, key: str, default_value: str = ""): + self.key: str = key + self.default_value: str = default_value + + @classmethod + def from_get_response_payload(cls, content: dict): + return cls(key=content["key"], default_value=content.get("defaultValue", "")) + + def asdict(self) -> dict[str, Any]: + return {"key": self.key, "defaultValue": self.default_value} + + +class GwcLayer(EntityModel): + def __init__( + self, + # Mandatory fields + workspace_name: str, + layer_name: str, + # Nullable fields + id: str | None = None, + enabled: bool | None = None, + grid_subsets: list[GridSubset] | None = None, + mime_formats: list[str] | None = None, + parameter_filters: list[ParameterFilter] | None = None, + meta_width_height: list[int] | None = None, + gutter: int | None = None, + expire_cache: int | None = None, + expire_clients: int | None = None, + cache_warning_skips: list[Any] | None = None, + ): + self.workspace_name: str = workspace_name + self.layer_name: str = layer_name + self.id: str | None = id + self.enabled: bool | None = enabled + self.grid_subsets: list[GridSubset] | None = grid_subsets + self.mime_formats: list[str] | None = mime_formats + self.parameter_filters: list[ParameterFilter] | None = parameter_filters + self.meta_width_height: list[int] | None = meta_width_height + self.gutter: int | None = gutter + self.expire_cache: int | None = expire_cache + self.expire_clients: int | None = expire_clients + self.cache_warning_skips: list[Any] | None = cache_warning_skips + + @property + def name(self) -> str: + return f"{self.workspace_name}:{self.layer_name}" + + @classmethod + def from_get_response_payload(cls, content: dict): + gwc_layer = content["GeoServerLayer"] + workspace_name, layer_name = gwc_layer["name"].split(":", 1) + grid_subsets = gwc_layer.get("gridSubsets") + if isinstance(grid_subsets, dict): + grid_subsets = grid_subsets.get("gridSubset", []) + parameter_filters = gwc_layer.get("parameterFilters") + meta_width_height = gwc_layer.get("metaWidthHeight") + if isinstance(meta_width_height, dict): + meta_width_height = meta_width_height.get("int", []) + return cls( + workspace_name=workspace_name, + layer_name=layer_name, + id=gwc_layer.get("id"), + enabled=gwc_layer.get("enabled"), + grid_subsets=( + [GridSubset.from_get_response_payload(item) for item in grid_subsets] + if grid_subsets is not None + else None + ), + mime_formats=gwc_layer.get("mimeFormats"), + parameter_filters=( + [ + ParameterFilter.from_get_response_payload(item) + for item in parameter_filters + ] + if parameter_filters is not None + else None + ), + meta_width_height=meta_width_height, + gutter=gwc_layer.get("gutter"), + expire_cache=gwc_layer.get("expireCache"), + expire_clients=gwc_layer.get("expireClients"), + cache_warning_skips=gwc_layer.get("cacheWarningSkips"), + ) + + def asdict(self) -> dict[str, Any]: + content: dict[str, Any] = {"name": self.name} + optional_items: dict[str, Any] = { + "id": self.id, + "enabled": self.enabled, + "gridSubsets": ( + {"gridSubset": [item.asdict() for item in self.grid_subsets]} + if self.grid_subsets is not None + else None + ), + "mimeFormats": self.mime_formats, + "parameterFilters": ( + [item.asdict() for item in self.parameter_filters] + if self.parameter_filters is not None + else None + ), + "metaWidthHeight": ( + {"int": self.meta_width_height} + if self.meta_width_height is not None + else None + ), + "gutter": self.gutter, + "expireCache": self.expire_cache, + "expireClients": self.expire_clients, + "cacheWarningSkips": self.cache_warning_skips, + } + return EntityModel.add_items_to_dict(content, optional_items) + + def post_payload(self) -> dict[str, Any]: + return {"GeoServerLayer": self.asdict()} + + def put_payload(self) -> dict[str, Any]: + return self.post_payload() diff --git a/geoservercloud/models/wmtsstore.py b/geoservercloud/models/wmtsstore.py new file mode 100644 index 0000000..201abdb --- /dev/null +++ b/geoservercloud/models/wmtsstore.py @@ -0,0 +1,121 @@ +import json +from typing import Any + +from geoservercloud.models.common import ( + EntityModel, + KeyDollarListDict, + ReferencedObjectModel, +) + + +class WmtsStore(EntityModel): + def __init__( + self, + workspace_name: str, + name: str, + capabilities_url: str, + type: str = "WMTS", + enabled: bool = True, + default: bool | None = None, + disable_on_conn_failure: bool | None = None, + use_connection_pooling: bool | None = None, + max_connections: int | None = None, + read_timeout: int | None = None, + connect_timeout: int | None = None, + date_created: str | None = None, + date_modified: str | None = None, + ) -> None: + self.workspace: ReferencedObjectModel = ReferencedObjectModel(workspace_name) + self._name: str = name + self.capabilities_url: str = capabilities_url + self.type: str = type + self.enabled: bool = enabled + self._default: bool | None = default + self.disable_on_conn_failure: bool | None = disable_on_conn_failure + self.metadata: KeyDollarListDict | None = None + if use_connection_pooling is not None: + self.metadata = KeyDollarListDict( + input_dict={"useConnectionPooling": str(use_connection_pooling).lower()} + ) + self.max_connections: int | None = max_connections + self.read_timeout: int | None = read_timeout + self.connect_timeout: int | None = connect_timeout + self.date_created: str | None = date_created + self.date_modified: str | None = date_modified + + @property + def name(self) -> str: + return self._name + + @property + def workspace_name(self) -> str: + return self.workspace.name + + @property + def use_connection_pooling(self) -> bool | None: + if self.metadata is None: + return None + value = self.metadata.get("useConnectionPooling") + return None if value is None else value == "true" + + def asdict(self) -> dict[str, Any]: + content: dict[str, Any] = { + "name": self._name, + "type": self.type, + "workspace": self.workspace_name, + "capabilitiesURL": self.capabilities_url, + } + optional_items: dict[str, Any] = { + "enabled": self.enabled, + "_default": self._default, + "disableOnConnFailure": self.disable_on_conn_failure, + "maxConnections": self.max_connections, + "readTimeout": self.read_timeout, + "connectTimeout": self.connect_timeout, + "dateCreated": self.date_created, + "dateModified": self.date_modified, + } + if self.metadata: + optional_items["metadata"] = {"entry": dict(self.metadata)} + return EntityModel.add_items_to_dict(content, optional_items) + + def post_payload(self) -> dict[str, Any]: + content = self.asdict() + content["workspace"] = {"name": self.workspace_name} + if self.metadata: + content["metadata"] = {"entry": self.metadata.serialize()} + return {"wmtsStore": content} + + def put_payload(self) -> dict[str, Any]: + return self.post_payload() + + @classmethod + def from_get_response_payload(cls, content: dict): + wmts_store = content["wmtsStore"] + use_connection_pooling: bool | None = None + metadata_entry = wmts_store.get("metadata", {}).get("entry") + if isinstance(metadata_entry, dict): + metadata_entry = [metadata_entry] + if metadata_entry: + metadata = KeyDollarListDict(input_list=metadata_entry) + value = metadata.get("useConnectionPooling") + if value is not None: + use_connection_pooling = str(value).lower() == "true" + return cls( + wmts_store["workspace"]["name"], + wmts_store["name"], + wmts_store["capabilitiesURL"], + wmts_store.get("type", "WMTS"), + wmts_store.get("enabled", True), + wmts_store.get("_default", None), + wmts_store.get("disableOnConnFailure", None), + use_connection_pooling, + wmts_store.get("maxConnections", None), + wmts_store.get("readTimeout", None), + wmts_store.get("connectTimeout", None), + wmts_store.get("dateCreated", None), + wmts_store.get("dateModified", None), + ) + + def __repr__(self) -> str: + return json.dumps(self.put_payload(), indent=4) diff --git a/geoservercloud/services/restservice.py b/geoservercloud/services/restservice.py index 249485f..c3494bd 100644 --- a/geoservercloud/services/restservice.py +++ b/geoservercloud/services/restservice.py @@ -13,6 +13,7 @@ from geoservercloud.models.datastores import DataStores from geoservercloud.models.featuretype import FeatureType from geoservercloud.models.featuretypes import FeatureTypes +from geoservercloud.models.gwclayer import GwcLayer from geoservercloud.models.layer import Layer from geoservercloud.models.layergroup import LayerGroup from geoservercloud.models.layergroups import LayerGroups @@ -22,6 +23,7 @@ from geoservercloud.models.wmslayer import WmsLayer from geoservercloud.models.wmssettings import WmsSettings from geoservercloud.models.wmsstore import WmsStore +from geoservercloud.models.wmtsstore import WmtsStore from geoservercloud.models.workspace import Workspace from geoservercloud.models.workspaces import Workspaces from geoservercloud.services.restclient import RestClient @@ -212,21 +214,19 @@ def delete_wms_layer( return response.content.decode(), response.status_code def create_wmts_store( - self, - workspace_name: str, - name: str, - capabilities: str, + self, workspace_name: str, wmts_store: WmtsStore ) -> tuple[str, int]: - payload = Templates.wmts_store(workspace_name, name, capabilities) if not self.resource_exists( - self.rest_endpoints.wmtsstore(workspace_name, name) + self.rest_endpoints.wmtsstore(workspace_name, wmts_store.name) ): response: Response = self.rest_client.post( - self.rest_endpoints.wmtsstores(workspace_name), json=payload + self.rest_endpoints.wmtsstores(workspace_name), + json=wmts_store.post_payload(), ) else: response = self.rest_client.put( - self.rest_endpoints.wmtsstore(workspace_name, name), json=payload + self.rest_endpoints.wmtsstore(workspace_name, wmts_store.name), + json=wmts_store.put_payload(), ) return response.content.decode(), response.status_code @@ -299,9 +299,7 @@ def get_gwc_layer( content = response.content.decode() return content, response.status_code - def publish_gwc_layer( - self, workspace_name: str, layer: str, epsg: int - ) -> tuple[str, int]: + def publish_gwc_layer(self, gwc_layer: GwcLayer) -> tuple[str, int]: # Reload config to make sure GWC is aware of GeoServer layers self.rest_client.post( self.gwc_endpoints.reload(), @@ -311,13 +309,14 @@ def publish_gwc_layer( # Do not re-publish an existing layer # TODO: fix template so that we can PUT an existing layer (/!\ check with an OGC client that the # layer is not corrupted after the second PUT) - content, code = self.get_gwc_layer(workspace_name, layer) + content, code = self.get_gwc_layer( + gwc_layer.workspace_name, gwc_layer.layer_name + ) if code == 200: return "", code - payload = Templates.gwc_layer(workspace_name, layer, f"EPSG:{epsg}") response: Response = self.rest_client.put( - self.gwc_endpoints.layer(workspace_name, layer), - json=payload, + self.gwc_endpoints.layer(gwc_layer.workspace_name, gwc_layer.layer_name), + json=gwc_layer.put_payload(), ) return response.content.decode(), response.status_code diff --git a/geoservercloud/templates.py b/geoservercloud/templates.py index bfdc2f7..9532063 100644 --- a/geoservercloud/templates.py +++ b/geoservercloud/templates.py @@ -4,21 +4,6 @@ class Templates: - @staticmethod - def wmts_store( - workspace: str, name: str, capabilities: str - ) -> dict[str, dict[str, Any]]: - return { - "wmtsStore": { - "name": name, - "type": "WMTS", - "capabilitiesURL": capabilities, - "workspace": {"name": workspace}, - "enabled": True, - "metadata": {"entry": {"@key": "useConnectionPooling", "text": True}}, - } - } - @staticmethod def geom_point_attribute() -> dict[str, Any]: return { @@ -70,15 +55,3 @@ def wmts_layer( if international_abstract: template["wmtsLayer"]["internationalAbstract"] = international_abstract return template - - @staticmethod - def gwc_layer( - workspace: str, layer: str, gridset: str - ) -> dict[str, dict[str, Any]]: - return { - "GeoServerLayer": { - "name": f"{workspace}:{layer}", - "enabled": "true", - "gridSubsets": {"gridSubset": [{"gridSetName": gridset}]}, - } - } diff --git a/tests/test_cascaded_wmts.py b/tests/test_cascaded_wmts.py index 7d688ef..a83b563 100644 --- a/tests/test_cascaded_wmts.py +++ b/tests/test_cascaded_wmts.py @@ -34,7 +34,7 @@ def wmts_store_payload() -> dict[str, dict[str, Any]]: "capabilitiesURL": CAPABILITIES_URL, "workspace": {"name": WORKSPACE}, "enabled": True, - "metadata": {"entry": {"@key": "useConnectionPooling", "text": True}}, + "metadata": {"entry": [{"@key": "useConnectionPooling", "$": "true"}]}, } } diff --git a/tests/test_gwc.py b/tests/test_gwc.py index 54577ce..2776065 100644 --- a/tests/test_gwc.py +++ b/tests/test_gwc.py @@ -48,7 +48,7 @@ def test_publish_gwc_layer(geoserver: GeoServerCloud) -> None: { "GeoServerLayer": { "name": f"{WORKSPACE}:{LAYER}", - "enabled": "true", + "enabled": True, "gridSubsets": { "gridSubset": [{"gridSetName": f"EPSG:{EPSG}"}] },