Add new service layer and use it for workspace, datastore, WMS, WFS, …

…WMTS

geoservercloud/models/__init__.py

@@ -1,4 +1,4 @@
-from .common import I18N, KeyDollarListDict
+from .common import BaseModel, EntityModel, ListModel, I18N, KeyDollarListDict
 from .datastore import PostGisDataStore
 from .datastores import DataStores
 from .featuretype import FeatureType
@@ -9,8 +9,11 @@
 from .workspaces import Workspaces
 
 __all__ = [
+    "BaseModel",
     "DataStores",
+    "EntityModel",
     "KeyDollarListDict",
+    "ListModel",
     "FeatureType",
     "FeatureTypes",
     "I18N",

geoservercloud/models/common.py

@@ -1,11 +1,34 @@
 import json
-import logging
 from typing import Any
 
-log = logging.getLogger()
+
+class BaseModel:
+    @classmethod
+    def from_get_response_payload(cls, content: dict):
+        raise NotImplementedError
+
+
+class EntityModel(BaseModel):
+    def asdict(self) -> dict[str, Any]:
+        raise NotImplementedError
+
+    def post_payload(self) -> dict[str, Any]:
+        raise NotImplementedError
+
+    def put_payload(self) -> dict[str, Any]:
+        raise NotImplementedError
+
+
+class ListModel(BaseModel):
+    def aslist(self) -> list:
+        raise NotImplementedError
 
 
 class KeyDollarListDict(dict):
+
+    key_prefix: str = "@key"
+    value_prefix: str = "$"
+
     def __init__(
         self,
         input_list: list | None = None,
@@ -14,13 +37,10 @@ def __init__(
         **kwargs
     ):
         super().__init__(*args, **kwargs)
-        self.key_prefix = "@key"
-        self.value_prefix = "$"
         if input_list:
             self.deserialize(input_list)
         if input_dict:
             self.update(input_dict)
-        log.debug(self)
 
     def deserialize(self, input_list: list):
         for item in input_list:

geoservercloud/models/datastore.py

@@ -1,74 +1,78 @@
 import json
-import logging
+from typing import Any
 
-from requests.models import Response
+from geoservercloud.models import EntityModel, KeyDollarListDict
 
-from . import KeyDollarListDict
-
-log = logging.getLogger()
-
-
-class PostGisDataStore:
 
+class PostGisDataStore(EntityModel):
     def __init__(
         self,
         workspace_name: str,
-        data_store_name: str,
+        name: str,
         connection_parameters: dict,
-        data_store_type: str = "PostGIS",
+        type: str = "PostGIS",
         enabled: bool = True,
         description: str | None = None,
+        default: bool | None = None,
+        disable_on_conn_failure: bool | None = None,
     ) -> None:
-        self.workspace_name = workspace_name
-        self.data_store_name = data_store_name
+        self.workspace_name: str = workspace_name
+        self._name: str = name
         self.connection_parameters = KeyDollarListDict(input_dict=connection_parameters)
-        self.data_store_type = data_store_type
-        self.description = description
-        self.enabled = enabled
+        self.type: str = type
+        self.description: str | None = description
+        self.enabled: bool = enabled
+        self._default: bool | None = default
+        self.disable_on_conn_failure: bool | None = disable_on_conn_failure
 
     @property
-    def name(self):
-        return self.data_store_name
+    def name(self) -> str:
+        return self._name
 
-    def post_payload(self):
-        payload = {
-            "dataStore": {
-                "name": self.data_store_name,
-                "type": self.data_store_type,
-                "connectionParameters": {
-                    "entry": self.connection_parameters.serialize()
-                },
-            }
+    def asdict(self) -> dict[str, Any]:
+        content: dict[str, Any] = {
+            "name": self._name,
+            "type": self.type,
+            "connectionParameters": {"entry": dict(self.connection_parameters)},
+            "workspace": self.workspace_name,
         }
         if self.description:
-            payload["dataStore"]["description"] = self.description
+            content["description"] = self.description
         if self.enabled:
-            payload["dataStore"]["enabled"] = self.enabled
-        return payload
+            content["enabled"] = self.enabled
+        if self._default is not None:
+            content["_default"] = self._default
+        if self.disable_on_conn_failure is not None:
+            content["disableOnConnFailure"] = self.disable_on_conn_failure
+        return content
+
+    def post_payload(self) -> dict[str, Any]:
+        content = self.asdict()
+        content["connectionParameters"] = {
+            "entry": self.connection_parameters.serialize()
+        }
+        content["workspace"] = {"name": self.workspace_name}
+        return {"dataStore": content}
 
-    def put_payload(self):
-        payload = self.post_payload()
-        return payload
+    def put_payload(self) -> dict[str, Any]:
+        return self.post_payload()
 
     @classmethod
-    def from_dict(cls, content: dict):
-        connection_parameters = cls.parse_connection_parameters(content)
+    def from_get_response_payload(cls, content: dict):
+        data_store = content["dataStore"]
+        connection_parameters = KeyDollarListDict(
+            input_list=data_store["connectionParameters"]["entry"]
+        )
         return cls(
-            content.get("dataStore", {}).get("workspace", {}).get("name", None),
-            content.get("dataStore", {}).get("name", None),
+            data_store["workspace"]["name"],
+            data_store["name"],
             connection_parameters,
-            content.get("dataStore", {}).get("type", "PostGIS"),
-            content.get("dataStore", {}).get("enabled", True),
-            content.get("dataStore", {}).get("description", None),
-        )
-
-    @classmethod
-    def parse_connection_parameters(cls, content):
-        return KeyDollarListDict(
-            content.get("dataStore", {})
-            .get("connectionParameters", {})
-            .get("entry", [])
+            data_store.get("type", "PostGIS"),
+            data_store.get("enabled", True),
+            data_store.get("description", None),
+            data_store.get("_default", None),
+            data_store.get("disableOnConnFailure", None),
         )
 
-    def __repr__(self):
+    def __repr__(self) -> str:
         return json.dumps(self.put_payload(), indent=4)

geoservercloud/models/datastores.py

@@ -1,32 +1,19 @@
-import logging
+from geoservercloud.models import ListModel
 
-from requests.models import Response
 
-log = logging.getLogger()
-
-
-class DataStores:
-
-    def __init__(self, workspace_name: str, datastores: list[str] = []) -> None:
-        self.workspace_name = workspace_name
-        self._datastores = datastores
-
-    @property
-    def datastores(self):
-        return self._datastores
+class DataStores(ListModel):
+    def __init__(self, datastores: list[dict[str, str]] = []) -> None:
+        self._datastores: list[dict[str, str]] = datastores
 
     @classmethod
-    def from_dict(cls, content: dict):
-        datastores = []
-        workspace_name = (
-            content.get("dataStores", {}).get("workspace", {}).get("name", None)
-        )
+    def from_get_response_payload(cls, content: dict):
+        datastores: str | dict = content["dataStores"]
+        if not datastores:
+            return cls()
+        return cls(datastores["dataStore"])  # type: ignore
 
-        for store in content.get("dataStores", {}).get("dataStore", []):
-            datastores.append(store["name"])
-            for data_store_name in datastores:
-                log.debug(f"Name: {data_store_name}")
-        return cls(workspace_name, datastores)
+    def __repr__(self) -> str:
+        return str(self._datastores)
 
-    def __repr__(self):
-        return str(self.datastores)
+    def aslist(self) -> list[dict[str, str]]:
+        return self._datastores

geoservercloud/models/featuretype.py

@@ -1,12 +1,10 @@
 import json
 
-from requests.models import Response
-
-from geoservercloud.models import I18N
+from geoservercloud.models import EntityModel, I18N
 
 
 # TODO: import more default values from Templates
-class FeatureType:
+class FeatureType(EntityModel):
     def __init__(
         self,
         namespace_name: str,

geoservercloud/models/featuretypes.py

@@ -1,10 +1,9 @@
 import json
 
-from requests.models import Response
+from geoservercloud.models import ListModel
 
 
-class FeatureTypes:
-
+class FeatureTypes(ListModel):
     def __init__(self, featuretypes: list = []) -> None:
         self._featuretypes = featuretypes
 

geoservercloud/models/style.py

@@ -1,11 +1,11 @@
 import json
 
 import xmltodict
-from requests.models import Response
 
+from geoservercloud.models import EntityModel
 
-class Style:
 
+class Style(EntityModel):
     def __init__(
         self,
         name: str,

geoservercloud/models/styles.py

@@ -1,8 +1,7 @@
-from requests.models import Response
+from geoservercloud.models import ListModel
 
 
-class Styles:
-
+class Styles(ListModel):
     def __init__(self, styles: list[str], workspace: str | None = None) -> None:
         self._workspace = workspace
         self._styles = styles

geoservercloud/models/workspace.py

@@ -1,31 +1,31 @@
 import json
-import logging
+from typing import Any
 
-from requests.models import Response
+from geoservercloud.models import EntityModel
 
-log = logging.getLogger()
-
-
-class Workspace:
 
+class Workspace(EntityModel):
     def __init__(self, name: str, isolated: bool = False) -> None:
-        self.name = name
-        self.isolated = isolated
+        self.name: str = name
+        self.isolated: bool = isolated
+
+    def asdict(self) -> dict[str, Any]:
+        return {
+            "name": self.name,
+            "isolated": self.isolated,
+        }
 
-    def put_payload(self):
-        payload = {"workspace": {"name": self.name}}
-        if self.isolated:
-            payload["workspace"]["isolated"] = self.isolated
-        return payload
+    def put_payload(self) -> dict[str, dict[str, Any]]:
+        return {"workspace": self.asdict()}
 
-    def post_payload(self):
+    def post_payload(self) -> dict[str, dict[str, str]]:
         return self.put_payload()
 
     @classmethod
-    def from_dict(cls, content: dict):
+    def from_get_response_payload(cls, content: dict):
         return cls(
-            content.get("workspace", {}).get("name", None),
-            content.get("workspace", {}).get("isolated", False),
+            content["workspace"]["name"],
+            content["workspace"]["isolated"],
         )
 
     def __repr__(self):

geoservercloud/models/workspaces.py

@@ -1,33 +1,25 @@
-import logging
+from geoservercloud.models import ListModel
 
-from requests.models import Response
 
-log = logging.getLogger()
-
-
-class Workspaces:
-
-    def __init__(self, workspaces: list = []) -> None:
+class Workspaces(ListModel):
+    def __init__(self, workspaces: list[dict[str, str]] = []) -> None:
         self._workspaces = workspaces
 
-    def find(self, workspace_name: str):
-        return self.workspaces.get(workspace_name, None)
-
-    @property
-    def workspaces(self):
-        return self._workspaces
+    def find(self, workspace_name: str) -> dict[str, str] | None:
+        for ws in self._workspaces:
+            if ws["name"] == workspace_name:
+                return ws
 
     @classmethod
-    def from_dict(cls, content: dict):
+    def from_get_response_payload(cls, content: dict):
 
-        workspaces = []
-        # Map the response to a list of Workspace instances
-        for ws in content.get("workspaces", {}).get("workspace", []):
-            workspaces.append(ws["name"])
+        workspaces: str | dict = content["workspaces"]
+        if not workspaces:
+            return cls()
+        return cls(workspaces["workspace"])  # type: ignore
 
-        # Now 'workspaces' is a list of Workspace instances
-        log.debug("Parsed Workspaces:")
-        for workspace in workspaces:
-            log.debug(f"Name: {workspace}")
+    def __repr__(self) -> str:
+        return str(self._workspaces)
 
-        return cls(workspaces)
+    def aslist(self) -> list[dict[str, str]]:
+        return self._workspaces