wip

geoservercloud/geoservercloud.py

@@ -14,20 +14,13 @@
     DataStores,
     FeatureType,
     FeatureTypes,
-    KeyDollarListDict,
     PostGisDataStore,
     Style,
     Styles,
     Workspace,
     Workspaces,
 )
-from geoservercloud.services import (
-    AclEndpoints,
-    GwcEndpoints,
-    OwsEndpoints,
-    RestEndpoints,
-    RestService,
-)
+from geoservercloud.services import RestService
 from geoservercloud.templates import Templates
 
 
@@ -44,10 +37,6 @@ def __init__(
         self.password: str = password
         self.auth: tuple[str, str] = (user, password)
         self.rest_service: RestService = RestService(url, self.auth)
-        self.acl_endpoints: AclEndpoints = AclEndpoints()
-        self.gwc_endpoints: GwcEndpoints = GwcEndpoints()
-        self.ows_endpoints: OwsEndpoints = OwsEndpoints()
-        self.rest_endpoints: RestEndpoints = RestEndpoints()
         self.wms: WebMapService_1_3_0 | None = None
         self.wmts: WebMapTileService | None = None
         self.default_workspace: str | None = None
@@ -84,50 +73,39 @@ def create_wmts(self) -> None:
             password=self.password,
         )
 
-    def get_workspaces(self) -> Workspaces:
-        response: Response = self.get_request(self.rest_endpoints.workspaces())
-        workspaces = Workspaces.from_dict(response.json())
-        return workspaces
+    def get_workspaces(self) -> list[str]:
+        return self.rest_service.get_workspaces().workspaces
 
     def create_workspace(
         self,
         workspace_name: str,
         isolated: bool = False,
         set_default_workspace: bool = False,
-    ) -> Response:
+    ) -> dict[str, Any]:
         """
         Create a workspace in GeoServer, if it does not already exist.
         It if exists, update it
         """
-        response: Response = self.post_request(
-            self.rest_endpoints.workspaces(),
-            json=Workspace(workspace_name, isolated).post_payload(),
-        )
-        if response.status_code == 409:
-            response = self.put_request(
-                self.rest_endpoints.workspace(workspace_name),
-                json=Workspace(workspace_name, isolated).put_payload(),
-            )
+        workspace = self.rest_service.create_workspace(
+            workspace_name, isolated
+        ).asdict()
         if set_default_workspace:
             self.default_workspace = workspace_name
-        return response
+        return workspace
 
-    def delete_workspace(self, workspace_name: str) -> Response:
+    def delete_workspace(self, workspace_name: str) -> None:
         """
         Delete a GeoServer workspace (recursively)
         """
-        response: Response = self.delete_request(
-            self.rest_endpoints.workspace(workspace_name), params={"recurse": "true"}
-        )
+        self.rest_service.delete_workspace(workspace_name)
         if self.default_workspace == workspace_name:
             self.default_workspace = None
             self.wms = None
             self.wmts = None
-        return response
 
     def recreate_workspace(
         self, workspace_name: str, set_default_workspace: bool = False
-    ) -> Response:
+    ) -> dict[str, Any]:
         """
         Create a workspace in GeoServer, and first delete it if it already exists.
         """
@@ -136,14 +114,11 @@ def recreate_workspace(
             workspace_name, set_default_workspace=set_default_workspace
         )
 
-    def publish_workspace(self, workspace_name) -> Response:
+    def publish_workspace(self, workspace_name) -> None:
         """
         Publish the WMS service for a given workspace
         """
-        data: dict[str, dict[str, Any]] = Templates.workspace_wms(workspace_name)
-        return self.put_request(
-            self.rest_endpoints.workspace_wms_settings(workspace_name), json=data
-        )
+        self.rest_service.publish_workspace(workspace_name)
 
     def set_default_locale_for_service(
         self, workspace_name: str, locale: str | None

geoservercloud/models/workspace.py

@@ -28,5 +28,11 @@ def from_dict(cls, content: dict):
             content.get("workspace", {}).get("isolated", False),
         )
 
+    def asdict(self):
+        return {
+            "name": self.name,
+            "isolated": self.isolated,
+        }
+
     def __repr__(self):
         return json.dumps(self.put_payload(), indent=4)

geoservercloud/services/__init__.py

@@ -1,10 +1,5 @@
-from .endpoints import AclEndpoints, GwcEndpoints, OwsEndpoints, RestEndpoints
 from .restservice import RestService
 
 __all__ = [
     "RestService",
-    "AclEndpoints",
-    "OwsEndpoints",
-    "GwcEndpoints",
-    "RestEndpoints",
 ]

geoservercloud/services/restclient.py

@@ -0,0 +1,87 @@
+from typing import Any
+
+import requests
+
+TIMEOUT = 120
+
+
+class RestClient:
+    def __init__(self, url: str, auth: tuple[str, str]) -> None:
+        self.url: str = url
+        self.auth: tuple[str, str] = auth
+
+    def get(
+        self,
+        path: str,
+        params: dict[str, str] | None = None,
+        headers: dict[str, str] | None = None,
+    ) -> requests.Response:
+        response: requests.Response = requests.get(
+            f"{self.url}{path}",
+            params=params,
+            headers=headers,
+            auth=self.auth,
+            timeout=TIMEOUT,
+        )
+        if response.status_code != 404:
+            response.raise_for_status()
+        return response
+
+    def post(
+        self,
+        path: str,
+        params: dict[str, str] | None = None,
+        headers: dict[str, str] | None = None,
+        json: dict[str, dict[str, Any]] | None = None,
+        data: bytes | None = None,
+    ) -> requests.Response:
+
+        response: requests.Response = requests.post(
+            f"{self.url}{path}",
+            params=params,
+            headers=headers,
+            json=json,
+            data=data,
+            auth=self.auth,
+            timeout=TIMEOUT,
+        )
+        if response.status_code != 409:
+            response.raise_for_status()
+        return response
+
+    def put(
+        self,
+        path: str,
+        params: dict[str, str] | None = None,
+        headers: dict[str, str] | None = None,
+        json: dict[str, dict[str, Any]] | None = None,
+        data: bytes | None = None,
+    ) -> requests.Response:
+        response: requests.Response = requests.put(
+            f"{self.url}{path}",
+            params=params,
+            headers=headers,
+            json=json,
+            data=data,
+            auth=self.auth,
+            timeout=TIMEOUT,
+        )
+        response.raise_for_status()
+        return response
+
+    def delete(
+        self,
+        path: str,
+        params: dict[str, str] | None = None,
+        headers: dict[str, str] | None = None,
+    ) -> requests.Response:
+        response: requests.Response = requests.delete(
+            f"{self.url}{path}",
+            params=params,
+            headers=headers,
+            auth=self.auth,
+            timeout=TIMEOUT,
+        )
+        if response.status_code != 404:
+            response.raise_for_status()
+        return response