fix style errors in code

backend/maelstro/core/clone.py

@@ -6,7 +6,7 @@
 from maelstro.metadata import Meta
 from maelstro.config import app_config as config
 from maelstro.common.types import GsLayer
-from .georchestra import GsRestWrapper, get_georchestra_handler
+from .georchestra import GeorchestraHandler, GsRestWrapper, get_georchestra_handler
 from .exceptions import ParamError, MaelstroDetail, raise_for_status
 
 
@@ -21,6 +21,8 @@ def __init__(self, src_name: str, dst_name: str, uuid: str):
         self.copy_meta = False
         self.copy_layers = False
         self.copy_styles = False
+        self.meta: Meta
+        self.geo_hnd: GeorchestraHandler
 
     def clone_dataset(
         self,
@@ -36,10 +38,9 @@ def clone_dataset(
         with get_georchestra_handler() as geo_hnd:
             self.geo_hnd = geo_hnd
             operations = self._clone_dataset(output_format)
-            self.geo_hnd = None
         return operations
 
-    def _clone_dataset(self, output_format):
+    def _clone_dataset(self, output_format: str) -> str | list[Any]:
 
         if self.uuid:
             gn = self.geo_hnd.get_gn_service(self.src_name, True)
@@ -59,16 +60,20 @@ def _clone_dataset(self, output_format):
                 "destinations": [src["gs_url"] for src in config.get_destinations()],
             }
             pre_info, post_info = self.meta.update_geoverver_urls(mapping)
-            self.geo_hnd.response_handler.responses.append({
-                "operation": "Update of geoserver links in zip archive",
-                "before": pre_info,
-                "after": post_info
-            })
+            self.geo_hnd.response_handler.responses.append(
+                {
+                    "operation": "Update of geoserver links in zip archive",
+                    "before": pre_info,
+                    "after": post_info,
+                }
+            )
             results = gn_dst.put_record_zip(BytesIO(self.meta.get_zip()))
-            self.geo_hnd.response_handler.responses.append({
-                "message": results["msg"],
-                "detail": results["detail"],
-            })
+            self.geo_hnd.response_handler.responses.append(
+                {
+                    "message": results["msg"],
+                    "detail": results["detail"],
+                }
+            )
 
         if output_format == "text/plain":
             return self.geo_hnd.response_handler.get_formatted_responses()

backend/maelstro/core/georchestra.py

@@ -1,58 +1,73 @@
 from contextlib import contextmanager
 import inspect
 import json
-from typing import Any
+from typing import Any, Iterator
 from geonetwork import GnApi
-from geoservercloud.services import RestService
+from geoservercloud.services import RestService  # type: ignore
+from geoservercloud.services.restclient import RestClient  # type: ignore
 from requests.exceptions import HTTPError
 from maelstro.config import ConfigError, app_config as config
-from geoservercloud.services.restclient import RestClient
-from .operations import ResponseHandler, LoggedRequests, add_gn_handling, add_gs_handling, gs_logger
+from .operations import (
+    ResponseHandler,
+    LoggedRequests,
+    add_gn_handling,
+    add_gs_handling,
+    gs_logger,
+)
 from .exceptions import ParamError, MaelstroDetail, AuthError
 
 
-WRAPPERS = {
-    "GnApiWrapper": add_gn_handling,
-    "GsClientWrapper": add_gs_handling
-}
+WRAPPERS = {"GnApiWrapper": add_gn_handling, "GsClientWrapper": add_gs_handling}
 
 
 class MethodWrapper:
-    def __init__(self, response_handler, *args, **kwargs):
+    def __init__(self, response_handler: ResponseHandler, *args: Any, **kwargs: Any):
         super().__init__(*args, **kwargs)
         self.response_handler = response_handler
 
-    def __init_subclass__(cls, **kwargs):
-        super.__init_subclass__(**kwargs)
+    def __init_subclass__(cls, **kwargs: Any):
+        super().__init_subclass__(**kwargs)
         for k, v in inspect.getmembers(cls, inspect.isfunction):
             setattr(cls, k, WRAPPERS[cls.__name__](v))
 
 
 class GnApiWrapper(GnApi, MethodWrapper):
-    def __init__(self, response_handler, *args, **kwargs):
+    def __init__(self, response_handler: ResponseHandler, *args: Any, **kwargs: Any):
         super().__init__(*args, **kwargs)
         self.response_handler = response_handler
 
 
-class GsClientWrapper(RestClient, MethodWrapper):
+class GsClientWrapper(RestClient, MethodWrapper):  # type: ignore
     pass
 
 
-class GsRestWrapper(RestService):
-    def __init__(self, *args, **kwargs):
+class GsRestWrapper(RestService):  # type: ignore
+    def __init__(self, *args: Any, **kwargs: Any):
         super().__init__(*args, **kwargs)
-        setattr(self, "rest_client", GsClientWrapper(self.rest_client.url, self.rest_client.auth))
+        setattr(
+            self,
+            "rest_client",
+            GsClientWrapper(self.rest_client.url, self.rest_client.auth),
+        )
 
 
 class GeorchestraHandler:
     def __init__(self, response_handler: ResponseHandler):
         self.response_handler = response_handler
 
     def get_gn_service(self, instance_name: str, is_source: bool) -> GnApiWrapper:
+        if not self.response_handler.valid:
+            raise RuntimeError(
+                "GeorchestraHandler context invalid, handler already close"
+            )
         gn_info = self.get_service_info(instance_name, is_source, True)
         return GnApiWrapper(self.response_handler, gn_info["url"], gn_info["auth"])
 
     def get_gs_service(self, instance_name: str, is_source: bool) -> GsRestWrapper:
+        if not self.response_handler.valid:
+            raise RuntimeError(
+                "GeorchestraHandler context invalid, handler already close"
+            )
         gs_info = self.get_service_info(instance_name, is_source, False)
         gsapi = GsRestWrapper(gs_info["url"], gs_info["auth"])
         try:
@@ -76,14 +91,18 @@ def get_gs_service(self, instance_name: str, is_source: bool) -> GsRestWrapper:
         )
         return gsapi
 
-    def get_service_info(self, url: str, is_source: bool, is_geonetwork: bool) -> dict[str, Any]:
+    def get_service_info(
+        self, url: str, is_source: bool, is_geonetwork: bool
+    ) -> dict[str, Any]:
         try:
             service_info = config.get_access_info(
                 is_src=is_source, is_geonetwork=is_geonetwork, instance_id=url
             )
         except ConfigError as err:
             gs_logger.debug(
-                "Key not found: %s\n Config:\n%s", url, json.dumps(config.config, indent=4)
+                "Key not found: %s\n Config:\n%s",
+                url,
+                json.dumps(config.config, indent=4),
             )
             raise ParamError(
                 MaelstroDetail(
@@ -96,6 +115,6 @@ def get_service_info(self, url: str, is_source: bool, is_geonetwork: bool) -> di
 
 
 @contextmanager
-def get_georchestra_handler() -> GeorchestraHandler:
+def get_georchestra_handler() -> Iterator[GeorchestraHandler]:
     with LoggedRequests() as response_handler:
         yield GeorchestraHandler(response_handler)

backend/maelstro/core/operations.py

@@ -16,7 +16,8 @@
 class ResponseHandler(Handler):
     def __init__(self) -> None:
         super().__init__()
-        self.responses: list[Response | None] = []
+        self.responses: list[Response | None | dict[str, Any]] = []
+        self.valid = False
 
     def emit(self, record: logging.LogRecord) -> None:
         try:
@@ -70,15 +71,17 @@ def add_gs_handling(app_function: Callable[..., Any]) -> Callable[..., Any]:
     METHODS_TO_LOG = ["get", "post", "put", "delete"]
 
     if app_function.__name__ in METHODS_TO_LOG:
+
         def wrapped_function(path, *args, **kwargs):  # type: ignore
             try:
                 result = app_function(path, *args, **kwargs)
 
                 gs_logger.debug(
                     "[%s] %s: %s",
                     app_function.__name__,
-                    result.status_code, path,
-                    extra={"response": result}
+                    result.status_code,
+                    path,
+                    extra={"response": result},
                 )
                 return result
             except RequestException as err:
@@ -87,11 +90,14 @@ def wrapped_function(path, *args, **kwargs):  # type: ignore
                     app_function.__name__,
                     path,
                     err.__class__.__name__,
-                    extra={"response": err.request}
+                    extra={"response": err.request},
                 )
                 raise HTTPException(
                     status_code=504,
-                    detail={"message": f"HTTP error {err.__class__.__name__} at {path}", "info": err},
+                    detail={
+                        "message": f"HTTP error {err.__class__.__name__} at {path}",
+                        "info": err,
+                    },
                 ) from err
 
         return wrapped_function
@@ -105,8 +111,10 @@ def __init__(self) -> None:
     def __enter__(self) -> ResponseHandler:
         gn_logger.addHandler(self.handler)
         gs_logger.addHandler(self.handler)
+        self.handler.valid = True
         return self.handler
 
     def __exit__(self, *args: Any) -> None:
+        self.handler.valid = False
         gn_logger.removeHandler(self.handler)
         gs_logger.removeHandler(self.handler)