From a1e826c7de6b6cef3c032ba1bc6ee2d56aaca6cf Mon Sep 17 00:00:00 2001 From: Bedram Tamang Date: Tue, 22 Sep 2026 12:03:28 -0700 Subject: [PATCH 1/2] fix(logging): resolve basedpyright errors in log channels Declare the `driver`/`max_level` attributes on BaseChannel via a LogDriver protocol and `channels` on MultiBaseChannel, and resolve driver/channel classes through explicit None-checks instead of calling an Optional. Co-Authored-By: Claude Opus 5.5 --- .../logging/channels/BaseChannel.py | 32 ++++++++++++++++++- .../logging/channels/DailyChannel.py | 3 +- .../logging/channels/MultiBaseChannel.py | 4 +++ .../logging/channels/SingleChannel.py | 3 +- .../logging/channels/SlackChannel.py | 5 +-- .../logging/channels/StackChannel.py | 5 +-- .../logging/channels/SyslogChannel.py | 3 +- .../logging/channels/TerminalChannel.py | 3 +- 8 files changed, 43 insertions(+), 15 deletions(-) diff --git a/fastapi_startkit/src/fastapi_startkit/logging/channels/BaseChannel.py b/fastapi_startkit/src/fastapi_startkit/logging/channels/BaseChannel.py index eb47b21f..83d43f3f 100644 --- a/fastapi_startkit/src/fastapi_startkit/logging/channels/BaseChannel.py +++ b/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() diff --git a/fastapi_startkit/src/fastapi_startkit/logging/channels/DailyChannel.py b/fastapi_startkit/src/fastapi_startkit/logging/channels/DailyChannel.py index 06bb6669..475a6400 100644 --- a/fastapi_startkit/src/fastapi_startkit/logging/channels/DailyChannel.py +++ b/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 ) diff --git a/fastapi_startkit/src/fastapi_startkit/logging/channels/MultiBaseChannel.py b/fastapi_startkit/src/fastapi_startkit/logging/channels/MultiBaseChannel.py index 2be7bba4..104d1ac1 100644 --- a/fastapi_startkit/src/fastapi_startkit/logging/channels/MultiBaseChannel.py +++ b/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")) diff --git a/fastapi_startkit/src/fastapi_startkit/logging/channels/SingleChannel.py b/fastapi_startkit/src/fastapi_startkit/logging/channels/SingleChannel.py index bb3667e5..e4fdbba9 100644 --- a/fastapi_startkit/src/fastapi_startkit/logging/channels/SingleChannel.py +++ b/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 ) diff --git a/fastapi_startkit/src/fastapi_startkit/logging/channels/SlackChannel.py b/fastapi_startkit/src/fastapi_startkit/logging/channels/SlackChannel.py index c329a663..c2d2a596 100644 --- a/fastapi_startkit/src/fastapi_startkit/logging/channels/SlackChannel.py +++ b/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 ) diff --git a/fastapi_startkit/src/fastapi_startkit/logging/channels/StackChannel.py b/fastapi_startkit/src/fastapi_startkit/logging/channels/StackChannel.py index a109bdfb..3141c85f 100644 --- a/fastapi_startkit/src/fastapi_startkit/logging/channels/StackChannel.py +++ b/fastapi_startkit/src/fastapi_startkit/logging/channels/StackChannel.py @@ -1,3 +1,4 @@ +from .BaseChannel import BaseChannel from .MultiBaseChannel import MultiBaseChannel @@ -5,10 +6,10 @@ 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: diff --git a/fastapi_startkit/src/fastapi_startkit/logging/channels/SyslogChannel.py b/fastapi_startkit/src/fastapi_startkit/logging/channels/SyslogChannel.py index c699c642..489588e1 100644 --- a/fastapi_startkit/src/fastapi_startkit/logging/channels/SyslogChannel.py +++ b/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 ) diff --git a/fastapi_startkit/src/fastapi_startkit/logging/channels/TerminalChannel.py b/fastapi_startkit/src/fastapi_startkit/logging/channels/TerminalChannel.py index 40fa5ea7..6fc79174 100644 --- a/fastapi_startkit/src/fastapi_startkit/logging/channels/TerminalChannel.py +++ b/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 ) From 111c5d765d36cc745b270e43c012961b1379114c Mon Sep 17 00:00:00 2001 From: Bedram Tamang Date: Tue, 22 Sep 2026 12:12:09 -0700 Subject: [PATCH 2/2] test(logging): cover unknown driver/channel errors and slack/syslog channels Co-Authored-By: Claude Opus 5.5 --- .../tests/logging/test_logging.py | 28 +++++++++++++++++++ 1 file changed, 28 insertions(+) diff --git a/fastapi_startkit/tests/logging/test_logging.py b/fastapi_startkit/tests/logging/test_logging.py index 3fa89272..095634e9 100644 --- a/fastapi_startkit/tests/logging/test_logging.py +++ b/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)