Add styles (partial), fix datastore

geoservercloud/geoservercloud.py

@@ -14,6 +14,8 @@
     DataStores,
     KeyDollarListDict,
     PostGisDataStore,
+    Style,
+    Styles,
     Workspace,
     Workspaces,
 )
@@ -169,6 +171,17 @@ def get_datastores(self, workspace_name: str) -> dict[str, Any]:
         response = self.get_request(self.rest_endpoints.datastores(workspace_name))
         return DataStores.from_response(response).datastores
 
+    def get_postgis_datastore(
+        self, workspace_name: str, datastore_name: str
+    ) -> dict[str, Any]:
+        """
+        Get a specific datastore
+        """
+        response = self.get_request(
+            self.rest_endpoints.datastore(workspace_name, datastore_name)
+        )
+        return PostGisDataStore.from_response(response)
+
     def create_pg_datastore(
         self,
         workspace_name: str,
@@ -444,6 +457,32 @@ def publish_gwc_layer(
             json=payload,
         )
 
+    def get_styles(self, workspace_name: str | None = None) -> dict[str, Any]:
+        """
+        Get all styles for a given workspace
+        """
+        path = (
+            self.rest_endpoints.styles()
+            if not workspace_name
+            else self.rest_endpoints.workspace_styles(workspace_name)
+        )
+        styles = Styles.from_response(self.get_request(path)).styles
+        return styles
+
+    def get_style(
+        self, style: str, workspace_name: str | None = None
+    ) -> dict[str, Any]:
+        """
+        Get a specific style
+        """
+        path = (
+            self.rest_endpoints.style(style)
+            if not workspace_name
+            else self.rest_endpoints.workspace_style(workspace_name, style)
+        )
+        return Style.from_response(self.get_request(path))
+
+    # TODO: add a create_style method that takes a Style object as input
     def create_style_from_file(
         self,
         style: str,

geoservercloud/models/__init__.py

@@ -1,13 +1,17 @@
 from .common import KeyDollarListDict
 from .dataStore import PostGisDataStore
 from .dataStores import DataStores
+from .style import Style
+from .styles import Styles
 from .workspace import Workspace
 from .workspaces import Workspaces
 
 __all__ = [
     "DataStores",
     "KeyDollarListDict",
     "PostGisDataStore",
+    "Style",
+    "Styles",
     "Workspaces",
     "Workspace",
 ]

geoservercloud/models/dataStore.py

@@ -1,3 +1,4 @@
+import json
 import logging
 
 from . import KeyDollarListDict
@@ -49,7 +50,8 @@ def put_payload(self):
 
     @classmethod
     def from_response(cls, response):
-
+        if response.status_code == 404:
+            return None
         json_data = response.json()
         connection_parameters = cls.parse_connection_parameters(json_data)
         return cls(
@@ -68,3 +70,6 @@ def parse_connection_parameters(cls, json_data):
             .get("connectionParameters", {})
             .get("entry", [])
         )
+
+    def __repr__(self):
+        return json.dumps(self.put_payload(), indent=4)

geoservercloud/models/dataStores.py

@@ -1,8 +1,5 @@
 import logging
 
-import jsonschema
-import requests
-
 log = logging.getLogger()
 
 

geoservercloud/models/style.py

@@ -0,0 +1,119 @@
+import json
+
+import xmltodict
+
+
+class Style:
+
+    def __init__(
+        self,
+        name: str,
+        workspace: str | None = None,
+        format: str | None = "sld",
+        language_version: dict | None = {"version": "1.0.0"},
+        filename: str | None = None,
+        date_created: str | None = None,
+        date_modified: str | None = None,
+        legend_url: str | None = None,
+        legend_format: str | None = None,
+        legend_width: str | None = None,
+        legend_height: str | None = None,
+    ) -> None:
+        self._workspace = workspace
+        self._name = name
+        self._format = format
+        self._language_version = language_version
+        self._filename = filename
+        self._date_created = date_created
+        self._date_modified = date_modified
+        self.create_legend(legend_url, legend_format, legend_width, legend_height)
+
+    # create one property for each attribute
+    @property
+    def workspace(self):
+        return self._workspace
+
+    @property
+    def name(self):
+        return self._name
+
+    @property
+    def format(self):
+        return self._format
+
+    @property
+    def language_version(self):
+        return self._language_version
+
+    @property
+    def filename(self):
+        return self._filename
+
+    @property
+    def date_created(self):
+        return self._date_created
+
+    @property
+    def date_modified(self):
+        return self._date_modified
+
+    @property
+    def legend(self):
+        return self._legend
+
+    def create_legend(self, url, image_format, width, height):
+        if any([url, image_format, width, height]):
+            self._legend = {}
+            if url:
+                self.legend["onlineResource"] = url
+            if image_format:
+                self.legend["format"] = image_format
+            if width:
+                self.legend["width"] = width
+            if height:
+                self.legend["height"] = height
+        else:
+            self._legend = None
+
+    def put_payload(self):
+        payload = {
+            "style": {
+                "name": self.name,
+                "format": self.format,
+                "languageVersion": self.language_version,
+                "filename": self.filename,
+            }
+        }
+        if self.legend:
+            payload["style"]["legend"] = self.legend
+        return payload
+
+    def post_payload(self):
+        return self.put_payload()
+
+    @classmethod
+    def from_response(cls, response):
+        json_data = response.json()
+        style_data = json_data.get("style", {})
+        return cls(
+            workspace=style_data.get("workspace"),
+            name=style_data.get("name"),
+            format=style_data.get("format"),
+            language_version=style_data.get("languageVersion", None),
+            filename=style_data.get("filename"),
+            date_created=style_data.get("dateCreated"),
+            date_modified=style_data.get("dateModified"),
+            legend_url=style_data.get("legend", {}).get("onlineResource"),
+            legend_format=style_data.get("legend", {}).get("format"),
+            legend_width=style_data.get("legend", {}).get("width"),
+            legend_height=style_data.get("legend", {}).get("height"),
+        )
+
+    def xml_post_payload(self):
+        return xmltodict.unparse(self.post_payload()).split("\n", 1)[1]
+
+    def xml_put_payload(self):
+        return xmltodict.unparse(self.put_payload()).split("\n", 1)[1]
+
+    def __repr__(self):
+        return json.dumps(self.put_payload(), indent=4)

geoservercloud/models/styles.py

@@ -0,0 +1,28 @@
+class Styles:
+
+    def __init__(self, styles: list[str], workspace: str | None = None) -> None:
+        self._workspace = workspace
+        self._styles = styles
+
+    @property
+    def workspace(self):
+        return self._workspace
+
+    @property
+    def styles(self):
+        return self._styles
+
+    @classmethod
+    def from_response(cls, response):
+        json_data = response.json()
+        styles = []
+        try:
+            workspace = json_data["styles"]["workspace"]
+        except KeyError:
+            workspace = None
+        try:
+            for style in json_data.get("styles", {}).get("style", []):
+                styles.append(style["name"])
+        except AttributeError:
+            styles = []
+        return cls(styles, workspace)

geoservercloud/models/workspace.py

@@ -1,8 +1,6 @@
+import json
 import logging
 
-import jsonschema
-import requests
-
 log = logging.getLogger()
 
 
@@ -29,3 +27,6 @@ def from_response(cls, response):
             json_data.get("workspace", {}).get("isolated", False),
         )
         return cls(json_data.get("workspace", {}).get("name", None))
+
+    def __repr__(self):
+        return json.dumps(self.put_payload(), indent=4)

geoservercloud/models/workspaces.py

@@ -1,8 +1,5 @@
 import logging
 
-import jsonschema
-import requests
-
 log = logging.getLogger()
 
 

tests/test_datastore.py

@@ -1,3 +1,4 @@
+import json
 from collections.abc import Generator
 from typing import Any
 
@@ -6,6 +7,7 @@
 from responses import matchers
 
 from geoservercloud.geoservercloud import GeoServerCloud
+from geoservercloud.models import PostGisDataStore  # Ensure this import is correct
 from tests.conftest import GEOSERVER_URL
 
 WORKSPACE = "test_workspace"
@@ -105,6 +107,40 @@ def test_get_datastores(
         assert datastores == ["test_store"]
 
 
+# Test the get_postgis_datastore method with a valid response
+def test_get_postgis_datastore_valid(
+    geoserver: GeoServerCloud, pg_payload: dict[str, dict[str, Any]]
+) -> None:
+    with responses.RequestsMock() as rsps:
+        rsps.get(
+            url=f"{GEOSERVER_URL}/rest/workspaces/{WORKSPACE}/datastores/{STORE}.json",
+            json=pg_payload,
+            status=200,
+        )
+        result = geoserver.get_postgis_datastore(WORKSPACE, STORE)
+        assert len(rsps.calls) == 1
+        # FIXME: I think the geoserver rest endpoint is wrong, might be a problem with the conftest.py stuff.
+        # assert rsps.calls[0].request.url == geoserver.rest_endpoints.datastore(WORKSPACE, STORE)
+        assert json.loads(str(result)) == pg_payload
+
+
+# Test the get_postgis_datastore method with a 404 error
+def test_get_postgis_datastore_not_found(geoserver: GeoServerCloud) -> None:
+    datastore_name = "non_existing_datastore"
+    with responses.RequestsMock() as rsps:
+        rsps.get(
+            url=f"{GEOSERVER_URL}/rest/workspaces/{WORKSPACE}/datastores/{datastore_name}.json",
+            json={"error": "Datastore not found"},
+            status=404,
+        )
+
+        not_existing_datastore = geoserver.get_postgis_datastore(
+            WORKSPACE, datastore_name
+        )
+        assert len(rsps.calls) == 1
+        assert not_existing_datastore is None
+
+
 def test_create_pg_datastore(
     geoserver: GeoServerCloud, pg_payload: dict[str, dict[str, Any]]
 ) -> None: