Add data class for GWC layer

geoservercloud/geoservercloud.py

@@ -12,6 +12,7 @@
 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
@@ -1065,9 +1066,36 @@ def get_gwc_layer(
         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)
+        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]:
         return self.rest_service.delete_gwc_layer(workspace_name, layer)

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/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
@@ -299,9 +300,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 +310,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

@@ -70,15 +70,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}]},
-            }
-        }

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