Copy layer group

geoservercloud/geoservercloud.py

@@ -10,6 +10,7 @@
 from geoservercloud.models.datastore import PostGisDataStore
 from geoservercloud.models.featuretype import FeatureType
 from geoservercloud.models.layer import Layer
+from geoservercloud.models.layergroup import LayerGroup
 from geoservercloud.models.style import Style
 from geoservercloud.models.wmssettings import WmsSettings
 from geoservercloud.models.workspace import Workspace
@@ -402,6 +403,30 @@ def delete_feature_type(
             workspace_name, datastore_name, layer_name
         )
 
+    def get_layer_groups(
+        self, workspace_name: str
+    ) -> tuple[list[dict[str, str]] | str, int]:
+        """
+        Get all layer groups for a given workspace
+        """
+        layer_groups, status_code = self.rest_service.get_layer_groups(workspace_name)
+        if isinstance(layer_groups, str):
+            return layer_groups, status_code
+        return layer_groups.aslist(), status_code
+
+    def get_layer_group(
+        self, workspace_name: str, layer_group_name: str
+    ) -> tuple[dict[str, Any] | str, int]:
+        """
+        Get a layer group by name
+        """
+        layer_group, status_code = self.rest_service.get_layer_group(
+            workspace_name, layer_group_name
+        )
+        if isinstance(layer_group, str):
+            return layer_group, status_code
+        return layer_group.asdict(), status_code
+
     def create_layer_group(
         self,
         group: str,
@@ -411,16 +436,38 @@ def create_layer_group(
         abstract: str | dict,
         epsg: int = 4326,
         mode: str = "SINGLE",
+        enabled: bool = True,
+        advertised: bool = True,
     ) -> tuple[str, int]:
         """
-        Create a layer group if it does not already exist.
+        Create a layer group or update it if it already exists.
         """
         workspace_name = workspace_name or self.default_workspace
         if not workspace_name:
             raise ValueError("Workspace not provided")
-        return self.rest_service.create_layer_group(
-            group, workspace_name, layers, title, abstract, epsg, mode
+        if not mode in LayerGroup.modes:
+            raise ValueError(
+                f"Invalid mode: {mode}, possible values are: {LayerGroup.modes}"
+            )
+        bounds = {
+            "minx": utils.EPSG_BBOX[epsg]["nativeBoundingBox"]["minx"],
+            "maxx": utils.EPSG_BBOX[epsg]["nativeBoundingBox"]["maxx"],
+            "miny": utils.EPSG_BBOX[epsg]["nativeBoundingBox"]["miny"],
+            "maxy": utils.EPSG_BBOX[epsg]["nativeBoundingBox"]["maxy"],
+            "crs": f"EPSG:{epsg}",
+        }
+        layer_group = LayerGroup(
+            name=group,
+            mode=mode,
+            workspace_name=workspace_name,
+            title=title,
+            abstract=abstract,
+            publishables=[f"{workspace_name}:{layer}" for layer in layers],
+            bounds=bounds,
+            enabled=enabled,
+            advertised=advertised,
         )
+        return self.rest_service.create_layer_group(group, workspace_name, layer_group)
 
     def create_wmts_layer(
         self,

geoservercloud/geoservercloudsync.py

@@ -1,6 +1,5 @@
 from argparse import ArgumentParser
 
-from geoservercloud.models.resourcedirectory import ResourceDirectory
 from geoservercloud.services import RestService
 
 
@@ -165,6 +164,21 @@ def copy_layer(
             return layer, status_code
         return self.dst_instance.update_layer(layer, workspace_name)
 
+    def copy_layer_group(
+        self, workspace_name: str, layer_group_name: str
+    ) -> tuple[str, int]:
+        """
+        Copy a layer group from source to destination GeoServer instance
+        """
+        layer_group, status_code = self.src_instance.get_layer_group(
+            workspace_name, layer_group_name
+        )
+        if isinstance(layer_group, str):
+            return layer_group, status_code
+        return self.dst_instance.create_layer_group(
+            layer_group_name, workspace_name, layer_group
+        )
+
     def copy_styles(
         self, workspace_name: str | None = None, include_images: bool = True
     ) -> tuple[str, int]:

geoservercloud/models/layergroup.py

@@ -0,0 +1,117 @@
+from typing import Any
+
+from geoservercloud.models.common import I18N, EntityModel, ReferencedObjectModel
+
+
+class LayerGroup(EntityModel):
+    modes = ["SINGLE", "OPAQUE_CONTAINER", "NAMED", "CONTAINER", "EO"]
+
+    def __init__(
+        self,
+        name: str | None = None,
+        mode: str | None = None,
+        enabled: bool | None = None,
+        advertised: bool | None = None,
+        workspace_name: str | None = None,
+        title: dict[str, str] | str | None = None,
+        abstract: dict[str, str] | str | None = None,
+        publishables: list[str] | None = None,
+        styles: list[str] | None = None,
+        bounds: dict[str, Any] | None = None,
+    ):
+        self.name: str | None = name
+        self.mode: str | None = mode
+        self.enabled: bool | None = enabled
+        self.advertised: bool | None = advertised
+        self.workspace: ReferencedObjectModel | None = (
+            ReferencedObjectModel(workspace_name) if workspace_name else None
+        )
+        self.title: I18N | None = (
+            I18N(("title", "internationalTitle"), title) if title else None
+        )
+        self.abstract: I18N | None = (
+            I18N(("abstract", "internationalAbstract"), abstract) if abstract else None
+        )
+        self.publishables: list[ReferencedObjectModel] | None = (
+            [ReferencedObjectModel(publishable) for publishable in publishables]
+            if publishables
+            else None
+        )
+        self.styles: list[ReferencedObjectModel] | None = (
+            [ReferencedObjectModel(style) for style in styles] if styles else None
+        )
+        self.bounds: dict[str, int | str] | None = bounds
+
+    @property
+    def workspace_name(self) -> str | None:
+        return self.workspace.name if self.workspace else None
+
+    @classmethod
+    def from_get_response_payload(cls, content):
+        layer_group: dict[str, Any] = content["layerGroup"]
+        # publishables: list of dict or dict (if only one layer)
+        publishables: list[dict[str, str]] | dict[str, str] = layer_group[
+            "publishables"
+        ]["published"]
+        if isinstance(publishables, dict):
+            publishables = [publishables]
+        # style: list of dict, dict (if only one layer) or list of empty strings (if using default layer styles)
+        styles: list[dict] | dict | list[str] = layer_group["styles"]["style"]
+        if isinstance(styles, dict):
+            styles = [styles["name"]]
+        if isinstance(styles, list):
+            styles = [s["name"] if isinstance(s, dict) else s for s in styles]
+        return cls(
+            name=layer_group["name"],
+            mode=layer_group["mode"],
+            enabled=layer_group.get("enabled"),
+            advertised=layer_group.get("advertised"),
+            workspace_name=layer_group["workspace"]["name"],
+            title=layer_group.get("internationalTitle", layer_group.get("title")),
+            abstract=layer_group.get(
+                "internationalAbstract", layer_group.get("abstract")
+            ),
+            publishables=[p["name"] for p in publishables],
+            styles=styles,
+            bounds=layer_group.get("bounds"),
+        )
+
+    def asdict(self) -> dict[str, Any]:
+        optional_items = {
+            "name": self.name,
+            "mode": self.mode,
+            "enabled": self.enabled,
+            "advertised": self.advertised,
+            "bounds": self.bounds,
+        }
+        content = EntityModel.add_items_to_dict({}, optional_items)
+        if self.workspace:
+            content["workspace"] = self.workspace.asdict()
+        if self.publishables:
+            content["publishables"] = {
+                "published": [
+                    {"@type": "layer", "name": p.name} for p in self.publishables
+                ]
+            }
+        if self.styles:
+            content["styles"] = {"style": [s.asdict() for s in self.styles]}
+        elif self.publishables:
+            content["styles"] = {"style": [{"name": ""}] * len(self.publishables)}
+        if self.title:
+            content.update(self.title.asdict())
+        if self.abstract:
+            content.update(self.abstract.asdict())
+        return content
+
+    def post_payload(self) -> dict[str, Any]:
+        return {"layerGroup": self.asdict()}
+
+    def put_payload(self) -> dict[str, Any]:
+
+        content = self.post_payload()
+        # Force a null value on non-i18ned attributes, otherwise GeoServer sets it to the first i18n value
+        if content["layerGroup"].get("internationalTitle"):
+            content["layerGroup"]["title"] = None
+        if content["layerGroup"].get("internationalAbstract"):
+            content["layerGroup"]["abstract"] = None
+        return content

geoservercloud/models/layergroups.py

@@ -0,0 +1,21 @@
+import json
+
+from geoservercloud.models.common import ListModel
+
+
+class LayerGroups(ListModel):
+    def __init__(self, layergroups: list = []) -> None:
+        self._layergroups = layergroups
+
+    @classmethod
+    def from_get_response_payload(cls, content: dict):
+        feature_types: str | dict = content["layerGroups"]
+        if not feature_types:
+            return cls()
+        return cls(feature_types["layerGroup"])  # type: ignore
+
+    def aslist(self) -> list[dict[str, str]]:
+        return self._layergroups
+
+    def __repr__(self):
+        return json.dumps(self._layergroups, indent=4)

geoservercloud/services/restservice.py

@@ -11,6 +11,8 @@
 from geoservercloud.models.featuretype import FeatureType
 from geoservercloud.models.featuretypes import FeatureTypes
 from geoservercloud.models.layer import Layer
+from geoservercloud.models.layergroup import LayerGroup
+from geoservercloud.models.layergroups import LayerGroups
 from geoservercloud.models.resourcedirectory import ResourceDirectory
 from geoservercloud.models.style import Style
 from geoservercloud.models.styles import Styles
@@ -297,34 +299,37 @@ def delete_feature_type(
         )
         return response.content.decode(), response.status_code
 
+    def get_layer_groups(self, workspace_name: str) -> tuple[LayerGroups | str, int]:
+        response: Response = self.rest_client.get(
+            self.rest_endpoints.layergroups(workspace_name)
+        )
+        return self.deserialize_response(response, LayerGroups)
+
+    def get_layer_group(
+        self, workspace_name: str, layer_group_name: str
+    ) -> tuple[LayerGroup | str, int]:
+        response: Response = self.rest_client.get(
+            self.rest_endpoints.layergroup(workspace_name, layer_group_name)
+        )
+        return self.deserialize_response(response, LayerGroup)
+
     def create_layer_group(
         self,
-        group: str,
+        layer_group_name: str,
         workspace_name: str,
-        layers: list[str],
-        title: str | dict,
-        abstract: str | dict,
-        epsg: int = 4326,
-        mode: str = "SINGLE",
+        layer_group: LayerGroup,
     ) -> tuple[str, int]:
-        payload: dict[str, dict[str, Any]] = Templates.layer_group(
-            group=group,
-            layers=layers,
-            workspace=workspace_name,
-            title=title,
-            abstract=abstract,
-            epsg=epsg,
-            mode=mode,
-        )
         if not self.resource_exists(
-            self.rest_endpoints.layergroup(workspace_name, group)
+            self.rest_endpoints.layergroup(workspace_name, layer_group_name)
         ):
             response: Response = self.rest_client.post(
-                self.rest_endpoints.layergroups(workspace_name), json=payload
+                self.rest_endpoints.layergroups(workspace_name),
+                json=layer_group.post_payload(),
             )
         else:
             response = self.rest_client.put(
-                self.rest_endpoints.layergroup(workspace_name, group), json=payload
+                self.rest_endpoints.layergroup(workspace_name, layer_group_name),
+                json=layer_group.put_payload(),
             )
         return response.content.decode(), response.status_code
 

geoservercloud/templates.py

@@ -28,54 +28,6 @@ def geom_point_attribute() -> dict[str, Any]:
             }
         }
 
-    @staticmethod
-    def layer_group(
-        group: str,
-        layers: list[str],
-        workspace: str,
-        title: str | dict[str, Any],
-        abstract: str | dict[str, Any],
-        epsg: int = 4326,
-        mode: str = "SINGLE",
-    ) -> dict[str, dict[str, Any]]:
-        modes = ["SINGLE", "OPAQUE_CONTAINER", "NAMED", "CONTAINER", "EO"]
-        if not mode in modes:
-            raise ValueError(f"Invalid mode: {mode}, possible values are: {modes}")
-        template = {
-            "layerGroup": {
-                "name": group,
-                "workspace": {"name": workspace},
-                "mode": mode,
-                "publishables": {
-                    "published": [
-                        {"@type": "layer", "name": f"{workspace}:{layer}"}
-                        for layer in layers
-                    ]
-                },
-                "styles": {"style": [{"name": ""}] * len(layers)},
-                "bounds": {
-                    "minx": EPSG_BBOX[epsg]["nativeBoundingBox"]["minx"],
-                    "maxx": EPSG_BBOX[epsg]["nativeBoundingBox"]["maxx"],
-                    "miny": EPSG_BBOX[epsg]["nativeBoundingBox"]["miny"],
-                    "maxy": EPSG_BBOX[epsg]["nativeBoundingBox"]["maxy"],
-                    "crs": f"EPSG:{epsg}",
-                },
-                "enabled": True,
-                "advertised": True,
-            }
-        }
-        if title:
-            if type(title) is dict:
-                template["layerGroup"]["internationalTitle"] = title
-            else:
-                template["layerGroup"]["title"] = title
-        if abstract:
-            if type(abstract) is dict:
-                template["layerGroup"]["internationalAbstract"] = abstract
-            else:
-                template["layerGroup"]["abstract"] = abstract
-        return template
-
     @staticmethod
     def wmts_layer(
         name: str,