WIP: import old models from importer_script for inspiration

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,87 @@
+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:
+    """
+    Class to handle internationalization of strings
+    items like title, abstract, etc. that can be internationalized
+    become a dictionary with the key being the language code
+    and their key in the payload changes to internationalizedTitle, internationalizedAbstract, etc.
+    """
+
+    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._payload = (self.str_key, self._value)
+        elif isinstance(value, dict):
+            self._payload = (self._i18n_key, self._value)  # type: ignore
+        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
+
+    @property
+    def payload_tuple(self):
+        return self._payload
+
+    def __repr__(self):
+        return json.dumps({self._payload[0]: self._payload[1]}, indent=4)

geoservercloud/models/datastore.py

@@ -0,0 +1,77 @@
+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: KeyDollarListDict,
+        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 = 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_response(cls, response: Response):
+        if response.status_code == 404:
+            return None
+        json_data = response.json()
+        connection_parameters = cls.parse_connection_parameters(json_data)
+        return cls(
+            json_data.get("dataStore", {}).get("workspace", {}).get("name", None),
+            json_data.get("dataStore", {}).get("name", None),
+            connection_parameters,
+            json_data.get("dataStore", {}).get("type", "PostGIS"),
+            json_data.get("dataStore", {}).get("enabled", True),
+            json_data.get("dataStore", {}).get("description", None),
+        )
+
+    @classmethod
+    def parse_connection_parameters(cls, json_data):
+        return KeyDollarListDict(
+            json_data.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_response(cls, response: Response):
+        json_data = response.json()
+        datastores = []
+        workspace_name = (
+            json_data.get("dataStores", {}).get("workspace", {}).get("name", None)
+        )
+        for store in json_data.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,134 @@
+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,
+                self._title.payload_tuple[0]: self._title.payload_tuple[1],
+                self._abstract.payload_tuple[0]: self._abstract.payload_tuple[1],
+                "srs": self.srs,
+                "keywords": self.keywords,
+            }
+        }
+        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_response(cls, response: Response):
+        json_data = response.json()
+        try:
+            abstract = json_data["featureType"]["abstract"]
+        except KeyError:
+            abstract = json_data["featureType"]["internationalAbstract"]
+        try:
+            title = json_data["featureType"]["title"]
+        except KeyError:
+            title = json_data["featureType"]["internationalTitle"]
+
+        return cls(
+            namespace_name=json_data["featureType"]["namespace"]["name"],
+            name=json_data["featureType"]["name"],
+            native_name=json_data["featureType"]["nativeName"],
+            title=title,
+            abstract=abstract,
+            srs=json_data["featureType"]["srs"],
+            keywords=json_data["featureType"]["keywords"],
+            attributes=json_data["featureType"].get("attributes", None),
+            metadata_url=json_data["featureType"]
+            .get("metadataLinks", {})
+            .get("metadataLink", {})
+            .get("content"),
+            metadata_type=json_data["featureType"]
+            .get("metadataLinks", {})
+            .get("metadataLink", {})
+            .get("metadataType"),
+            metadata_format=json_data["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,24 @@
+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_response(cls, response: Response):
+        featuretypes = []
+        json_data = response.json()
+        for featuretype in json_data.get("featureTypes", {}).get("featureType", []):
+            featuretypes.append(featuretype["name"])
+        return cls(featuretypes)
+
+    def __repr__(self):
+        return json.dumps(self._featuretypes, indent=4)