chg: Use new annotations

.pre-commit-config.yaml

@@ -2,14 +2,14 @@
 # See https://pre-commit.com/hooks.html for more hooks
 repos:
 -   repo: https://github.com/pre-commit/pre-commit-hooks
-    rev: v4.1.0
+    rev: v4.5.0
     hooks:
     -   id: trailing-whitespace
     -   id: end-of-file-fixer
     -   id: check-yaml
     -   id: check-added-large-files
 -   repo: https://github.com/asottile/pyupgrade
-    rev: v2.31.1
+    rev: v3.15.0
     hooks:
     -   id: pyupgrade
         args: [--py38-plus]

bin/capture_manager.py

@@ -1,5 +1,7 @@
 #!/usr/bin/env python3
 
+from __future__ import annotations
+
 import asyncio
 import logging
 import logging.config
@@ -19,13 +21,13 @@
 
 class CaptureManager(AbstractManager):
 
-    def __init__(self, loglevel: Optional[int]=None):
+    def __init__(self, loglevel: Optional[int]=None) -> None:
         super().__init__(loglevel)
         self.script_name = 'capture_manager'
-        self.captures: Set[Task] = set()
+        self.captures: Set[Task] = set()  # type: ignore[type-arg]
         self.lacus = Lacus()
 
-    async def clear_dead_captures(self):
+    async def clear_dead_captures(self) -> None:
         ongoing = {capture.get_name(): capture for capture in self.captures}
         max_capture_time = get_config('generic', 'max_capture_time')
         oldest_start_time = datetime.now() - timedelta(seconds=max_capture_time + 300)
@@ -44,7 +46,7 @@ async def clear_dead_captures(self):
                 except asyncio.CancelledError:
                     self.logger.warning(f'{expected_uuid} is canceled now.')
 
-    async def _to_run_forever_async(self):
+    async def _to_run_forever_async(self) -> None:
         await self.clear_dead_captures()
         if self.force_stop:
             return
@@ -61,7 +63,7 @@ async def _to_run_forever_async(self):
         #       be decremented when it finishes
         self.set_running(len(self.captures) + 1)
 
-    async def _wait_to_finish_async(self):
+    async def _wait_to_finish_async(self) -> None:
         while self.captures:
             self.logger.info(f'Waiting for {len(self.captures)} capture(s) to finish...')
             self.logger.info(f'Ongoing captures: {", ".join(capture.get_name() for capture in self.captures)}')
@@ -70,7 +72,7 @@ async def _wait_to_finish_async(self):
         self.logger.info('No more captures')
 
 
-def main():
+def main() -> None:
     p = CaptureManager()
 
     loop = asyncio.new_event_loop()

bin/run_backend.py

@@ -24,26 +24,26 @@ def check_running(name: str) -> bool:
         return False
 
 
-def launch_cache(storage_directory: Optional[Path]=None):
+def launch_cache(storage_directory: Optional[Path]=None) -> None:
     if not storage_directory:
         storage_directory = get_homedir()
     if not check_running('cache'):
         Popen(["./run_redis.sh"], cwd=(storage_directory / 'cache'))
 
 
-def shutdown_cache(storage_directory: Optional[Path]=None):
+def shutdown_cache(storage_directory: Optional[Path]=None) -> None:
     if not storage_directory:
         storage_directory = get_homedir()
     r = Redis(unix_socket_path=get_socket_path('cache'))
     r.shutdown(save=True)
     print('Redis cache database shutdown.')
 
 
-def launch_all():
+def launch_all() -> None:
     launch_cache()
 
 
-def check_all(stop: bool=False):
+def check_all(stop: bool=False) -> None:
     backends: Dict[str, bool] = {'cache': False}
     while True:
         for db_name in backends.keys():
@@ -65,11 +65,11 @@ def check_all(stop: bool=False):
         time.sleep(1)
 
 
-def stop_all():
+def stop_all() -> None:
     shutdown_cache()
 
 
-def main():
+def main() -> None:
     parser = argparse.ArgumentParser(description='Manage backend DBs.')
     parser.add_argument("--start", action='store_true', default=False, help="Start all")
     parser.add_argument("--stop", action='store_true', default=False, help="Stop all")

bin/shutdown.py

@@ -5,7 +5,7 @@
 from lacus.default import AbstractManager
 
 
-def main():
+def main() -> None:
     AbstractManager.force_shutdown()
     time.sleep(5)
     while True:

bin/start.py

@@ -5,7 +5,7 @@
 from lacus.default import get_homedir
 
 
-def main():
+def main() -> None:
     # Just fail if the env isn't set.
     get_homedir()
     print('Start backend (redis)...')

bin/start_website.py

@@ -15,13 +15,13 @@
 
 class Website(AbstractManager):
 
-    def __init__(self, loglevel: Optional[int]=None):
+    def __init__(self, loglevel: Optional[int]=None) -> None:
         super().__init__(loglevel)
         self.script_name = 'website'
         self.process = self._launch_website()
         self.set_running()
 
-    def _launch_website(self):
+    def _launch_website(self) -> Popen:  # type: ignore[type-arg]
         website_dir = get_homedir() / 'website'
         ip = get_config('generic', 'website_listen_ip')
         port = get_config('generic', 'website_listen_port')
@@ -33,7 +33,7 @@ def _launch_website(self):
                      cwd=website_dir)
 
 
-def main():
+def main() -> None:
     w = Website()
     w.run(sleep_in_sec=10)
 

bin/stop.py

@@ -8,7 +8,7 @@
 from lacus.default import get_homedir, get_socket_path
 
 
-def main():
+def main() -> None:
     get_homedir()
     p = Popen(['shutdown'])
     p.wait()

bin/stop_capture_manager.py

@@ -12,7 +12,7 @@
 logging.config.dictConfig(get_config('logging'))
 
 
-def main():
+def main() -> None:
     parser = argparse.ArgumentParser(description='Sends a SIGTERM to the capture_manager so you can restart it cleanly.')
     parser.parse_args()
 

bin/update.py

@@ -15,14 +15,14 @@
 logging.config.dictConfig(get_config('logging'))
 
 
-def compute_hash_self():
+def compute_hash_self() -> bytes:
     m = hashlib.sha256()
     with (get_homedir() / 'bin' / 'update.py').open('rb') as f:
         m.update(f.read())
         return m.digest()
 
 
-def keep_going(ignore=False):
+def keep_going(ignore: bool=False) -> None:
     if ignore:
         return
     keep_going = input('Continue? (y/N) ')
@@ -31,7 +31,7 @@ def keep_going(ignore=False):
         sys.exit()
 
 
-def run_command(command, expect_fail: bool=False, capture_output: bool=True):
+def run_command(command: str, expect_fail: bool=False, capture_output: bool=True) -> None:
     args = shlex.split(command)
     homedir = get_homedir()
     process = subprocess.run(args, cwd=homedir, capture_output=capture_output)
@@ -42,7 +42,7 @@ def run_command(command, expect_fail: bool=False, capture_output: bool=True):
         sys.exit()
 
 
-def check_poetry_version():
+def check_poetry_version() -> None:
     args = shlex.split("poetry self -V")
     homedir = get_homedir()
     process = subprocess.run(args, cwd=homedir, capture_output=True)
@@ -58,7 +58,7 @@ def check_poetry_version():
         sys.exit()
 
 
-def main():
+def main() -> None:
     parser = argparse.ArgumentParser(description='Pull latest release, update dependencies, update and validate the config files, update 3rd deps for the website.')
     parser.add_argument('--yes', default=False, action='store_true', help='Run all commands without asking.')
     parser.add_argument('--init', default=False, action='store_true', help='Run all commands without starting the service.')

lacus/default/__init__.py

@@ -7,8 +7,25 @@
 # and allow to update them easily.
 # You should not have to change anything in this file below this line.
 
+import os  # noqa
+
 from .abstractmanager import AbstractManager  # noqa
 
 from .exceptions import MissingEnv, CreateDirectoryException, ConfigError  # noqa
 
 from .helpers import get_homedir, load_configs, get_config, safe_create_dir, get_socket_path, try_make_file  # noqa
+
+os.chdir(get_homedir())
+
+__all__ = [
+    'AbstractManager',
+    'MissingEnv',
+    'CreateDirectoryException',
+    'ConfigError',
+    'get_homedir',
+    'load_configs',
+    'get_config',
+    'safe_create_dir',
+    'get_socket_path',
+    'try_make_file',
+]

lacus/default/abstractmanager.py

@@ -1,5 +1,7 @@
 #!/usr/bin/env python3
 
+from __future__ import annotations
+
 import asyncio
 import logging
 import os
@@ -20,18 +22,18 @@ class AbstractManager(ABC):
 
     script_name: str
 
-    def __init__(self, loglevel: Optional[int]=None):
+    def __init__(self, loglevel: int | None=None):
         self.loglevel: int = loglevel if loglevel is not None else get_config('generic', 'loglevel') or logging.INFO
         self.logger = logging.getLogger(f'{self.__class__.__name__}')
         self.logger.setLevel(self.loglevel)
         self.logger.info(f'Initializing {self.__class__.__name__}')
-        self.process: Optional[Popen] = None
+        self.process: Popen | None = None  # type: ignore[type-arg]
         self.__redis = Redis(unix_socket_path=get_socket_path('cache'), db=1, decode_responses=True)
 
         self.force_stop = False
 
     @staticmethod
-    def is_running() -> List[Tuple[str, float]]:
+    def is_running() -> list[tuple[str, float]]:
         try:
             r = Redis(unix_socket_path=get_socket_path('cache'), db=1, decode_responses=True)
             for script_name, score in r.zrangebyscore('running', '-inf', '+inf', withscores=True):
@@ -52,22 +54,22 @@ def is_running() -> List[Tuple[str, float]]:
             return []
 
     @staticmethod
-    def clear_running():
+    def clear_running() -> None:
         try:
             r = Redis(unix_socket_path=get_socket_path('cache'), db=1, decode_responses=True)
             r.delete('running')
         except RedisConnectionError:
             print('Unable to connect to redis, the system is down.')
 
     @staticmethod
-    def force_shutdown():
+    def force_shutdown() -> None:
         try:
             r = Redis(unix_socket_path=get_socket_path('cache'), db=1, decode_responses=True)
             r.set('shutdown', 1)
         except RedisConnectionError:
             print('Unable to connect to redis, the system is down.')
 
-    def set_running(self, number: Optional[int]=None) -> None:
+    def set_running(self, number: int | None=None) -> None:
         if number == 0:
             self.__redis.zrem('running', self.script_name)
         else:
@@ -111,7 +113,7 @@ def shutdown_requested(self) -> bool:
     def _to_run_forever(self) -> None:
         raise NotImplementedError('This method must be implemented by the child')
 
-    def _kill_process(self):
+    def _kill_process(self) -> None:
         if self.process is None:
             return
         kill_order = [signal.SIGWINCH, signal.SIGTERM, signal.SIGINT, signal.SIGKILL]
@@ -167,7 +169,7 @@ def run(self, sleep_in_sec: int) -> None:
     def _wait_to_finish(self) -> None:
         self.logger.info('Not implemented, nothing to wait for.')
 
-    async def stop(self):
+    async def stop(self) -> None:
         self.force_stop = True
 
     async def _to_run_forever_async(self) -> None:
@@ -176,7 +178,7 @@ async def _to_run_forever_async(self) -> None:
     async def _wait_to_finish_async(self) -> None:
         self.logger.info('Not implemented, nothing to wait for.')
 
-    async def stop_async(self):
+    async def stop_async(self) -> None:
         """Method to pass the signal handler:
             loop.add_signal_handler(signal.SIGTERM, lambda: loop.create_task(p.stop()))
         """

lacus/default/helpers.py

@@ -1,4 +1,7 @@
 #!/usr/bin/env python3
+
+from __future__ import annotations
+
 import json
 import logging
 import os
@@ -9,7 +12,7 @@
 from . import env_global_name
 from .exceptions import ConfigError, CreateDirectoryException, MissingEnv
 
-configs: Dict[str, Dict[str, Any]] = {}
+configs: dict[str, dict[str, Any]] = {}
 logger = logging.getLogger('Helpers')
 
 
@@ -34,7 +37,7 @@ def get_homedir() -> Path:
 
 
 @lru_cache(64)
-def load_configs(path_to_config_files: Optional[Union[str, Path]]=None):
+def load_configs(path_to_config_files: str | Path | None=None) -> None:
     global configs
     if configs:
         return
@@ -57,7 +60,7 @@ def load_configs(path_to_config_files: Optional[Union[str, Path]]=None):
 
 
 @lru_cache(64)
-def get_config(config_type: str, entry: Optional[str]=None, quiet: bool=False) -> Any:
+def get_config(config_type: str, entry: str | None=None, quiet: bool=False) -> Any:
     """Get an entry from the given config_type file. Automatic fallback to the sample file"""
     global configs
     if not configs:
@@ -96,7 +99,7 @@ def get_socket_path(name: str) -> str:
     return str(get_homedir() / mapping[name])
 
 
-def try_make_file(filename: Path):
+def try_make_file(filename: Path) -> bool:
     try:
         filename.touch(exist_ok=False)
         return True