[local] Linting

Author: mcopik

sebs/local/config.py

@@ -1,6 +1,6 @@
 import json
 
-from typing import cast, Optional
+from typing import cast, Optional, Set
 
 from sebs.cache import Cache
 from sebs.faas.config import Config, Credentials, Resources
@@ -28,7 +28,7 @@ def __init__(self, storage_cfg: Optional[MinioConfig] = None):
         self._path: str = ""
         super().__init__(name="local")
         self._storage = storage_cfg
-        self._allocated_ports = set()
+        self._allocated_ports: Set[int] = set()
 
     @property
     def storage_config(self) -> Optional[MinioConfig]:
@@ -39,27 +39,33 @@ def allocated_ports(self) -> set:
         return self._allocated_ports
 
     def serialize(self) -> dict:
-        out = {
-            "allocated_ports": list(self._allocated_ports),
-            "storage": self._storage.serialize()
-        }
+        out: dict = {}
+        out["allocated_ports"] = list(self._allocated_ports)
+        if self._storage is not None:
+            out["storage"] = self._storage.serialize()
         return out
 
     @staticmethod
     def initialize(res: Resources, config: dict):
 
+        resources = cast(LocalResources, res)
         # Check for new config
         if "storage" in config:
-            res._storage = MinioConfig.deserialize(config["storage"])
-            res.logging.info("Using user-provided configuration of storage for local containers.")
+            resources._storage = MinioConfig.deserialize(config["storage"])
+            resources.logging.info(
+                "Using user-provided configuration of storage for local containers."
+            )
 
         if "allocated_ports" in config:
-            res._allocated_ports = set(config["allocated_ports"])
+            resources._allocated_ports = set(config["allocated_ports"])
 
     def update_cache(self, cache: Cache):
         super().update_cache(cache)
-        cache.update_config(val=list(self._allocated_ports), keys=["local", "resources", "allocated_ports"])
-        self._storage.update_cache(["local", "resources", "storage"], cache)
+        cache.update_config(
+            val=list(self._allocated_ports), keys=["local", "resources", "allocated_ports"]
+        )
+        if self._storage is not None:
+            self._storage.update_cache(["local", "resources", "storage"], cache)
 
     @staticmethod
     def deserialize(config: dict, cache: Cache, handlers: LoggingHandlers) -> Resources:
@@ -116,11 +122,7 @@ def deserialize(config: dict, cache: Cache, handlers: LoggingHandlers) -> Config
         return config_obj
 
     def serialize(self) -> dict:
-        out = {
-            "name": "local",
-            "region": self._region,
-            "resources": self._resources.serialize()
-        }
+        out = {"name": "local", "region": self._region, "resources": self._resources.serialize()}
         return out
 
     def update_cache(self, cache: Cache):

sebs/local/local.py

@@ -222,7 +222,7 @@ def create_function(self, code_package: Benchmark, func_name: str) -> "LocalFunc
                 )
 
             container_kwargs["command"] = f"/bin/bash /sebs/run_server.sh {port}"
-            container_kwargs["ports"] = {f'{port}/tcp': port}
+            container_kwargs["ports"] = {f"{port}/tcp": port}
 
         container = self._docker_client.containers.run(**container_kwargs)
 

sebs/utils.py

@@ -252,10 +252,12 @@ def logging_handlers(self, handlers: LoggingHandlers):
 def has_platform(name: str) -> bool:
     return os.environ.get(f"SEBS_WITH_{name.upper()}", "False").lower() == "true"
 
+
 # Check if the system is Linux and that it's not WSL
 def is_linux() -> bool:
     return platform.system() == "Linux" and "microsoft" not in platform.release().lower()
 
+
 def catch_interrupt():
 
     import signal