fix(logging): resolve basedpyright errors in log channels

fastapi_startkit/src/fastapi_startkit/logging/channels/BaseChannel.py

@@ -1,8 +1,35 @@
+from typing import Any, Callable, Protocol
+
 import pendulum
 from fastapi_startkit.facades import Config
 
+from ..factory import DriverFactory
+
+
+class LogDriver(Protocol):
+    emergency: Callable[..., Any]
+    alert: Callable[..., Any]
+    critical: Callable[..., Any]
+    error: Callable[..., Any]
+    warning: Callable[..., Any]
+    notice: Callable[..., Any]
+    info: Callable[..., Any]
+    debug: Callable[..., Any]
+
+    def should_run(self, level: str, max_level: str | None) -> bool: ...
+
 
 class BaseChannel:
+    driver: LogDriver
+    max_level: str | None
+
+    @staticmethod
+    def driver_class(driver: str | None) -> Callable[..., LogDriver]:
+        driver_class = DriverFactory.make(driver)
+        if driver_class is None:
+            raise ValueError(f"Unknown log driver: {driver!r}")
+        return driver_class
+
     def get_time(self):
         return pendulum.now().in_tz(Config.get("logging.channels.timezone", "UTC"))
 
@@ -60,4 +87,7 @@ def debug(self, message, *args, **kwargs):
     def channel(self, channel):
         from ..ChannelFactory import ChannelFactory
 
-        return ChannelFactory().make(channel)()
+        channel_class = ChannelFactory.make(channel)
+        if channel_class is None:
+            raise ValueError(f"Unknown log channel: {channel!r}")
+        return channel_class()

fastapi_startkit/src/fastapi_startkit/logging/channels/DailyChannel.py

@@ -2,7 +2,6 @@
 
 from fastapi_startkit.facades import Config
 from .BaseChannel import BaseChannel
-from ..factory import DriverFactory
 from ..file import make_directory
 
 
@@ -12,7 +11,7 @@ def __init__(self, driver=None, path=None):
         path = os.path.join(path, self.get_time().to_date_string() + ".log")
         self.max_level = Config.get("logging.channels.daily.level")
         make_directory(path)
-        self.driver = DriverFactory.make(driver or Config.get("logging.channels.daily.driver"))(
+        self.driver = self.driver_class(driver or Config.get("logging.channels.daily.driver"))(
             path=path, max_level=self.max_level
         )
 

fastapi_startkit/src/fastapi_startkit/logging/channels/MultiBaseChannel.py

@@ -1,8 +1,12 @@
 import pendulum
 from fastapi_startkit.facades import Config
 
+from .BaseChannel import BaseChannel
+
 
 class MultiBaseChannel:
+    channels: list[BaseChannel]
+
     def get_time(self):
         return pendulum.now().in_tz(Config.get("logging.channels.timezone", "UTC"))
 

fastapi_startkit/src/fastapi_startkit/logging/channels/SingleChannel.py

@@ -1,4 +1,3 @@
-from ..factory import DriverFactory
 from fastapi_startkit.facades import Config
 from ..file import make_directory
 from .BaseChannel import BaseChannel
@@ -9,6 +8,6 @@ def __init__(self, driver=None, path=None):
         path = path or Config.get("logging.channels.single.path")
         make_directory(path)
         self.max_level = Config.get("logging.channels.single.level")
-        self.driver = DriverFactory.make(driver or Config.get("logging.channels.single.driver"))(
+        self.driver = self.driver_class(driver or Config.get("logging.channels.single.driver"))(
             path=path, max_level=self.max_level
         )

fastapi_startkit/src/fastapi_startkit/logging/channels/SlackChannel.py

@@ -1,7 +1,4 @@
 from fastapi_startkit.facades import Config
-from ..factory import DriverFactory
-
-from ..factory import DriverFactory
 from .BaseChannel import BaseChannel
 
 
@@ -12,7 +9,7 @@ def __init__(self, driver=None, path=None):
         emoji = Config.get("logging.channels.slack.emoji")
         username = Config.get("logging.channels.slack.username")
         self.max_level = Config.get("logging.channels.slack.level")
-        self.driver = DriverFactory.make(driver or Config.get("logging.channels.slack.driver"))(
+        self.driver = self.driver_class(driver or Config.get("logging.channels.slack.driver"))(
             emoji=emoji, username=username, token=token, channel=channel
         )
 

fastapi_startkit/src/fastapi_startkit/logging/channels/StackChannel.py

@@ -1,14 +1,15 @@
+from .BaseChannel import BaseChannel
 from .MultiBaseChannel import MultiBaseChannel
 
 
 class StackChannel(MultiBaseChannel):
     def __init__(self, channels=None):
         from fastapi_startkit.facades import Config
 
-        channels = channels or Config.get("logging.channels.stack.channels", [])
+        channels = channels or Config.get("logging.channels.stack.channels") or []
         from ..ChannelFactory import ChannelFactory
 
-        self.channels = []
+        self.channels: list[BaseChannel] = []
         for channel in channels:
             channel_class = ChannelFactory.make(channel)
             if channel_class:

fastapi_startkit/src/fastapi_startkit/logging/channels/SyslogChannel.py

@@ -1,4 +1,3 @@
-from ..factory import DriverFactory
 from fastapi_startkit.facades import Config
 from ..file import make_directory
 from .BaseChannel import BaseChannel
@@ -9,6 +8,6 @@ def __init__(self, driver=None, path=None):
         path = path or Config.get("logging.channels.syslog.path")
         make_directory(path)
         self.max_level = Config.get("logging.channels.syslog.level")
-        self.driver = DriverFactory.make(driver or Config.get("logging.channels.syslog.driver"))(
+        self.driver = self.driver_class(driver or Config.get("logging.channels.syslog.driver"))(
             path=path, max_level=self.max_level
         )

fastapi_startkit/src/fastapi_startkit/logging/channels/TerminalChannel.py

@@ -1,12 +1,11 @@
 from fastapi_startkit.facades import Config
 
-from ..factory import DriverFactory
 from ..channels.BaseChannel import BaseChannel
 
 
 class TerminalChannel(BaseChannel):
     def __init__(self, driver=None, path=None):
         self.max_level = Config.get("logging.channels.terminal.level", "debug")
-        self.driver = DriverFactory.make(driver or Config.get("logging.channels.terminal.driver"))(
+        self.driver = self.driver_class(driver or Config.get("logging.channels.terminal.driver"))(
             path=path, max_level=self.max_level
         )

fastapi_startkit/tests/logging/test_logging.py

@@ -9,7 +9,9 @@
 from fastapi_startkit.logging.channels import (
     DailyChannel,
     SingleChannel,
+    SlackChannel,
     StackChannel,
+    SyslogChannel,
     TerminalChannel,
 )
 from fastapi_startkit.logging.channels.BaseChannel import BaseChannel
@@ -174,6 +176,14 @@ def test_channel_builds_a_new_channel_instance(self):
         channel = self._channel()
         self.assertIsInstance(channel.channel("terminal"), TerminalChannel)
 
+    def test_channel_raises_for_unknown_channel(self):
+        with self.assertRaisesRegex(ValueError, "Unknown log channel"):
+            self._channel().channel("does-not-exist")
+
+    def test_driver_class_raises_for_unknown_driver(self):
+        with self.assertRaisesRegex(ValueError, "Unknown log driver"):
+            BaseChannel.driver_class("does-not-exist")
+
 
 class MultiBaseChannelTest(unittest.TestCase):
     def _multi(self, should_run=True):
@@ -434,6 +444,24 @@ def test_daily_channel_writes_dated_file(self):
                     channel.driver.log.removeHandler(handler)
                     handler.close()
 
+    def test_slack_channel_builds_slack_driver(self):
+        channel = SlackChannel(driver="slack")
+        self.assertIsInstance(channel.driver, LogSlackDriver)
+
+    def test_syslog_channel_builds_syslog_driver(self):
+        import os
+        import tempfile
+
+        path = os.path.join(tempfile.mkdtemp(), "syslog")
+        root = logging.getLogger("root")
+        with patch("fastapi_startkit.logging.drivers.LogSyslogDriver.logging.handlers.SysLogHandler") as handler_cls:
+            channel = SyslogChannel(driver="syslog", path=path)
+        try:
+            self.assertIsInstance(channel.driver, LogSyslogDriver)
+            handler_cls.assert_called_once_with(address=path)
+        finally:
+            root.removeHandler(handler_cls.return_value)
+
     def test_stack_channel_collects_known_channels(self):
         channel = StackChannel(channels=["terminal"])
         self.assertEqual(len(channel.channels), 1)