Create a REST client service and add a first set of datamodels

geoservercloud/models/__init__.py

@@ -0,0 +1,22 @@
+from .common import I18N, KeyDollarListDict
+from .datastore import PostGisDataStore
+from .datastores import DataStores
+from .featuretype import FeatureType
+from .featuretypes import FeatureTypes
+from .style import Style
+from .styles import Styles
+from .workspace import Workspace
+from .workspaces import Workspaces
+
+__all__ = [
+    "DataStores",
+    "KeyDollarListDict",
+    "FeatureType",
+    "FeatureTypes",
+    "I18N",
+    "PostGisDataStore",
+    "Style",
+    "Styles",
+    "Workspaces",
+    "Workspace",
+]

geoservercloud/models/common.py

@@ -0,0 +1,104 @@
+import json
+import logging
+from typing import Any
+
+log = logging.getLogger()
+
+
+class KeyDollarListDict(dict):
+    def __init__(
+        self,
+        input_list: list | None = None,
+        input_dict: dict | None = None,
+        *args,
+        **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:
+            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):  # type: ignore
+        for key, value in other.items():
+            super().__setitem__(key, value)
+
+
+class I18N:
+    """
+    Geoserver handles internationalization with 2 possible (mutually exclusive) keys in the rest payload:
+    either:
+    - [key: string]
+    or
+    - [internationalKey: dictionary]
+
+    example:
+    a) as key: string we get {"title": "Test Title"}
+    b) as key: dict we get {"internationalTitle": {"en": "Test Title", "es": "Título de Prueba"}}
+
+    This class gives a layer of abstraction to handle both cases.
+
+    Usage:
+    Call the class by adding both possible keys in a tuple and the value.
+
+    Parameters:
+    keys: tuple[str, str] example : ("title", "internationalTitle")
+    value: str | dict example: "Test Title" | {"en": "Test Title", "es": "Título de Prueba"}
+
+    Example:
+
+    my_i18n = I18N(("title", "internationalTitle"), "Test Title")
+    """
+
+    def __init__(self, keys: tuple[str, Any], value: str | dict) -> None:
+        self._str_key = keys[0]
+        self._i18n_key = keys[1]
+        self._value = value
+        if isinstance(value, str):
+            self._content = {self.str_key: self._value}
+        elif isinstance(value, dict):
+            self._content = {self._i18n_key: self._value}
+        else:
+            raise ValueError("Invalid value type")
+
+    @property
+    def str_key(self):
+        return self._str_key
+
+    @property
+    def i18n_key(self):
+        return self._i18n_key
+
+    @property
+    def value(self):
+        return self._value
+
+    def asdict(self):
+        return self._content
+
+    def __repr__(self):
+        return json.dumps(self._content, indent=4)

geoservercloud/models/datastore.py

@@ -0,0 +1,74 @@
+import json
+import logging
+
+from requests.models import Response
+
+from . import KeyDollarListDict
+
+log = logging.getLogger()
+
+
+class PostGisDataStore:
+
+    def __init__(
+        self,
+        workspace_name: str,
+        data_store_name: str,
+        connection_parameters: dict,
+        data_store_type: str = "PostGIS",
+        enabled: bool = True,
+        description: str | None = None,
+    ) -> None:
+        self.workspace_name = workspace_name
+        self.data_store_name = data_store_name
+        self.connection_parameters = KeyDollarListDict(input_dict=connection_parameters)
+        self.data_store_type = data_store_type
+        self.description = description
+        self.enabled = enabled
+
+    @property
+    def name(self):
+        return self.data_store_name
+
+    def post_payload(self):
+        payload = {
+            "dataStore": {
+                "name": self.data_store_name,
+                "type": self.data_store_type,
+                "connectionParameters": {
+                    "entry": self.connection_parameters.serialize()
+                },
+            }
+        }
+        if self.description:
+            payload["dataStore"]["description"] = self.description
+        if self.enabled:
+            payload["dataStore"]["enabled"] = self.enabled
+        return payload
+
+    def put_payload(self):
+        payload = self.post_payload()
+        return payload
+
+    @classmethod
+    def from_dict(cls, content: dict):
+        connection_parameters = cls.parse_connection_parameters(content)
+        return cls(
+            content.get("dataStore", {}).get("workspace", {}).get("name", None),
+            content.get("dataStore", {}).get("name", None),
+            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", [])
+        )
+
+    def __repr__(self):
+        return json.dumps(self.put_payload(), indent=4)

geoservercloud/models/datastores.py

@@ -0,0 +1,32 @@
+import logging
+
+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
+
+    @classmethod
+    def from_dict(cls, content: dict):
+        datastores = []
+        workspace_name = (
+            content.get("dataStores", {}).get("workspace", {}).get("name", None)
+        )
+
+        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):
+        return str(self.datastores)

geoservercloud/models/featuretype.py

@@ -0,0 +1,133 @@
+import json
+
+from requests.models import Response
+
+from geoservercloud.models import I18N
+
+
+# TODO: import more default values from Templates
+class FeatureType:
+    def __init__(
+        self,
+        namespace_name: str,
+        name: str,
+        native_name: str,
+        srs: str = "EPSG:4326",
+        title: str | dict = "New Layer",
+        abstract: str | dict = "New Layer",
+        keywords={},
+        metadata_url=None,
+        metadata_type="TC211",
+        metadata_format="text/xml",
+        attributes: dict | None = None,
+    ) -> None:
+        self._namespace_name = namespace_name
+        self._name = name
+        self._title = I18N(("title", "internationalTitle"), title)
+        self._abstract = I18N(("abstract", "internationalAbstract"), abstract)
+        self._native_name = native_name
+        self._srs = srs
+        self._keywords = keywords
+        self._attributes = attributes
+        self.create_metadata_link(metadata_url, metadata_type, metadata_format)
+
+    @property
+    def namespace_name(self):
+        return self._namespace_name
+
+    @property
+    def name(self):
+        return self._name
+
+    @property
+    def title(self):
+        return self._title
+
+    @property
+    def abstract(self):
+        return self._abstract
+
+    @property
+    def native_name(self):
+        return self._native_name
+
+    @property
+    def srs(self):
+        return self._srs
+
+    @property
+    def keywords(self):
+        return self._keywords
+
+    @property
+    def metadataLink(self):
+        return self._metadataLink
+
+    @property
+    def attributes(self):
+        return self._attributes
+
+    def post_payload(self):
+        payload = {
+            "featureType": {
+                "name": self.name,
+                "nativeName": self.native_name,
+                "srs": self.srs,
+                "keywords": self.keywords,
+            }
+        }
+        payload["featureType"].update(self.title.asdict())
+        payload["featureType"].update(self.abstract.asdict())
+        if self.metadataLink != {}:
+            payload["featureType"]["metadataLinks"] = self.metadataLink
+        if self.attributes:
+            payload["featureType"]["attributes"] = self.attributes
+        return payload
+
+    def create_metadata_link(
+        self, metadata_url=None, metadata_type="TC211", metadata_format="text/xml"
+    ):
+        self._metadataLink = {}
+        if metadata_url:
+            self._metadataLink["metadataLink"] = {
+                "type": metadata_format,
+                "metadataType": metadata_type,
+                "content": metadata_url,
+            }
+
+    @classmethod
+    def from_dict(cls, content: dict):
+        try:
+            abstract = content["featureType"]["abstract"]
+        except KeyError:
+            abstract = content["featureType"]["internationalAbstract"]
+        try:
+            title = content["featureType"]["title"]
+        except KeyError:
+            title = content["featureType"]["internationalTitle"]
+
+        return cls(
+            namespace_name=content["featureType"]["namespace"]["name"],
+            name=content["featureType"]["name"],
+            native_name=content["featureType"]["nativeName"],
+            title=title,
+            abstract=abstract,
+            srs=content["featureType"]["srs"],
+            keywords=content["featureType"]["keywords"],
+            attributes=content["featureType"].get("attributes", None),
+            metadata_url=content["featureType"]
+            .get("metadataLinks", {})
+            .get("metadataLink", {})
+            .get("content"),
+            metadata_type=content["featureType"]
+            .get("metadataLinks", {})
+            .get("metadataLink", {})
+            .get("metadataType"),
+            metadata_format=content["featureType"]
+            .get("metadataLinks", {})
+            .get("metadataLink", {})
+            .get("type"),
+        )
+
+    def __repr__(self):
+        return json.dumps(self.post_payload(), indent=4)

geoservercloud/models/featuretypes.py

@@ -0,0 +1,23 @@
+import json
+
+from requests.models import Response
+
+
+class FeatureTypes:
+
+    def __init__(self, featuretypes: list = []) -> None:
+        self._featuretypes = featuretypes
+
+    @property
+    def featuretypes(self):
+        return self._featuretypes
+
+    @classmethod
+    def from_dict(cls, content: dict):
+        featuretypes = []
+        for featuretype in content.get("featureTypes", {}).get("featureType", []):
+            featuretypes.append(featuretype["name"])
+        return cls(featuretypes)
+
+    def __repr__(self):
+        return json.dumps(self._featuretypes, indent=4)