wip

geoservercloud/geoservercloud.py

@@ -254,7 +254,7 @@ def create_wmts_store(
 
     def get_feature_types(
         self, workspace_name: str, datastore_name: str
-    ) -> tuple[list[dict[str, Any]] | str, int]:
+    ) -> tuple[list[str] | str, int]:
         """
         Get all feature types for a given workspace and datastore
         """

geoservercloud/geoservercloudsync.py

@@ -49,3 +49,55 @@ def copy_workspace(self, workspace_name: str) -> tuple[str, int]:
         if isinstance(workspace, str):
             return workspace, status_code
         return self.dst_instance.create_workspace(workspace)
+
+    def copy_pg_datastore(
+        self, workspace_name: str, datastore_name: str
+    ) -> tuple[str, int]:
+        """
+        Shallow copy a datastore from source to destination GeoServer instance
+        """
+        datastore, status_code = self.src_instance.get_pg_datastore(
+            workspace_name, datastore_name
+        )
+        if isinstance(datastore, str):
+            return datastore, status_code
+        return self.dst_instance.create_pg_datastore(workspace_name, datastore)
+
+    def copy_feature_types_in_datastore(
+        self, workspace_name: str, datastore_name: str
+    ) -> list[tuple[str, int]] | 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
+        return [
+            self.copy_feature_type(workspace_name, datastore_name, feature_type_name)
+            for feature_type_name in feature_types.aslist()
+        ]
+
+    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_style(self, workspace_name: str, style_name: str) -> tuple[str, int]:
+        """
+        Copy a style from source to destination GeoServer instance
+        curl -vf -XPUT -u testadmin:testadmin -H "Content-type:application/vnd.ogc.sld+xml" -T "metro_canton_situation_polygon.sld"
+        """
+        style, status_code = self.src_instance.get_style(style_name, workspace_name)
+        if isinstance(style, str):
+            return style, status_code
+        return self.dst_instance.create_style(style, workspace_name)

geoservercloud/models/featuretypes.py

@@ -5,13 +5,16 @@
 
 class FeatureTypes(ListModel):
     def __init__(self, featuretypes: list = []) -> None:
-        self._featuretypes = featuretypes
+        self._featuretypes: list[str] = featuretypes
 
     @classmethod
     def from_get_response_payload(cls, content: dict):
-        return cls(content["featureTypes"]["featureType"])
+        feature_types: str | dict = content["featureTypes"]
+        if not feature_types:
+            return cls([])
+        return cls([feature_type["name"] for feature_type in feature_types["featureType"]])  # type: ignore
 
-    def aslist(self) -> list[dict[str, str]]:
+    def aslist(self) -> list[str]:
         return self._featuretypes
 
     def __repr__(self):

geoservercloud/services/restservice.py

@@ -347,17 +347,23 @@ def create_style(
         path = (
             self.rest_endpoints.styles()
             if not workspace_name
-            else self.rest_endpoints.workspace_styles(workspace_name)
+            else self.rest_endpoints.workspace_styles(workspace_name, format="xml")
         )
         resource_path = (
             self.rest_endpoints.style(style.name)
             if not workspace_name
             else self.rest_endpoints.workspace_style(workspace_name, style.name)
         )
+        headers = {"Content-Type": "text/xml"}
+        print(style.xml_post_payload())
         if not self.resource_exists(resource_path):
-            response: Response = self.rest_client.post(path, json=style.post_payload())
+            response: Response = self.rest_client.post(
+                path, data=style.xml_post_payload().encode("utf-8"), headers=headers
+            )
         else:
-            response = self.rest_client.put(resource_path, json=style.put_payload())
+            response = self.rest_client.put(
+                resource_path, data=style.xml_put_payload().encode(), headers=headers
+            )
         return response.content.decode(), response.status_code
 
     def create_style_from_file(
@@ -636,8 +642,10 @@ class RestEndpoints:
         def __init__(self, base_url: str = "/rest") -> None:
             self.base_url: str = base_url
 
-        def styles(self) -> str:
-            return f"{self.base_url}/styles.json"
+        def styles(self, format="json") -> str:
+            if format == "json":
+                return f"{self.base_url}/styles.json"
+            return f"{self.base_url}/styles"
 
         def style(self, style_name: str) -> str:
             return f"{self.base_url}/styles/{style_name}.json"
@@ -648,8 +656,10 @@ def workspaces(self) -> str:
         def workspace(self, workspace_name: str) -> str:
             return f"{self.base_url}/workspaces/{workspace_name}.json"
 
-        def workspace_styles(self, workspace_name: str) -> str:
-            return f"{self.base_url}/workspaces/{workspace_name}/styles.json"
+        def workspace_styles(self, workspace_name: str, format="json") -> str:
+            if format == "json":
+                return f"{self.base_url}/workspaces/{workspace_name}/styles.json"
+            return f"{self.base_url}/workspaces/{workspace_name}/styles"
 
         def workspace_style(self, workspace_name: str, style_name: str) -> str:
             return (

tests/models/test_featuretypes.py

@@ -17,9 +17,7 @@ def test_featuretypes_from_dict_valid():
 
     feature_types_instance = FeatureTypes.from_get_response_payload(mock_response)
 
-    assert (
-        feature_types_instance.aslist() == mock_response["featureTypes"]["featureType"]
-    )
+    assert feature_types_instance.aslist() == ["feature1", "feature2"]
 
 
 def test_featuretypes_repr():

tests/test_feature_type.py

@@ -160,9 +160,7 @@ def test_get_feature_types(
             workspace_name=WORKSPACE, datastore_name=STORE
         )
 
-        assert (
-            content == feature_types_get_response_payload["featureTypes"]["featureType"]
-        )
+        assert content == ["featuretype1", "featuretype2", "featuretype3"]
         assert code == 200