Enhancement: Rich and StructLog integration

[cofin]

Create a documented example and/or integration to allow for rich traceback handling and logging handler.

Here's an example using the LoggingHandler

https://rich.readthedocs.io/en/latest/logging.html

[john-ingles]

I can get log output like this. I don't know about traceback handling though.

import logging
from starlite.logging import LoggingConfig

config = LoggingConfig(
    handlers={
        "console": {
            "class": "logging.StreamHandler",
            "level": "DEBUG",
            "formatter": "standard",
        },
        "rich": {
            "class": "rich.logging.RichHandler",
            "level": "DEBUG",
            "formatter": "standard",
        },
        "queue_listener": {
            "class": "starlite.QueueListenerHandler",
            "handlers": ["cfg://handlers.console"],
        },
    },
    loggers={"app": {"level": "INFO", "handlers": ["rich"]}},
)

config.configure()

logger = logging.getLogger("app")

logger.info("This is a test")

[john-ingles]

Maybe add "rich_tracebacks": True to the rich handler in handlers would be enough?

[john-ingles]

I got tracebacks to work by setting rich_tracebacks to True and the level to Error.

Now the question is, should this be an integration or should we just use the code above in a documented example? Any thoughts @Goldziher?

One reason why I think it won't work as an integration is that I don't think Rich works well with the queuehandler and queuelistener (at least I got thread errors when I tried adding the RichHandler to the queue_listener handlers list.)

[Goldziher]

@cofin can you review this pls?

[cofin]

I think a simple documented example would be great.

FWIW, here is something I've used with FastAPI and recently adapter for Starlite. It's been working for me in my limited testing, but i haven't done any testing with multiprocessing (or threads).

Maybe we could add some notes on when this configuration might be a problem?

class RichHandler(_RichHandler):
    """Rich Handler Config"""

    def __init__(self, *args, **kwargs) -> None:  # type: ignore
        super().__init__(
            rich_tracebacks=settings.app.LOG_LEVEL.lower() == "debug",
            console=Console(markup=True),
            markup=True,
            show_path=False,
            omit_repeated_times=False,
        )

And I used this for LoggingConfig.handlers:

handlers={
        "console": {
            "class": "app.config.logging.RichHandler",
            "level": "DEBUG",
            "formatter": "standard",
        },
        "queue_listener": {
            "class": "app.config.logging.QueueListenerHandler",
            "handlers": ["cfg://handlers.console"],
        },
    },
    ```

[cofin]

Added structlog integration to the task

[grillazz]

hi, i tested below setup which giving nice traceback:

import logging
from functools import lru_cache

from rich.console import Console
from rich.logging import RichHandler

console = Console(color_system="256", width=200, style="blue")


@lru_cache()
def get_logger(module_name):
    logger = logging.getLogger(module_name)
    handler = RichHandler(rich_tracebacks=True, console=console, tracebacks_show_locals=True)
    handler.setFormatter(logging.Formatter("[ %(threadName)s:%(funcName)s:%(lineno)d ] - %(message)s"))
    logger.addHandler(handler)
    logger.setLevel(logging.DEBUG)
    return logger

[john-ingles]

@cofin @Goldziher If the title is now "Enhancement: Rich and StructLog integration", does that mean we want to add rich and structlog to the default LoggingConfig model (along with those libraries to the starlite package, ex. starlite[rich]). For example, here is a new default __init__ file for starlite.logging

from logging import config
from typing import Any, Dict, List, Optional, Union

from pydantic import BaseModel
from typing_extensions import Literal

from starlite.logging.standard import QueueListenerHandler

__all__ = ["LoggingConfig", "QueueListenerHandler"]


class LoggingConfig(BaseModel):
    """Convenience `pydantic` model for configuring logging.

    For detailed instructions consult [standard library docs](https://docs.python.org/3/library/logging.config.html).
    """

    version: Literal[1] = 1
    """The only valid value at present is 1."""
    incremental: bool = False
    """Whether the configuration is to be interpreted as incremental to the existing configuration. """
    disable_existing_loggers: bool = False
    """Whether any existing non-root loggers are to be disabled."""
    filters: Optional[Dict[str, Dict[str, Any]]] = None
    """A dict in which each key is a filter id and each value is a dict describing how to configure the corresponding Filter instance."""
    propagate: bool = True
    """If messages must propagate to handlers higher up the logger hierarchy from this logger."""
    formatters: Dict[str, Dict[str, Any]] = {
        "standard": {
            "format": "%(levelname)s - %(asctime)s - %(name)s - %(module)s - %(message)s"
        },
        "rich": {"format": "%(name)s - %(message)s"},
    }
    handlers: Dict[str, Dict[str, Any]] = {
        "console": {
            "class": "logging.StreamHandler",
            "formatter": "standard",
        },
        "d_rich": {
            "class": "rich.logging.RichHandler",
            "rich_tracebacks": True,
            "formatter": "rich",
        },
        "queue_listener_console": {
            "class": "starlite.QueueListenerHandler",
            "handlers": ["cfg://handlers.console"],
        },
        "queue_listener_rich": {
            "class": "starlite.QueueListenerHandler",
            "handlers": ["cfg://handlers.d_rich"],
        },
    }
    """A dict in which each key is a handler id and each value is a dict describing how to configure the corresponding Handler instance."""
    loggers: Dict[str, Dict[str, Any]] = {
        "starlite": {
            "level": "INFO",
            "handlers": ["queue_listener_console"],
        },
        "rich": {
            "level": "INFO",
            "handlers": ["queue_listener_rich"],
        },
    }
    """A dict in which each key is a logger name and each value is a dict describing how to configure the corresponding Logger instance."""
    root: Dict[str, Union[Dict[str, Any], List[Any], str]] = {
        "handlers": ["queue_listener_console"],
        "level": "INFO",
    }
    """This will be the configuration for the root logger. Processing of the configuration will be as for any logger,
    except that the propagate setting will not be applicable."""

    def configure(self) -> None:
        """Configured logger with the given configuration."""
        config.dictConfig(self.dict(exclude_none=True))

Note a few gotchas:

1. The name of the rich handler is "d_rich" because python's dictConfig function sorts the handlers alphabetically to try to disambiguate them since they might refer to other handlers. In this case, naming it just "rich" causes "queue_listener_rich" to get configured first which doesn't work since it refers to the rich handler using cfg://.
2. The above code necessitates a modification to the QueueListenerHandler class in starlite.logging.standard.py. The current code starts the listener as soon as it's configured. Since there are now more than one, it would start up two listeners before the user even calls getLogger and specifies which one they want to use. The change would be to modify the QueueListenerHandler class to only start up the listener when the instance is told to.

class QueueListenerHandler(QueueHandler):
    """Configures queue listener and handler to support non-blocking logging
    configuration."""

    def __init__(self, handlers: List[Any], respect_handler_level: bool = False, queue: Queue = Queue(-1)):
        """Configures queue listener and handler to support non-blocking
        logging configuration.

        Args:
            handlers (list): list of handler names.
            respect_handler_level (bool): A handler’s level is respected (compared with the level for the message) when
                deciding whether to pass messages to that handler.
        """
        super().__init__(queue)
        self.handlers = resolve_handlers(handlers)
        self._listener: QueueListener = QueueListener(
            self.queue, *self.handlers, respect_handler_level=respect_handler_level
        )
    def start_listener(self):
        self._listener.start()

and then start the listener after getting the logger.

import logging
from starlite.logging import LoggingConfig

config = LoggingConfig()
config.configure()
logger = logging.getLogger("rich")
logger.handlers[0].start_listener()
logger.propagate = False

logger.info("1")
logger.info("2")
logger.info("3")
logger.info("4")
logger.info("5")

while True:
    pass

[09/04/22 18:52:13] INFO     rich - 1                                           test_logging.py:23
                    INFO     rich - 2                                           test_logging.py:24
                    INFO     rich - 3                                           test_logging.py:25
                    INFO     rich - 4                                           test_logging.py:26
                    INFO     rich - 5                                           test_logging.py:27

[Goldziher]

Hiya, and thanks for contributing! This is @cofin 's domain. Im fine with whatever he says.

[Swannbm]

I will instanciate logging in the middlewarelogging stuff. Great work !

[Swannbm]

I am unsure of the process, but do you think you could make a PR with your code ?

I also would suggest to change your last line from config.dictConfig(...) with logging.config.dictConfig(...). It makes clearer the use of the generic config and not the call to another starlite related config method.

Also what do you think to add the default config dict in the docstring ?

[john-ingles]

@Swannbm The last line config.dictConfig(....) is how it's written in the current __init__ file. I can change that if we are all in agreement. I'll work on putting together a PR.

My only concern is the changing of QueueListenerHandler to require the user the start the listener. It's a change that could break people's existing pipelines. Ideally, I'd like it to call itself in the situation of someone using the root logger but I haven't figured it out yet.

[john-ingles]

@Swannbm Can you explain your comment "I will instanciate logging in the middlewarelogging stuff.". I did a search of the repo for middleware and couldn't find anything related to logging. I just want to know if it's something I should do in the PR.

[Swannbm]

I am working on #411 and I want to init logging at start up (keep in mind i am quite new at working on open source software). I will open a PR soon (hopefully this morning).

[john-ingles]

@Swannbm Ok, I'll wait for your work to be finished before doing mine to ensure there is no conflict. Take you time

[Swannbm]

I've submited a PR (draft mode). Have a look and let me know.

[john-ingles]

I don't see anything in the draft that would interfere with this.

[Goldziher]

Are you adding a PR?

@cofin can you get involved here please?

[cofin]

I've been a bit MIA due to the holiday. Nice work that's been made.

What's the reason for pre-configuring all of the handlers as opposed to providing documentation on how to configure the selected on?

from logging import config
from typing import Any, Dict, List, Optional, Union

from pydantic import BaseModel
from typing_extensions import Literal

from starlite.logging.standard import QueueListenerHandler

__all__ = ["LoggingConfig", "QueueListenerHandler"]


class LoggingConfig(BaseModel):
    """Convenience `pydantic` model for configuring logging.

    For detailed instructions consult [standard library docs](https://docs.python.org/3/library/logging.config.html).
    """

    version: Literal[1] = 1
    """The only valid value at present is 1."""
    incremental: bool = False
    """Whether the configuration is to be interpreted as incremental to the existing configuration. """
    disable_existing_loggers: bool = False
    """Whether any existing non-root loggers are to be disabled."""
    filters: Optional[Dict[str, Dict[str, Any]]] = None
    """A dict in which each key is a filter id and each value is a dict describing how to configure the corresponding Filter instance."""
    propagate: bool = True
    """If messages must propagate to handlers higher up the logger hierarchy from this logger."""
    formatters: Dict[str, Dict[str, Any]] = {
        "standard": {
            "format": "%(levelname)s - %(asctime)s - %(name)s - %(module)s - %(message)s"
        },
        "rich": {"format": "%(name)s - %(message)s"},
    }
    handlers: Dict[str, Dict[str, Any]] = {
        "console": {
            "class": "logging.StreamHandler",
            "formatter": "standard",
        },
        "d_rich": {
            "class": "rich.logging.RichHandler",
            "rich_tracebacks": True,
            "formatter": "rich",
        },
        "queue_listener_console": {
            "class": "starlite.QueueListenerHandler",
            "handlers": ["cfg://handlers.console"],
        },
        "queue_listener_rich": {
            "class": "starlite.QueueListenerHandler",
            "handlers": ["cfg://handlers.d_rich"],
        },
    }
    """A dict in which each key is a handler id and each value is a dict describing how to configure the corresponding Handler instance."""
    loggers: Dict[str, Dict[str, Any]] = {
        "starlite": {
            "level": "INFO",
            "handlers": ["queue_listener_console"],
        },
        "rich": {
            "level": "INFO",
            "handlers": ["queue_listener_rich"],
        },
    }
    """A dict in which each key is a logger name and each value is a dict describing how to configure the corresponding Logger instance."""
    root: Dict[str, Union[Dict[str, Any], List[Any], str]] = {
        "handlers": ["queue_listener_console"],
        "level": "INFO",
    }
    """This will be the configuration for the root logger. Processing of the configuration will be as for any logger,
    except that the propagate setting will not be applicable."""

    def configure(self) -> None:
        """Configured logger with the given configuration."""
        config.dictConfig(self.dict(exclude_none=True))

[cofin]

Also, if the user is using Picologging and Rich, then there needs to be some additional work done. For instance, here is a quick implementation of the RichHandler in picologging:

https://github.com/cofin/starlite-bedrock/blob/main/src/starlite_bedrock/starlite/logging.py#L20

Here is how I am using that with the existing log config:

log_config = LoggingConfig(
    filters={
        "health_filter": {
            "()": AccessLogFilter,
            "path_re": "^/health$",
        }
    },
    handlers={
        "console": {
            "class": "starlite_bedrock.starlite.logging.RichPicologgingHandler",
            "formatter": "standard",
        },
        "queue_listener": {
            "class": "starlite.logging.picologging.QueueListenerHandler",
            "handlers": ["cfg://handlers.console"],
        },
    },
    formatters={"standard": {"format": "[%(name)s][%(funcName)s]   %(message)s"}},
    loggers={
        "starlite_bedrock": {
            "propagate": True,
        },
        "uvicorn.access": {
            "propagate": True,
            "filters": ["health_filter"],
        },
        "uvicorn.error": {
            "propagate": True,
        },
        "gunicorn.access": {
            "propagate": True,
            "filters": ["health_filter"],
        },
        "gunicorn.error": {
            "propagate": True,
        },
        "sqlalchemy": {
            "propagate": True,
        },
        "starlite": {
            "level": "WARNING",
            "propagate": True,
        },
        "pydantic_openapi_schema": {
            "propagate": True,
            "level": "WARNING",
            "handlers": ["queue_listener"],
        },
    },
)

EDIT: OK, so maybe something like below doesn't work. I also believe there is a way to pass commands into the configuration by using the args attribute. I haven't tested this, but I think it's something like the following:

handlers={
        "console": {
            "class": "starlite_bedrock.starlite.logging.RichPicologgingHandler",
            "formatter": "standard",
            "args": { "rich_tracebacks": True}
        },
        "queue_listener": {
            "class": "starlite.logging.picologging.QueueListenerHandler",
            "handlers": ["cfg://handlers.console"],
        },
    },

[john-ingles]

I was thinking there could be presets for the user to select in addition to allowing them to pass their own configuration. This is what I think of when I here integration. Maybe this is off the mark and it would be better just to provide documentation on how to load your own config. It would avoid having to add any other dependencies that the user might not use. @Swannbm wants the logger config to be passed in as an init variable to app.py Starlite(Router) and this seems sensible. It would probably also be best to decouple the default config from the starlite logging init and put it in a new place like starlite/config/logging.py.

[cofin]

I like the idea of adding in the LoggingConfig as parameter to the app, and I also like to the idea of easy log configurations (i.e. zero config color logging similar to loguru would be nice). I think those should definitely be included. I also think we need to include the RichPicologgingHandler referenced above.

My main questions is around the default logging implementations. Based on the example above, we'd also need to manage picologging defaults in there.

Instead of adding these additional configurations, could we not enhance the LoggingConfig.configure method to dynamically start the additional listeners if found in config?

[john-ingles]

Those all sound like good ideas. The configure method can look at the various handlers defined in the config and recognize if they are installed or not and start them if they are or remove them if they aren't and give a warning.

[john-ingles]

The handlers would be added manually to the config and passed in on startup, or they could be added by starlite if the user passes various enum flags somewhere.

[john-ingles]

I'm going to try to get some work done on it this week or perhaps the weekend.

[cofin]

Excellent. Sounds good.

I did some testing today. Here is how I was able to dynamically create a RichPicologgingHandler with the LoggingConfig class. You should also be able to use the syntax used on health_filter. path_re is the parameter that AccessLogFilter expects.

I also believe that this section in the logging docs could be helpful.

log_config = LoggingConfig(
    filters={
        "health_filter": {
            "()": AccessLogFilter,
            "path_re": "^/health$",
        }
    },
    handlers={
        "console": {
            "()": lambda: RichPicologgingHandler(markup=True, rich_tracebacks=True),
        },
        "queue_listener": {
            "class": "starlite.logging.picologging.QueueListenerHandler",
            "handlers": ["cfg://handlers.console"],
        },
    },
    formatters={
        "standard": {"format": "%(message)s"}
    },
    loggers={
        "": {
            "level": "INFO",
            "propagate": True,
            "handlers": ["queue_listener"],
        },
        "starlite": {
            "level": "WARNING",
            "propagate": True,
        },
        "pydantic_openapi_schema": {
            "propagate": True,
            "level": "WARNING",
        },
    },
)
log_config.configure()

[john-ingles]

I was able to add parameters to the handler class by just added extra arguments to the handler dict. As in this example with rich_tracebacks.

handlers: Dict[str, Dict[str, Any]] = {
        "console": {
            "class": "logging.StreamHandler",
            "formatter": "standard",
        },
        "d_rich": {
            "class": "rich.logging.RichHandler",
            "rich_tracebacks": True,
            "formatter": "rich",
        },
        "queue_listener_console": {
            "class": "starlite.QueueListenerHandler",
            "handlers": ["cfg://handlers.console"],
        },
        "queue_listener_rich": {
            "class": "starlite.QueueListenerHandler",
            "handlers": ["cfg://handlers.d_rich"],
        },
    }

As far as I know this should work for any class init variables

[cofin]

Well, I was certainly overcomplicating it!

[john-ingles]

Just made a PR for decoupling logging config from the logging module. #468

[john-ingles]

I think next step for me would be to choose where in the program to call

logging = LoggingConfig()
logging.configure()

Should it be an instance arg in the starlite.app.Starlite class with default LoggingConfig()? That's what I assume anyway but let me know if it shouldn't be. So, something like this

class Starlite(Router):
    __slots__ = (
        ...,
        "logging_config",
        ...
    )

    @validate_arguments(config={"arbitrary_types_allowed": True})
    def __init__(
        self,
        ...,
        logging_config: Opitonal[Dict[str,Any]] = None
        ...,
    ):
        """The Starlite application.
        `Starlite` is the root level of the app - it has the base path of "/" and all root level
        Controllers, Routers and Route Handlers should be registered on it.
        It inherits from the [Router][starlite.router.Router] class.
        Args:
            ...
            logging_config: A dict of blah blah blah
            ...
        """

        ...
        self.logging_config = logging_config
        ...

       #somewhere in the main body
       def _configure_logging(self, config):
           if config:
               logger_config = LoggingConfig(config)
           else:
               logger_config = LoggingConfig(self.logging_config)
           logger_config.configure()

       def configure_logging(self, config: Optional[Dict[str,Any]] = None):
           _configure_logging(config)
       
       configure_logging()

or something like that. I want to leave space for someone to come in after app is created and call app.configure_logger() in case they don't want to pass it in the constructor/want to change it later during runtime. Is that possible (I'm not sure)