Add userinfo, permissions, wms handlers and async-execution polling

src/geonoderest/datasets.py

@@ -10,6 +10,7 @@
 from geonoderest.cmdprint import show_list, print_json
 from geonoderest.geonodetypes import GeonodeCmdOutListKey, GeonodeCmdOutDictKey
 from geonoderest.executionrequest import GeonodeExecutionRequestHandler
+from geonoderest.exceptions import GeoNodeRestException
 
 
 class GeonodeDatasetsHandler(GeonodeResourceHandler):
@@ -107,26 +108,18 @@ def __wait_for_upload__(self, exec_id: str, poll_interval: int = 5) -> List[int]
         execution_request_handler = GeonodeExecutionRequestHandler(
             env=self.gn_credentials
         )
-        elapsed = 0
-        while True:
-            er = execution_request_handler.get(exec_id=exec_id)
-            status = er.get("status", "")
-            if status in ("finished", "failed"):
-                break
-            logging.info(f"waiting for upload to finish ({elapsed}s) ...")
-            time.sleep(poll_interval)
-            elapsed += poll_interval
-
-        if er.get("status") == "failed":
-            logging.error("upload failed ...")
-            logging.error(er)
+        try:
+            er = execution_request_handler.wait_for_completion(
+                exec_id=exec_id, poll_interval=poll_interval
+            )
+        except GeoNodeRestException as exc:
+            logging.error(str(exc))
             sys.exit(1)
-
         logging.info("upload finished ...")
-        pks = []
-        for resource in er.get("output_params", {}).get("resources", []):
-            pks.append(resource["id"])
-        return pks
+        return [
+            resource["id"]
+            for resource in er.get("output_params", {}).get("resources", [])
+        ]
 
     def upload(
         self,

src/geonoderest/executionrequest.py

@@ -1,11 +1,15 @@
 from typing import Dict, List, Optional
 import logging
+import time
 
 from geonoderest.geonodetypes import GeonodeCmdOutListKey, GeonodeCmdOutObjectKey
 from geonoderest.rest import GeonodeRest
+from geonoderest.exceptions import GeoNodeRestException
 
 from geonoderest.cmdprint import print_list_on_cmd, print_json
 
+TERMINAL_STATUSES = ("finished", "failed")
+
 
 class GeonodeExecutionRequestHandler(GeonodeRest):
     ENDPOINT_NAME = "executionrequest"
@@ -63,3 +67,60 @@ def list(self, **kwargs) -> Optional[Dict]:
         if r is None:
             return None
         return r[self.JSON_OBJECT_NAME]
+
+    def wait_for_completion(
+        self,
+        exec_id: str,
+        poll_interval: int = 2,
+        timeout: int = 600,
+        on_poll=None,
+    ) -> Dict:
+        """Poll an execution request until it reaches a terminal status.
+
+        Used by any async-acknowledged operation that returns an execution_id:
+        uploads, async resource deletes, permission changes, etc.
+
+        Args:
+            exec_id (str): Execution request UUID.
+            poll_interval (int): Seconds between polls. Defaults to 2.
+            timeout (int): Max seconds to wait before giving up. Defaults to 600.
+            on_poll (callable, optional): Invoked with the latest record after
+                each poll. Useful for surfacing progress to the user.
+
+        Returns:
+            Dict: The terminal execution record (status == "finished").
+
+        Raises:
+            GeoNodeRestException: If the execution reports `failed` or the
+                timeout elapses before reaching a terminal status.
+        """
+        elapsed = 0
+        last: Optional[Dict] = None
+        while elapsed <= timeout:
+            er = self.get(exec_id=exec_id)
+            if er is None:
+                raise GeoNodeRestException(
+                    f"execution {exec_id} not found while polling"
+                )
+            last = er
+            status = (er.get("status") or "").lower()
+            if on_poll is not None:
+                on_poll(er)
+            if status == "finished":
+                logging.info(f"execution {exec_id} finished in ~{elapsed}s")
+                return er
+            if status == "failed":
+                raise GeoNodeRestException(
+                    f"execution {exec_id} failed: "
+                    f"{er.get('log') or er.get('output_params')}"
+                )
+            logging.debug(
+                f"waiting for execution {exec_id} (status={status}, "
+                f"elapsed={elapsed}s) ..."
+            )
+            time.sleep(poll_interval)
+            elapsed += poll_interval
+        raise GeoNodeRestException(
+            f"execution {exec_id} did not reach a terminal status within "
+            f"{timeout}s (last status: {last.get('status') if last else 'unknown'})"
+        )

src/geonoderest/permissions.py

@@ -0,0 +1,66 @@
+from typing import Dict, Optional
+
+from geonoderest.rest import GeonodeRest
+from geonoderest.executionrequest import GeonodeExecutionRequestHandler
+
+
+class GeonodePermissionsHandler(GeonodeRest):
+    """GET / PUT /api/v2/resources/{pk}/permissions.
+
+    The PUT endpoint is asynchronous: the response carries an
+    ``execution_id`` and ``status_url`` rather than the applied state.
+    Callers should invoke :meth:`wait_for_completion` (or poll the
+    ``execution_id`` themselves) before assuming the change is visible to
+    downstream readers.
+    """
+
+    def get(self, pk: int) -> Optional[Dict]:
+        """Return the current permission set for a resource.
+
+        Args:
+            pk (int): Resource primary key.
+
+        Returns:
+            Dict with keys ``users``, ``groups``, ``organizations``, each
+            holding a list of entries with id, name, and permission level,
+            or None on error.
+        """
+        return self.http_get(endpoint=f"resources/{pk}/permissions")
+
+    def set(self, pk: int, payload: Dict) -> Optional[Dict]:
+        """Submit a new permission set. Returns the async receipt.
+
+        The receipt has the shape::
+
+            {
+              "status": "ready",
+              "execution_id": "<uuid>",
+              "status_url": "<absolute url>"
+            }
+
+        Pass the ``execution_id`` to :meth:`wait_for_completion` to block
+        until the permission change has been applied.
+        """
+        return self.http_put(
+            endpoint=f"resources/{pk}/permissions", json_content=payload
+        )
+
+    def wait_for_completion(
+        self,
+        exec_id: str,
+        poll_interval: int = 2,
+        timeout: int = 120,
+        on_poll=None,
+    ) -> Dict:
+        """Poll an in-flight permission change until it reaches a terminal state.
+
+        Delegates to :class:`GeonodeExecutionRequestHandler` so the polling
+        behavior matches uploads and async deletes.
+        """
+        handler = GeonodeExecutionRequestHandler(env=self.gn_credentials)
+        return handler.wait_for_completion(
+            exec_id=exec_id,
+            poll_interval=poll_interval,
+            timeout=timeout,
+            on_poll=on_poll,
+        )

src/geonoderest/resources.py

@@ -1,10 +1,11 @@
-from typing import List
+from typing import List, Dict, Optional
 import requests
 import logging
 
 from geonoderest.geonodeobject import GeonodeObjectHandler
 from geonoderest.geonodetypes import GeonodeCmdOutListKey, GeonodeCmdOutDictKey
 from geonoderest.exceptions import GeoNodeRestException
+from geonoderest.executionrequest import GeonodeExecutionRequestHandler
 
 SUPPORTED_METADATA_TYPES: List[str] = [
     "Atom",
@@ -62,3 +63,35 @@ def metadata(
         link: str
         link = [m for m in r["links"] if m["name"] == metadata_type][0]["url"]
         return self.http_get_download(link)
+
+    def delete_async(self, pk: int) -> Optional[Dict]:
+        """Asynchronous delete via ``DELETE /api/v2/resources/{pk}/delete``.
+
+        Unlike :meth:`delete` (which hits the synchronous endpoint), this
+        returns an async receipt — ``{status, execution_id, status_url}`` —
+        suitable for passing to :meth:`wait_for_completion`. Use this when
+        the caller needs to confirm the deletion completed (e.g. before
+        relying on it in a downstream assertion).
+        """
+        return self.http_delete(endpoint=f"{self.ENDPOINT_NAME}/{pk}/delete")
+
+    def wait_for_completion(
+        self,
+        exec_id: str,
+        poll_interval: int = 2,
+        timeout: int = 180,
+        on_poll=None,
+    ) -> Dict:
+        """Poll an async resource operation (delete, copy, …) to completion.
+
+        Delegates to :class:`GeonodeExecutionRequestHandler` so the polling
+        behavior is consistent across uploads, deletes, and permission
+        changes.
+        """
+        handler = GeonodeExecutionRequestHandler(env=self.gn_credentials)
+        return handler.wait_for_completion(
+            exec_id=exec_id,
+            poll_interval=poll_interval,
+            timeout=timeout,
+            on_poll=on_poll,
+        )

src/geonoderest/rest.py

@@ -244,6 +244,72 @@ def http_get(self, endpoint: str, params: Dict = {}) -> Optional[Dict]:
             return None
         return r.json()
 
+    @network_exception_handling
+    def http_put(
+        self, endpoint: str, json_content: Dict = {}, params: Dict = {}, **kwargs
+    ) -> Optional[Dict]:
+        """
+        Execute HTTP PUT request on the specified endpoint with optional parameters.
+
+        Args:
+            endpoint (str): The API endpoint to send the PUT request to.
+            json_content (Dict, optional): A dictionary of JSON data to include in the request body.
+            params (Dict, optional): A dictionary of query parameters to include in the request.
+
+        Returns:
+            Dict: The JSON response from the server, or None if an error occurred.
+        """
+        url = self.url + endpoint
+        try:
+            logging.debug(
+                f"PUT URL: {url}, headers: {self.header}, params: {params}, json: {json_content}"
+            )
+            r = requests.put(
+                url,
+                headers=self.header,
+                json=json_content,
+                params=params,
+                verify=self.verify,
+            )
+            r.raise_for_status()
+        except requests.exceptions.HTTPError as err:
+            if r is not None:
+                logging.error(f"PUT error response: {r.text}")
+            logging.error(err)
+            return None
+        return r.json()
+
+    @network_exception_handling
+    def http_get_anonymous(
+        self,
+        endpoint: str = "",
+        url: Optional[str] = None,
+        params: Dict = {},
+    ) -> requests.Response:
+        """
+        Execute an HTTP GET without sending session credentials.
+
+        Useful for probes that must verify the API behavior for unauthenticated
+        callers — e.g. confirming a permission restriction prevents anonymous
+        reads. No Authorization header, no session cookie. Returns the raw
+        `requests.Response` so callers can inspect status code, headers, and
+        body; HTTP error statuses are NOT raised because they are frequently
+        the expected outcome.
+
+        Args:
+            endpoint (str): Endpoint relative to the configured API base URL.
+                Ignored when `url` is provided.
+            url (Optional[str]): Absolute URL to GET. Takes precedence over
+                `endpoint`.
+            params (Dict): Query-string parameters.
+
+        Returns:
+            requests.Response: The raw response.
+        """
+        target_url = url if url else self.url + endpoint
+        logging.debug(f"GET (anonymous) URL: {target_url}, params: {params}")
+        return requests.get(target_url, params=params, verify=self.verify)
+
     @network_exception_handling
     def http_patch(
         self, endpoint: str, json_content: Dict = {}, params: Dict = {}, **kwargs

src/geonoderest/userinfo.py

@@ -0,0 +1,18 @@
+from typing import Dict, Optional
+
+from geonoderest.rest import GeonodeRest
+
+
+class GeonodeUserInfoHandler(GeonodeRest):
+    """GET /api/v2/userinfo — identity claims for the authenticated user.
+
+    On GeoNode deployments that grant a session token, the response also
+    includes an `access_token` usable to authenticate downstream services
+    such as the bundled GeoServer.
+    """
+
+    ENDPOINT_NAME = "userinfo"
+
+    def get(self, **kwargs) -> Optional[Dict]:
+        """Return the userinfo claims dict, or None on error."""
+        return self.http_get(endpoint=f"{self.ENDPOINT_NAME}/")