Merge pull request #44 from ninsbl/job_terminate

Add support for terminate job endpoint and unify request_and_check function

Author: anikaweinmann

.github/workflows/tests.yml

@@ -14,10 +14,10 @@ jobs:
     steps:
     - uses: actions/checkout@v4
     - name: Start containers
-      run: docker-compose -f "docker/docker-compose-test.yml" up -d --build
+      run: docker compose -f "docker/docker-compose-test.yml" up -d --build
     - name: List running docker
       run: docker ps
     - name: Running tests
       run: make test
     - name: Stop containers
-      run: docker-compose -f "docker/docker-compose-test.yml" down
+      run: docker compose -f "docker/docker-compose-test.yml" down

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "actinia-python-client"
-version = "0.3.1"
+version = "0.4.5"
 authors = [
   { name="Anika Weinmann", email="weinmann@mundialis.de" },
 ]

src/actinia/__init__.py

@@ -28,7 +28,7 @@
 __maintainer__ = "Anika Weinmann"
 
 from pkg_resources import get_distribution, DistributionNotFound
-from actinia.actinia import Actinia
+from actinia.actinia import Actinia as Actinia
 
 try:
     # Change here if project is renamed and does not equal the package name

src/actinia/actinia.py

@@ -27,13 +27,12 @@
 __copyright__ = "Copyright 2022, mundialis GmbH & Co. KG"
 __maintainer__ = "Anika Weinmann"
 
-import json
 import re
-import requests
 
 from actinia.location import Location
 from actinia.resources.templating import tplEnv
 from actinia.resources.logger import log
+from actinia.utils import request_and_check
 
 
 class Actinia:
@@ -43,13 +42,16 @@ def __init__(
         api_version="latest",
         user=None,
         pw=None,
+        connect_timeout=None,
+        read_timeout=None,
     ):
         self.api_prefix = api_version
         self.base_url = url
         self.headers = {"content-type": "application/json; charset=utf-8"}
         self.user = None
         self.__password = None
         self.__auth = None
+        self.timeout = (connect_timeout, read_timeout)
         if user and pw:
             self.set_authentication(user, pw)
         self.locations = dict()
@@ -67,11 +69,9 @@ def __set_url(self):
 
     def __check_version(self):
         version_url = f"{self.url}/version"
-        resp = requests.get(version_url)
-        if resp.status_code != 200:
-            raise Exception("Connection to actinia server failed!")
-
-        data = json.loads(resp.text)
+        data = request_and_check(
+            "GET", version_url, **{"timeout": self.timeout}
+        )
 
         if len(data) > 2:
             log.debug(f"{self.url} is working and will be used.")
@@ -102,23 +102,16 @@ def get_version(self):
         """
 
         version_url = f"{self.url}/version"
-        resp = requests.get(version_url)
-        if resp.status_code != 200:
-            raise Exception("Connection to actinia server failed!")
-        data = json.loads(resp.text)
-        return data
+        return request_and_check(
+            "GET", version_url, **{"timeout": self.timeout}
+        )
 
     def __check_auth(self):
         url = f"{self.url}/locations"
-        resp = requests.get(url, auth=(self.__auth))
-        if resp.status_code == 401:
-            raise Exception(
-                "Wrong user or password. Please check your inputs."
-            )
-        elif resp.status_code != 200:
-            raise Exception(f"Error {resp.status_code}: {resp.text}")
-        else:
-            log.debug(f"{self.user} is logged in.")
+        request_and_check(
+            "GET", url, **{"timeout": self.timeout, "auth": (self.__auth)}
+        )
+        log.debug(f"{self.user} is logged in.")
 
     def set_authentication(self, user, pw):
         """
@@ -155,11 +148,9 @@ def __request_locations(self):
             raise Exception("Authentication is not set.")
 
         url = f"{self.url}/locations"
-        resp = requests.get(url, auth=(self.__auth))
-        if resp.status_code != 200:
-            raise Exception(f"Error {resp.status_code}: {resp.text}")
-
-        loc_names = json.loads(resp.text)["locations"]
+        loc_names = request_and_check(
+            "GET", url, **{"timeout": self.timeout, "auth": (self.__auth)}
+        )["locations"]
         loc = {
             lname: Location(lname, self, self.__auth) for lname in loc_names
         }
@@ -178,18 +169,16 @@ def create_location(self, name, epsgcode):
         postbody = tpl.render(epsgcode=epsgcode)
 
         url = f"{self.url}/locations/{name}"
-        resp = requests.post(
+        request_and_check(
+            "POST",
             url,
-            auth=(self.__auth),
-            headers=self.headers,
-            data=postbody,
+            **{
+                "timeout": self.timeout,
+                "auth": (self.__auth),
+                "headers": self.headers,
+                "data": postbody,
+            },
         )
-        if resp.status_code != 200:
-            try:
-                msg = json.loads(resp.text)["message"]
-            except Exception:
-                msg = resp.text
-            raise Exception(f"Error {resp.status_code}: {msg}")
 
         location = Location(name, self, self.__auth)
         if len(self.locations) == 0:

src/actinia/job.py

@@ -27,11 +27,10 @@
 __copyright__ = "Copyright 2022, mundialis GmbH & Co. KG"
 __maintainer__ = "Anika Weinmann"
 
-import json
-import requests
 from time import sleep
 
 from actinia.resources.logger import log
+from actinia.utils import request_and_check
 
 
 class Job:
@@ -62,15 +61,13 @@ def poll(self, quiet=False):
         if self.status not in ["accepted", "running"]:
             log.warning("The job is not running and can not be updated.")
 
-        kwargs = dict()
-        kwargs["headers"] = self.__actinia.headers
-        kwargs["auth"] = self.__auth
+        kwargs = {
+            "headers": self.__actinia.headers,
+            "auth": self.__auth,
+            "timeout": self.__actinia.timeout,
+        }
         url = self.urls["status"]
-        try:
-            actiniaResp = requests.get(url, **kwargs)
-        except requests.exceptions.ConnectionError as e:
-            raise e
-        resp = json.loads(actiniaResp.text)
+        resp = request_and_check("GET", url, status_code=(200, 400), **kwargs)
 
         if "process_results" not in resp:
             resp["process_results"] = {}
@@ -114,9 +111,12 @@ def poll_until_finished(self, waiting_time=5, quiet=False):
                 )
                 log.info(msg)
 
-    # def terminate(self):
-    #     """
-    #     Terminate job.
-    #     """
-    #
-    # TODO
+    def terminate(self):
+        """Terminate the current job"""
+        kwargs = {"auth": self._Job__auth, "timeout": self.__actinia.timeout}
+        url = (
+            f"{self._Job__actinia.url}/resources/"
+            f"{self.user_id}/{self.resource_id}"
+        )
+        request_and_check("DELETE", url, **kwargs)
+        log.info("Termination request for job {self.resource_id} committed.")

src/actinia/location.py

@@ -36,7 +36,7 @@
 from actinia.region import Region
 from actinia.mapset import Mapset
 from actinia.job import Job
-from actinia.utils import set_job_names
+from actinia.utils import set_job_names, request_and_check
 
 
 class Location:
@@ -56,11 +56,11 @@ def __request_info(self):
             raise Exception("Authentication is not set.")
 
         url = f"{self.__actinia.url}/locations/{self.name}/info"
-        resp = requests.get(url, auth=(self.__auth))
-        if resp.status_code != 200:
-            raise Exception(f"Error {resp.status_code}: {resp.text}")
-
-        proc_res = json.loads(resp.text)["process_results"]
+        proc_res = request_and_check(
+            "GET",
+            url,
+            **{"auth": (self.__auth), "timeout": self.__actinia.timeout},
+        )["process_results"]
         self.projection = proc_res["projection"]
         self.region = Region(**proc_res["region"])
 
@@ -86,9 +86,11 @@ def __request_mapsets(self):
     def delete(self):
         """Delete a location via delete request."""
         url = f"{self.__actinia.url}/locations/{self.name}"
-        resp = requests.delete(url, auth=self.__auth)
-        if resp.status_code != 200:
-            raise Exception(f"Error {resp.status_code}: {resp.text}")
+        request_and_check(
+            "DELETE",
+            url,
+            **{"auth": self.__auth, "timeout": self.__actinia.timeout},
+        )
         del self.__actinia.locations[self.name]
 
     def get_mapsets(self):
@@ -123,7 +125,7 @@ def delete_mapset(self, name):
         Mapset.delete_mapset_request(
             name, self.name, self.__actinia, self.__auth
         )
-        if name is name in self.mapsets:
+        if name and name in self.mapsets:
             del self.mapsets[name]
         return self.mapsets
 
@@ -140,6 +142,7 @@ def __validate_process_chain(self, pc, type):
             auth=self.__auth,
             headers=self.__actinia.headers,
             data=json.dumps(pc),
+            timeout=self.__actinia.timeout,
         )
         return resp
 
@@ -183,9 +186,11 @@ def create_processing_export_job(self, pc, name=None):
             "processing_async_export"
         )
         # make POST request
-        postkwargs = dict()
-        postkwargs["headers"] = self.__actinia.headers
-        postkwargs["auth"] = self.__auth
+        postkwargs = {
+            "headers": self.__actinia.headers,
+            "auth": self.__auth,
+            "timeout": self.__actinia.timeout,
+        }
         if isinstance(pc, str):
             if os.path.isfile(pc):
                 with open(pc, "r") as pc_file:
@@ -197,12 +202,8 @@ def create_processing_export_job(self, pc, name=None):
         else:
             raise Exception("Given process chain has no valid type.")
 
-        try:
-            actiniaResp = requests.post(url, **postkwargs)
-        except requests.exceptions.ConnectionError as e:
-            raise e
+        resp = request_and_check("POST", url, **postkwargs)
         # create a job
-        resp = json.loads(actiniaResp.text)
         job = Job(orig_name, self.__actinia, self.__auth, resp)
         self.__actinia.jobs[name] = job
         return job