chore: Make diverse type checkers happy

Signed-off-by: Kai Blin <kblin@biosustain.dtu.dk>

Author: kblin

aiostandalone/__init__.py

@@ -1,3 +1,3 @@
 __version__ = '0.1.0'
 
-from aiostandalone.app import StandaloneApplication
+from aiostandalone.app import StandaloneApplication  # noqa

aiostandalone/signal.py aiostandalone/aiosignal.pyaiostandalone/signal.py renamed to aiostandalone/aiosignal.py

@@ -9,18 +9,17 @@ class Signal(list):
     """
     __slots__ = ('_app', )
 
-    def __init__(self, app):
+    def __init__(self, app) -> None:
         """Connect the signal to the app
 
         :param app: app the signal is conencted to
         """
         super().__init__()
         self._app = app
 
-    async def send(self, *args, **kwargs):
+    async def send(self, *args, **kwargs) -> None:
         """Send args and kwargs to all registered callbacks"""
         for callback in self:
             res = callback(*args, **kwargs)
             if asyncio.iscoroutine(res) or isinstance(res, asyncio.Future):
                 await res
-

aiostandalone/app.py

@@ -1,13 +1,20 @@
 """Standalone application class"""
 import asyncio
-from .signal import Signal
+from asyncio import AbstractEventLoop, CancelledError, Task
+from logging import Logger
+from typing import Any, Callable, Coroutine, Optional
+
+from .aiosignal import Signal
 from .log import fake_logger
 
+AppTask = Callable[["StandaloneApplication"], Coroutine[None, None, None]]
+
 
 class StandaloneApplication:
     """A standalone async application to run"""
 
-    def __init__(self, *, logger=fake_logger):
+    def __init__(self, *, logger: Logger = fake_logger,
+                 loop: Optional[AbstractEventLoop] = None) -> None:
         """Initialize the application to run
 
         :param logger: The logger class to use
@@ -18,100 +25,86 @@ def __init__(self, *, logger=fake_logger):
         self._on_shutdown = Signal(self)
         self._on_cleanup = Signal(self)
 
-        self._state = {}
-        self._loop = None
-        self._started_tasks = []
-        self.tasks = []
-        self.main_task = None
+        self._state: dict[str, Any] = {}
+        self._loop = loop
+        self._started_tasks: list[Task] = []
+        self.tasks: list[AppTask] = []
+        self.main_task: Optional[AppTask] = None
 
-    def __getitem__(self, key):
+    def __getitem__(self, key) -> Any:
         return self._state[key]
 
-    def __setitem__(self, key, value):
+    def __setitem__(self, key, value) -> None:
         self._state[key] = value
 
     @property
-    def on_startup(self):
+    def on_startup(self) -> Signal:
         return self._on_startup
 
     @property
-    def on_shutdown(self):
+    def on_shutdown(self) -> Signal:
         return self._on_shutdown
 
     @property
-    def on_cleanup(self):
+    def on_cleanup(self) -> Signal:
         return self._on_cleanup
 
-    async def startup(self):
+    async def startup(self) -> None:
         """Trigger the startup callbacks"""
         await self.on_startup.send(self)
 
-    async def shutdown(self):
+    async def shutdown(self) -> None:
         """Trigger the shutdown callbacks
 
         Call this before calling cleanup()
         """
         await self.on_shutdown.send(self)
 
-    async def cleanup(self):
+    async def cleanup(self) -> None:
         """Trigger the cleanup callbacks
 
         Calls this after calling shutdown()
         """
         await self.on_cleanup.send(self)
 
     @property
-    def loop(self):
+    def loop(self) -> AbstractEventLoop:
+        if not self._loop:
+            self._loop = asyncio.new_event_loop()
         return self._loop
 
-    @loop.setter
-    def loop(self, loop):
-        if loop is None:
-            loop = asyncio.get_event_loop()
-
-        if self._loop is not None and self._loop is not loop:
-            raise RuntimeError("Can't override event loop after init")
-
-        self._loop = loop
-
-    def start_task(self, func):
+    def start_task(self, func: AppTask) -> Task:
         """Start up a task"""
         task = self.loop.create_task(func(self))
         self._started_tasks.append(task)
 
-        def done_callback(done_task):
+        def done_callback(done_task) -> None:
             self._started_tasks.remove(done_task)
 
         task.add_done_callback(done_callback)
         return task
 
-    def run(self, loop=None):
-        """Actually run the application
-
-        :param loop: Custom event loop or None for default
-        """
-        if loop is None:
-            loop = asyncio.get_event_loop()
-
-        self.loop = loop
-
+    def run(self) -> None:
+        """ Actually run the application """
+        loop = self.loop
         loop.run_until_complete(self.startup())
 
         for func in self.tasks:
             self.start_task(func)
 
+        def shutdown_exception_handler(_loop: AbstractEventLoop, context: dict[str, Any]) -> None:
+            if "exception" not in context or not isinstance(context["exception"], CancelledError):
+                _loop.default_exception_handler(context)
+        loop.set_exception_handler(shutdown_exception_handler)
+
         try:
+            assert self.main_task
             task = self.start_task(self.main_task)
             loop.run_until_complete(task)
         except (KeyboardInterrupt, SystemError):
             print("Attempting graceful shutdown, press Ctrl-C again to exit", flush=True)
 
-            def shutdown_exception_handler(_loop, context):
-                if "exception" not in context or not isinstance(context["exception"], asyncio.CancelledError):
-                    _loop.default_exception_handler(context)
-            loop.set_exception_handler(shutdown_exception_handler)
-
-            tasks = asyncio.gather(*self._started_tasks, loop=loop, return_exceptions=True)
+            tasks = asyncio.gather(*self._started_tasks, return_exceptions=True)
             tasks.add_done_callback(lambda _: loop.stop())
             tasks.cancel()
 

aiostandalone/log.py

@@ -1,7 +1,8 @@
 """Fake logger configuration"""
+from logging import Logger
 
 
-class FakeLogger:
+class FakeLogger(Logger):
     """Fake logger that simply discards all logs"""
 
     def debug(self, msg, *args, **kwargs):
@@ -25,4 +26,4 @@ def critical(self, msg, *args, **kwargs):
         pass
 
 
-fake_logger = FakeLogger()
+fake_logger = FakeLogger("FakeLogger")

setup.cfg

@@ -0,0 +1,2 @@
+[flake8]
+max-line-length = 100

setup.py

@@ -8,8 +8,7 @@
 if os.path.exists('README.rst'):
     long_description = open('README.rst').read()
 
-install_requires = [
-]
+install_requires: list[str] = []
 
 
 tests_require = [
@@ -61,4 +60,3 @@ def run_tests(self):
         'testing': tests_require,
     },
 )
-

tests/test_app.py

@@ -33,7 +33,7 @@ async def fake_main_task(app):
 
 
 def test_run(loop):
-    app = StandaloneApplication()
+    app = StandaloneApplication(loop=loop)
 
     app.on_startup.append(fake_startup)
     app.on_shutdown.append(fake_shutdown)
@@ -50,20 +50,15 @@ def test_run(loop):
     assert app['ran_main_task']
 
 
-def test_loop():
-    loop = asyncio.get_event_loop()
-    app = StandaloneApplication()
-    app.loop = loop
-
-    assert app._loop == loop
-
-    app.loop = None
+def test_loop(loop):
+    app = StandaloneApplication(loop=loop)
+    assert app.loop == loop
 
-    with pytest.raises(RuntimeError):
-        app.loop = 23
+    app = StandaloneApplication()
+    assert app.loop and app.loop != loop
 
 
-async def fake_failing_task(_):
+async def fake_aborting_task(_):
     await asyncio.sleep(0.01)
     raise SystemError
 
@@ -73,14 +68,34 @@ async def long_running_side_task(_):
 
 
 def test_abort(loop):
-    app = StandaloneApplication()
+    app = StandaloneApplication(loop=loop)
+    app.on_startup.append(fake_startup)
+    app.on_shutdown.append(fake_shutdown)
+    app.on_cleanup.append(fake_cleanup)
+
+    app.main_task = fake_aborting_task
+    app.tasks.append(long_running_side_task)
+
+    app.run()
+    assert app['ran_shutdown']
+    assert app['ran_cleanup']
+
+
+async def fake_failing_task(_):
+    await asyncio.sleep(0.01)
+    raise RuntimeError("Ruh roh")
+
+
+def test_exception(loop):
+    app = StandaloneApplication(loop=loop)
     app.on_startup.append(fake_startup)
     app.on_shutdown.append(fake_shutdown)
     app.on_cleanup.append(fake_cleanup)
 
     app.main_task = fake_failing_task
     app.tasks.append(long_running_side_task)
+    with pytest.raises(RuntimeError, match="Ruh roh"):
+        app.run()
 
-    app.run(loop)
     assert app['ran_shutdown']
     assert app['ran_cleanup']

tests/test_logger.py

@@ -3,7 +3,7 @@
 
 
 def test_object():
-    fake = FakeLogger()
+    fake = FakeLogger("Fake")
     fake.debug("Nope")
     fake.info("Nope")
     fake.warning("Nope")

tests/test_signal.py

@@ -2,7 +2,7 @@
 from collections import defaultdict
 import pytest
 
-from aiostandalone.signal import Signal
+from aiostandalone.aiosignal import Signal
 
 
 def test_init():