Add data class for GWC layer and WMTS datastore

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()

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)

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
 

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}]},
-            }
-        }