Migrate logic to RestService layer

And adapt / extend existing data classes

geoservercloud/models/__init__.py

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

geoservercloud/models/common.py

@@ -1,11 +1,51 @@
 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 ReferencedObjectModel(BaseModel):
+    def __init__(self, name: str, href: str | None = None):
+        self.name = name
+        self.href = href
+
+    @classmethod
+    def from_get_response_payload(cls, content: dict):
+        cls.name = content["name"]
+        cls.href = content["href"]
+
+    def asdict(self) -> dict[str, str]:
+        content = {"name": self.name}
+        if self.href:
+            content["href"] = self.href
+        return content
 
 
 class KeyDollarListDict(dict):
+
+    key_prefix: str = "@key"
+    value_prefix: str = "$"
+
     def __init__(
         self,
         input_list: list | None = None,
@@ -14,13 +54,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,82 @@
 import json
-import logging
+from typing import Any
 
-from requests.models import Response
+from geoservercloud.models import EntityModel, KeyDollarListDict, ReferencedObjectModel
 
-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: ReferencedObjectModel = ReferencedObjectModel(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()
-                },
-            }
+    @property
+    def workspace_name(self) -> str:
+        return self.workspace.name
+
+    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 put_payload(self):
-        payload = self.post_payload()
-        return payload
+    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) -> 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