Add data model for Workspaces

geoservercloud/models/__init__.py

@@ -0,0 +1,5 @@
+from .common import KeyDollarListDict
+from .workspace import Workspace
+from .workspaces import Workspaces
+
+__all__ = ["KeyDollarListDict", "Workspaces"]

geoservercloud/models/common.py

@@ -0,0 +1,39 @@
+import json
+import logging
+
+log = logging.getLogger()
+
+
+class KeyDollarListDict(dict):
+    def __init__(self, input_list=None, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.key_prefix = "@key"
+        self.value_prefix = "$"
+        if input_list:
+            self.deserialize(input_list)
+        log.debug(self)
+
+    def deserialize(self, input_list):
+        for item in input_list:
+            key = item[self.key_prefix]
+            if self.value_prefix in item:
+                value = item[self.value_prefix]
+            else:
+                value = None
+            super().__setitem__(key, value)
+
+    def serialize(self):
+        return [
+            {self.key_prefix: key, self.value_prefix: value}
+            for key, value in self.items()
+        ]
+
+    def __repr__(self) -> str:
+        return str(self.serialize())
+
+    def __str__(self):
+        return json.dumps(self.serialize())
+
+    # def update(self, other: dict):
+    #     for key, value in other.items():
+    #         super().__setitem__(key, value)

geoservercloud/models/workspace.py

@@ -0,0 +1,72 @@
+import logging
+
+import jsonschema
+import requests
+
+log = logging.getLogger()
+
+
+class Workspace:
+
+    def __init__(self, name, isolated: bool = False) -> None:
+        self.name = name
+        self.isolated = isolated
+
+    _responseSchema = {
+        "$schema": "http://json-schema.org/draft-07/schema#",
+        "type": "object",
+        "properties": {
+            "workspace": {
+                "type": "object",
+                "properties": {
+                    "name": {"type": "string"},
+                    "isolated": {"type": "boolean"},
+                    "dateCreated": {"type": "string", "format": "date-time"},
+                    "dataStores": {"type": "string", "format": "uri"},
+                    "coverageStores": {"type": "string", "format": "uri"},
+                    "wmsStores": {"type": "string", "format": "uri"},
+                    "wmtsStores": {"type": "string", "format": "uri"},
+                },
+                "required": [
+                    "name",
+                    "isolated",
+                    "dataStores",
+                    "coverageStores",
+                    "wmsStores",
+                    "wmtsStores",
+                ],
+            }
+        },
+        "required": ["workspace"],
+    }
+
+    def put_payload(self):
+        payload = {"workspace": {"name": self.name}}
+        if self.isolated:
+            payload["workspace"]["isolated"] = self.isolated
+        return payload
+
+    def post_payload(self):
+        return self.put_payload()
+
+    @classmethod
+    def from_response(cls, response):
+        json_data = response.json()
+        cls.validate(json_data)
+        return cls(
+            json_data.get("workspace", {}).get("name", None),
+            json_data.get("workspace", {}).get("isolated", False),
+        )
+
+    @classmethod
+    def validate(cls, response):
+        try:
+            jsonschema.validate(response, cls.responseSchema())
+        except jsonschema.exceptions.ValidationError as err:
+            print(err)
+            raise Exception("Invalid from workspace response")
+
+    @classmethod
+    def responseSchema(cls):
+        # Define the JSON schema here
+        return cls._responseSchema

geoservercloud/models/workspaces.py

@@ -0,0 +1,76 @@
+import logging
+
+import jsonschema
+import requests
+
+log = logging.getLogger()
+
+
+class Workspaces:
+
+    def __init__(self) -> None:
+        self._workspaces: dict[str, str] = {}
+
+    _response_schema = {
+        "$schema": "http://json-schema.org/draft-07/schema#",
+        "type": "object",
+        "properties": {
+            "workspaces": {
+                "type": "object",
+                "properties": {
+                    "workspace": {
+                        "type": "array",
+                        "items": {
+                            "type": "object",
+                            "properties": {
+                                "name": {"type": "string"},
+                                "href": {"type": "string", "format": "uri"},
+                            },
+                            "required": ["name", "href"],
+                        },
+                    }
+                },
+                "required": ["workspace"],
+            }
+        },
+        "required": ["workspaces"],
+    }
+
+    @property
+    def response_schema(self):
+        return self._response_schema
+
+    def validate(self, response):
+        try:
+            jsonschema.validate(response, self.response_schema)
+        except jsonschema.exceptions.ValidationError as err:
+            print(err)
+            return False
+        return True
+
+    def find(self, workspace_name):
+        return self.workspaces.get(workspace_name, None)
+
+    @property
+    def workspaces(self):
+        return self._workspaces
+
+    def parseResponse(self, response):
+
+        # Check if the request was successful (status code 200)
+        if response.status_code == 200:
+            # Parse the JSON response
+            json_data = response.json()
+            if not self.validate(json_data):
+                raise Exception("Invalid response from workspaces")
+
+            # Map the response to a list of Workspace instances
+            for ws in json_data.get("workspaces", {}).get("workspace", []):
+                self._workspaces[ws["name"]] = ws["href"]
+
+            # Now 'workspaces' is a list of Workspace instances
+            log.debug("Parsed Workspaces:")
+            for workspace, href in self._workspaces.items():
+                log.debug(f"Name: {workspace}, Href: {href}")
+        else:
+            log.error(f"Error: {response.status_code}")