black 79

Author: ninsbl

src/actinia/actinia.py

@@ -69,7 +69,9 @@ def __set_url(self):
 
     def __check_version(self):
         version_url = f"{self.url}/version"
-        data = request_and_check("GET", version_url, **{"timeout": self.timeout})
+        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.")
@@ -89,7 +91,9 @@ def __check_version(self):
                     self.api_version = "v1"
                     self.__check_version()
             else:
-                raise Exception(f"Connection to actinia server <{self.url}> failed!")
+                raise Exception(
+                    f"Connection to actinia server <{self.url}> failed!"
+                )
 
     def get_version(self):
         """
@@ -98,7 +102,9 @@ def get_version(self):
         """
 
         version_url = f"{self.url}/version"
-        return request_and_check("GET", version_url, **{"timeout": self.timeout})
+        return request_and_check(
+            "GET", version_url, **{"timeout": self.timeout}
+        )
 
     def __check_auth(self):
         url = f"{self.url}/locations"
@@ -145,7 +151,9 @@ def __request_locations(self):
         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}
+        loc = {
+            lname: Location(lname, self, self.__auth) for lname in loc_names
+        }
         self.locations = loc
 
     def create_location(self, name, epsgcode):

src/actinia/job.py

@@ -92,7 +92,10 @@ def poll_until_finished(self, waiting_time=5, quiet=False):
             self.poll(quiet=True)
             if self.status not in ["accepted", "running"]:
                 status_accepted_running = False
-                msg = f"Status of {self.name} job is {self.status}: " f"{self.message}"
+                msg = (
+                    f"Status of {self.name} job is {self.status}: "
+                    f"{self.message}"
+                )
                 if self.status in ["terminated", "error"]:
                     log.error(msg)
                     return 1
@@ -102,7 +105,10 @@ def poll_until_finished(self, waiting_time=5, quiet=False):
             sleep(waiting_time)
             if self.status != status and not quiet:
                 status = self.status
-                msg = f"Status of {self.name} job is {self.status}: " f"{self.message}"
+                msg = (
+                    f"Status of {self.name} job is {self.status}: "
+                    f"{self.message}"
+                )
                 log.info(msg)
 
     def terminate(self):

src/actinia/location.py

@@ -122,7 +122,9 @@ def delete_mapset(self, name):
         """
         if self.mapsets is None or len(self.mapsets) == 0:
             self.__request_mapsets()
-        Mapset.delete_mapset_request(name, self.name, self.__actinia, self.__auth)
+        Mapset.delete_mapset_request(
+            name, self.name, self.__actinia, self.__auth
+        )
         if name and name in self.mapsets:
             del self.mapsets[name]
         return self.mapsets
@@ -160,7 +162,9 @@ def validate_process_chain_async(self, pc, name=None):
         actiniaResp = self.__validate_process_chain(pc, "async")
         orig_name, name = set_job_names(name, "unknown_validation_job")
         if actiniaResp.status_code != 200:
-            raise Exception(f"Error {actiniaResp.status_code}: {actiniaResp.text}")
+            raise Exception(
+                f"Error {actiniaResp.status_code}: {actiniaResp.text}"
+            )
         resp = json.loads(actiniaResp.text)
         job = Job(orig_name, self.__actinia, self.__auth, resp)
         self.__actinia.jobs[name] = job
@@ -177,7 +181,10 @@ def create_processing_export_job(self, pc, name=None):
         # set name
         orig_name, name = set_job_names(name)
         # set endpoint in url
-        url = f"{self.__actinia.url}/locations/{self.name}/" "processing_async_export"
+        url = (
+            f"{self.__actinia.url}/locations/{self.name}/"
+            "processing_async_export"
+        )
         # make POST request
         postkwargs = {
             "headers": self.__actinia.headers,

tests/test_utils.py

@@ -51,7 +51,9 @@ def test_request_and_check_wrong_url(self):
         url = f"{ACTINIA_BASEURL}api/{ACTINIA_VERSION}/version_fail"
         err_msg = "The requested URL was not found on the server."
         with pytest.raises(Exception) as excinfo:
-            request_and_check("GET", url, status_code=(200,), **{"auth": ACTINIA_AUTH})
+            request_and_check(
+                "GET", url, status_code=(200,), **{"auth": ACTINIA_AUTH}
+            )
         assert err_msg in str(excinfo.value)
 
     def test_request_and_check_wrong_auth(self):
@@ -60,7 +62,9 @@ def test_request_and_check_wrong_auth(self):
         err_msg = "Wrong user or password. Please check your inputs."
         wrong_auth = ("actinia-gdi", "wrong_pw")
         with pytest.raises(Exception) as excinfo:
-            request_and_check("GET", url, status_code=(200,), **{"auth": wrong_auth})
+            request_and_check(
+                "GET", url, status_code=(200,), **{"auth": wrong_auth}
+            )
         assert err_msg in str(excinfo.value)
 
     def test_set_job_names(self):