Add log-level option to the cli app (#22)

Makefile

@@ -20,7 +20,7 @@ rebuild-lockfiles: .pdm
 
 .PHONY: format  ## Auto-format python source files
 format: .pdm
-	pdm run ruff --fix $(sources)
+	pdm run ruff check --fix $(sources)
 	pdm run ruff format $(sources)
 
 .PHONY: lint  ## Lint python source files

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "message-flow"
-version = "0.3.2"
+version = "0.3.3"
 description = "Asynchronous Communication Framework"
 authors = [
     {name = "Valentin Vorobyev", email = "[email protected]"},
@@ -18,8 +18,8 @@ requires = ["pdm-backend"]
 build-backend = "pdm.backend"
 
 [project.optional-dependencies]
-reload = [
-    "watchfiles>=0.21.0",
+rabbitmq = [
+    "message-flow-rabbitmq>=0.2.2",
 ]
 
 
@@ -56,23 +56,23 @@ testing = [
 
 [tool.ruff]
 line-length = 120
-extend-select = ['Q', 'RUF100', 'C90', 'UP', 'I', 'T']
-extend-ignore = ['D105', 'D107', 'D205', 'D415']
-flake8-quotes = {inline-quotes = 'double', multiline-quotes = 'double'}
-mccabe = { max-complexity = 14 }
-isort = { known-first-party = ['message_flow', 'tests'] }
+lint.extend-select = ['Q', 'RUF100', 'C90', 'UP', 'I', 'T']
+lint.extend-ignore = ['D105', 'D107', 'D205', 'D415']
+lint.flake8-quotes = {inline-quotes = 'double', multiline-quotes = 'double'}
+lint.mccabe = { max-complexity = 14 }
+lint.isort = { known-first-party = ['message_flow', 'tests'] }
 target-version = "py37"
 
-[tool.ruff.extend-per-file-ignores]
+[tool.ruff.lint.extend-per-file-ignores]
 "tests/**/*.py" = ['T', 'E721', 'F811']
 
 [tool.ruff.format]
 quote-style = 'double'
 
-[tool.ruff.pydocstyle]
+[tool.ruff.lint.pydocstyle]
 convention = "google"
 
-[tool.ruff.per-file-ignores]
+[tool.ruff.lint.per-file-ignores]
 '*/__init__.py' = ['F405', 'F403', 'D']
 '*/__init__.pyi' = ['F405', 'F403', 'D']
 'tests/*' = ['D']

src/message_flow/__init__.py

@@ -1,14 +1,5 @@
-import logging
-
 from .app import *
 from .channel import *
 from .message import *
 
 __all__ = app.__all__ + channel.__all__ + message.__all__
-
-logger = logging.getLogger(__name__)
-
-ch = logging.StreamHandler()
-formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
-ch.setFormatter(formatter)
-logger.addHandler(ch)

src/message_flow/app/_message_management/dispatcher.py

@@ -11,20 +11,22 @@
 @final
 @internal
 class Dispatcher:
-    def __init__(self, channels: Channels, message_consumer: MessageConsumer, producer: Producer) -> None:
-        self._logger = logging.getLogger(__name__)
+    def __init__(
+        self, channels: Channels, message_consumer: MessageConsumer, producer: Producer, logger: logging.Logger
+    ) -> None:
+        self._logger = logger
 
         self._channels = channels
         self._message_consumer = message_consumer
         self._producer = producer
 
     def initialize(self) -> None:
-        self._logger.info("Initializing dispatcher")
+        self._logger.debug("Initializing dispatcher")
         self._message_consumer.subscribe(
             self._channels.addresses,
             self.message_handler,
         )
-        self._logger.info("Initialized dispatcher")
+        self._logger.debug("Initialized dispatcher")
 
     def message_handler(self, payload: bytes, headers: dict[str, str]) -> None:
         if (

src/message_flow/app/_simple_messaging/simple_consumer.py

@@ -11,9 +11,13 @@
 @internal
 class SimpleMessageConsumer(MessageConsumer):
     def __init__(
-        self, file_path: str = "/tmp/message-flow-queue.txt", dry_run: bool = False, throw_error: bool = False
+        self,
+        logger: logging.Logger,
+        file_path: str = "/tmp/message-flow-queue.txt",
+        dry_run: bool = False,
+        throw_error: bool = False,
     ) -> None:
-        self._logger = logging.getLogger(__name__)
+        self._logger = logger
 
         self._dry_run = dry_run
         self._throw_error = throw_error
@@ -60,10 +64,9 @@ def _process_message(self, message: str | None) -> None:
             self._logger.debug("Got message empty message. Start sleeping...")
             time.sleep(1)
         except Exception as error:
-            self._logger.info("An error occurred while consuming events", exc_info=error)
+            self._logger.debug("An error occurred while consuming events", exc_info=error)
         finally:
             self._commit_message(message)
-            self._logger.debug("Message %s is committed.")
 
     def _handle_message(self, message: str) -> None:
         channel, payload, headers = self._parse_message(message)
@@ -83,3 +86,4 @@ def _commit_message(self, message: str | None) -> None:
         if message is not None:
             self._position += len(message)
             self._fp.seek(self._position)
+            self._logger.debug("Message %s is committed.", message)

src/message_flow/app/_simple_messaging/simple_producer.py

@@ -9,15 +9,15 @@
 @final
 @internal
 class SimpleMessageProducer(MessageProducer):
-    def __init__(self, file_path: str = "/tmp/message-flow-queue.txt") -> None:
-        self._logger = logging.getLogger(__name__)
+    def __init__(self, logger: logging.Logger, file_path: str = "/tmp/message-flow-queue.txt") -> None:
+        self._logger = logger
 
         self.closed = False
 
         self._file_path = file_path
 
     def send(self, channel: str, payload: bytes, headers: dict[str, str] | None = None) -> None:
-        self._logger.info(f"Send message to {channel}")
+        self._logger.debug(f"Send message to {channel}")
         with open(self._file_path, "a+") as fp:
             fp.write(f"{channel}\t{payload.decode()}\t{json.dumps(headers)}\n")
 

src/message_flow/app/message_flow.py

@@ -6,7 +6,7 @@
 
 from ..channel import Channel
 from ..message import Message
-from ..utils import external
+from ..utils import external, logger
 from ._internal import Channels, Info, MessageFlowSchema
 from ._message_management import Dispatcher, Producer
 from ._simple_messaging import SimpleMessageConsumer, SimpleMessageProducer
@@ -127,8 +127,27 @@ def __init__(
                 """
             ),
         ] = "0.1.0",
+        logger: Annotated[
+            logging.Logger,
+            Doc(
+                """
+                The custom logger for the application.
+
+                **Example**
+
+                ```python
+                import logging
+                from message_flow import MessageFlow
+
+                custom_logger = logging.getLogger("custom_logger")
+
+                app = MessageFlow(logger=custom_logger)
+                ```
+                """
+            ),
+        ] = logger,
     ) -> None:
-        self._logger = logging.getLogger(__name__)
+        self._logger = logger
 
         self.asyncapi_version: Annotated[
             str,
@@ -160,8 +179,8 @@ def __init__(
         self.version = version
 
         self._channels = Channels(channels=channels)
-        self._message_producer = message_producer or SimpleMessageProducer()
-        self._message_consumer = message_consumer or SimpleMessageConsumer()
+        self._message_producer = message_producer or SimpleMessageProducer(self._logger)
+        self._message_consumer = message_consumer or SimpleMessageConsumer(self._logger)
 
     @property
     def producer(self) -> Producer:
@@ -172,7 +191,7 @@ def producer(self) -> Producer:
     @property
     def dispatcher(self) -> Dispatcher:
         if not hasattr(self, "_dispatcher"):
-            self._dispatcher = Dispatcher(self._channels, self._message_consumer, self.producer)
+            self._dispatcher = Dispatcher(self._channels, self._message_consumer, self.producer, self._logger)
         return self._dispatcher
 
     def add_channel(self, channel: Annotated[Channel, Doc("The channel to add.")]) -> None:
@@ -379,6 +398,7 @@ def dispatch(self) -> None:
         """
         try:
             self.dispatcher.initialize()
+            self._logger.info("Message Flow app starting...")
             self._message_consumer.start_consuming()
         except Exception as error:
             self._logger.error("An error occurred while dispatching events", exc_info=error)
@@ -403,3 +423,9 @@ def make_async_api_schema(self) -> str:
         )
 
         return json.dumps(schema)
+
+    def set_logging_level(self, level: int) -> None:
+        """
+        Set logging level for the app logger.
+        """
+        self._logger.setLevel(level=level)

src/message_flow/cli/cli_app.py

@@ -1,18 +1,33 @@
+import logging
+from collections import defaultdict
 from importlib.util import module_from_spec, spec_from_file_location
 from pathlib import Path
-from typing import Any
+from typing import DefaultDict
 
 import typer
 
+from ..app import MessageFlow
 from ..utils import internal
+from .logging_level import LoggingLevel
 
 
 @internal
 class CLIApp:
-    def __init__(self, app_path: str, reload: bool = False) -> None:
+    LOGGING_LEVELS: DefaultDict[str, int] = defaultdict(
+        lambda: logging.INFO,
+        **{
+            LoggingLevel.CRITICAL: logging.CRITICAL,
+            LoggingLevel.ERROR: logging.ERROR,
+            LoggingLevel.WARNING: logging.WARNING,
+            LoggingLevel.INFO: logging.INFO,
+            LoggingLevel.DEBUG: logging.DEBUG,
+        },
+    )
+
+    def __init__(self, app_path: str, log_level: LoggingLevel) -> None:
         self.app_path = app_path
 
-        self.reload = reload
+        self.instance.set_logging_level(self.LOGGING_LEVELS[log_level])
 
     @property
     def app_path(self) -> str:
@@ -49,19 +64,20 @@ def app_name(self) -> str:
         return self._app_name
 
     @property
-    def instance(self) -> Any:
-        try:
-            app_object = self._import()
-        except FileNotFoundError as e:
-            typer.echo(e, err=True)
-            raise typer.BadParameter("Please, input module like [python_module:message_flow_app_name]") from e
-        else:
-            return app_object  # type: ignore
+    def instance(self) -> MessageFlow:
+        if not hasattr(self, "_instance"):
+            try:
+                self._instance = self._import()
+            except FileNotFoundError as e:
+                typer.echo(e, err=True)
+                raise typer.BadParameter("Please, input module like [python_module:message_flow_app_name]") from e
+
+        return self._instance
 
     def dispatch(self) -> None:
         self.instance.dispatch()
 
-    def _import(self) -> Any:
+    def _import(self) -> MessageFlow:
         spec = spec_from_file_location(
             "mode",
             f"{self.module_path}.py",

src/message_flow/cli/logging_level.py

@@ -0,0 +1,12 @@
+from enum import Enum
+
+from ..utils import internal
+
+
+@internal
+class LoggingLevel(str, Enum):
+    CRITICAL = "critical"
+    ERROR = "error"
+    WARNING = "warning"
+    INFO = "info"
+    DEBUG = "debug"

src/message_flow/cli/main.py

@@ -1,6 +1,7 @@
 import typer
 
 from .cli_app import CLIApp
+from .logging_level import LoggingLevel
 
 __all__ = ["cli"]
 
@@ -20,16 +21,16 @@ def dispatch(
         ...,
         help="[python_module:MessageFlow] - path to your application",
     ),
-    reload: bool = typer.Option(
-        False,
-        "--reload",
-        is_flag=True,
-        help="Restart app at directory files changes",
+    log_level: LoggingLevel = typer.Option(
+        LoggingLevel.INFO,
+        case_sensitive=False,
+        show_default=False,
+        help="[INFO] default",
     ),
 ):
     """
     Shoot the portal gun
     """
-    cli_app = CLIApp(app, reload)
+    cli_app = CLIApp(app, log_level)
 
     cli_app.dispatch()

src/message_flow/utils/__init__.py

@@ -1,3 +1,4 @@
+from .log import *
 from .visibility import *
 
-__all__ = visibility.__all__
+__all__ = visibility.__all__ + log.__all__