Deep copy workspace

Copy datastores, feature types and layers in workspace

geoservercloud/geoservercloud.py

@@ -142,9 +142,7 @@ def unset_default_locale_for_service(self, workspace_name) -> tuple[str, int]:
         """
         return self.set_default_locale_for_service(workspace_name, None)
 
-    def get_datastores(
-        self, workspace_name: str
-    ) -> tuple[list[dict[str, str]] | str, int]:
+    def get_datastores(self, workspace_name: str) -> tuple[list[str] | str, int]:
         """
         Get all datastores for a given workspace
         """

geoservercloud/geoservercloudsync.py

@@ -47,7 +47,8 @@ def copy_workspace(
     ) -> tuple[str, int]:
         """
         Copy a workspace from the source to the destination GeoServer instance.
-        If deep_copy is True, the copy includes the styles in the workspace (including images).
+        If deep_copy is True, the copy includes the PostGIS datastores, the feature types in the datastores,
+        the corresponding layers and the styles in the workspace (including images).
         """
         workspace, status_code = self.src_instance.get_workspace(workspace_name)
         if isinstance(workspace, str):
@@ -61,8 +62,103 @@ def copy_workspace(
             content, status_code = self.copy_styles(workspace_name)
             if self.not_ok(status_code):
                 return content, status_code
+            content, status_code = self.copy_pg_datastores(
+                workspace_name, deep_copy=True
+            )
+            if self.not_ok(status_code):
+                return content, status_code
         return new_workspace, new_ws_status_code
 
+    def copy_pg_datastores(
+        self, workspace_name: str, deep_copy: bool = False
+    ) -> tuple[str, int]:
+        """
+        Copy all the datastores in given workspace.
+        If deep_copy is True, copy all feature types and the corresponding layers in each datastore
+        """
+        datastores, status_code = self.src_instance.get_datastores(workspace_name)
+        if isinstance(datastores, str):
+            return datastores, status_code
+        for datastore_name in datastores.aslist():
+            content, status_code = self.copy_pg_datastore(
+                workspace_name, datastore_name, deep_copy=deep_copy
+            )
+            if self.not_ok(status_code):
+                return content, status_code
+        return content, status_code
+
+    def copy_pg_datastore(
+        self, workspace_name: str, datastore_name: str, deep_copy: bool = False
+    ) -> tuple[str, int]:
+        """
+        Copy a datastore from source to destination GeoServer instance
+        If deep_copy is True, copy all feature types and the corresponding layers
+        """
+        datastore, status_code = self.src_instance.get_pg_datastore(
+            workspace_name, datastore_name
+        )
+        if isinstance(datastore, str):
+            return datastore, status_code
+        new_ds, new_ds_status_code = self.dst_instance.create_pg_datastore(
+            workspace_name, datastore
+        )
+        if self.not_ok(new_ds_status_code):
+            return new_ds, new_ds_status_code
+        if deep_copy:
+            self.copy_feature_types(workspace_name, datastore_name, copy_layers=True)
+        return new_ds, new_ds_status_code
+
+    def copy_feature_types(
+        self, workspace_name: str, datastore_name: str, copy_layers: bool = False
+    ) -> tuple[str, int]:
+        """
+        Copy all feature types in a datastore from source to destination GeoServer instance
+        """
+        feature_types, status_code = self.src_instance.get_feature_types(
+            workspace_name, datastore_name
+        )
+        if isinstance(feature_types, str):
+            return feature_types, status_code
+        for feature_type in feature_types.aslist():
+            content, status_code = self.copy_feature_type(
+                workspace_name, datastore_name, feature_type["name"]
+            )
+            if self.not_ok(status_code):
+                return content, status_code
+            if copy_layers:
+                content, status_code = self.copy_layer(
+                    workspace_name, feature_type["name"]
+                )
+                if self.not_ok(status_code):
+                    return content, status_code
+        return content, status_code
+
+    def copy_feature_type(
+        self, workspace_name: str, datastore_name: str, feature_type_name: str
+    ) -> tuple[str, int]:
+        """
+        Copy a feature type from source to destination GeoServer instance
+        """
+        feature_type, status_code = self.src_instance.get_feature_type(
+            workspace_name, datastore_name, feature_type_name
+        )
+        if isinstance(feature_type, str):
+            return feature_type, status_code
+        return self.dst_instance.create_feature_type(feature_type)
+
+    def copy_layer(
+        self, workspace_name: str, feature_type_name: str
+    ) -> tuple[str, int]:
+        """
+        Copy a layer from source to destination GeoServer instance
+        """
+        layer, status_code = self.src_instance.get_layer(
+            workspace_name, feature_type_name
+        )
+        if isinstance(layer, str):
+            return layer, status_code
+        return self.dst_instance.update_layer(layer, workspace_name)
+
     def copy_styles(
         self, workspace_name: str | None = None, include_images: bool = True
     ) -> tuple[str, int]:

geoservercloud/models/datastores.py

@@ -2,18 +2,18 @@
 
 
 class DataStores(ListModel):
-    def __init__(self, datastores: list[dict[str, str]] = []) -> None:
-        self._datastores: list[dict[str, str]] = datastores
+    def __init__(self, datastores: list[str] = []) -> None:
+        self._datastores: list[str] = datastores
 
     @classmethod
     def from_get_response_payload(cls, content: dict):
-        datastores: str | dict = content["dataStores"]
+        datastores: dict | str = content["dataStores"]
         if not datastores:
             return cls()
-        return cls(datastores["dataStore"])  # type: ignore
+        return cls([ds["name"] for ds in datastores["dataStore"]])  # type: ignore
 
     def __repr__(self) -> str:
-        return str(self._datastores)
+        return str([{"name": ds} for ds in self._datastores])
 
-    def aslist(self) -> list[dict[str, str]]:
+    def aslist(self) -> list[str]:
         return self._datastores

geoservercloud/services/restservice.py

@@ -383,6 +383,14 @@ def create_style(
             response = self.rest_client.put(resource_path, data=style, headers=headers)
         return response.content.decode(), response.status_code
 
+    def get_layer(
+        self, workspace_name: str, layer_name: str
+    ) -> tuple[Layer | str, int]:
+        response: Response = self.rest_client.get(
+            self.rest_endpoints.workspace_layer(workspace_name, layer_name)
+        )
+        return self.deserialize_response(response, Layer)
+
     def update_layer(self, layer: Layer, workspace_name: str) -> tuple[str, int]:
         response: Response = self.rest_client.put(
             self.rest_endpoints.workspace_layer(workspace_name, layer.name),

tests/models/test_datastores.py

@@ -4,33 +4,28 @@
 
 
 @fixture(scope="module")
-def mock_datastore():
-    return {
-        "name": "DataStore1",
-        "href": "http://example.com/ds1",
-    }
-
-
-@fixture(scope="module")
-def mock_response(mock_datastore):
+def mock_response():
     return {
         "dataStores": {
-            "dataStore": [mock_datastore],
+            "dataStore": [
+                {
+                    "name": "DataStore1",
+                    "href": "http://example.com/ds1",
+                },
+                {
+                    "name": "DataStore2",
+                    "href": "http://example.com/ds2",
+                },
+            ],
         }
     }
 
 
-def test_datastores_initialization(mock_datastore):
-    ds = DataStores([mock_datastore])
-
-    assert ds.aslist() == [mock_datastore]
-
-
-def test_datastores_from_get_response_payload(mock_datastore, mock_response):
+def test_datastores_from_get_response_payload(mock_response):
 
     ds = DataStores.from_get_response_payload(mock_response)
 
-    assert ds.aslist() == [mock_datastore]
+    assert ds.aslist() == ["DataStore1", "DataStore2"]
 
 
 def test_datastores_from_get_response_payload_empty():
@@ -41,9 +36,9 @@ def test_datastores_from_get_response_payload_empty():
     assert ds.aslist() == []
 
 
-def test_datastores_repr(mock_datastore):
-    ds = DataStores([mock_datastore])
+def test_datastores_repr():
+    ds = DataStores(["DataStore1", "DataStore2"])
 
-    expected_repr = "[{'name': 'DataStore1', 'href': 'http://example.com/ds1'}]"
+    expected_repr = "[{'name': 'DataStore1'}, {'name': 'DataStore2'}]"
 
     assert repr(ds) == expected_repr

tests/test_datastore.py

@@ -132,7 +132,7 @@ def test_get_datastores(
         )
 
         datastores, status_code = geoserver.get_datastores(workspace_name=WORKSPACE)
-        assert datastores == datastores_get_response["dataStores"]["dataStore"]
+        assert datastores == ["test_store"]
         assert status_code == 200