How to send logger information to seq

Question

I sent my logger information into 'seq' module. I have:

_log_format = f"%(asctime)s - [%(levelname)s] - request_id=%(request_id)s - %(name)s - (%(filename)s).%(funcName)s(%(lineno)d) - %(message)s"
logger.info("Hello, {name}", name="world")

As result in 'seq' I have only:

No one from: f"%(asctime)s - [%(levelname)s] - request_id=%(request_id)s - %(name)s - (%(filename)s).%(funcName)s(%(lineno)d) - %(message)s" was add into 'seq'.

In stream is all right:

2021-08-20 14:40:24,244 - [INFO] - request_id=None - app - (__init__.py).create_app(43) - Hello, world

I sent logs to 'seq' by this way:

import seqlog

seqlog.log_to_seq(
   server_url="http://localhost:5341/",
   api_key="My API Key",
   level=logging.INFO,
   batch_size=10,
   auto_flush_timeout=10,  # seconds
   override_root_logger=True,
)

Answer 1

Update your file like this

import logging
import seqlog

# Set up Seq logging
seqlog.log_to_seq(
    server_url="http://localhost:5341/",
    api_key="My API Key",
    level=logging.INFO,
    batch_size=10,
    auto_flush_timeout=10,  # seconds
    override_root_logger=True,
)

# Create a logger with the Seq JSON formatter
logger = seqlog.log
seqlog.configure_json(
    logger,
    json_encoder=seqlog.JsonEncoder,
    level_names=seqlog.DEFAULT_LEVEL_NAMES,
    extra_fields=[
        ("request_id", "r_id"),
        ("name", "n"),
    ],
)

# Log a message
logger.info("Hello, {name}", name="world")

Answer 2

I use this one:

import functools
import inspect
import logging
import socket
from pathlib import Path

import seqlog
from flask import current_app
from flask_log_request_id import current_request_id


def send_log_to_seq(msg, properties=None, level="info", **kwargs):
    if properties is None:
        properties = {}
    request_id = current_request_id()
    func_name = inspect.getframeinfo(inspect.currentframe()
                                     .f_back).function
    parts = Path(inspect.currentframe()
                 .f_back.f_code.co_filename).with_suffix("").parts
    module_name = ".".join(parts[-3:])
    lineno = inspect.currentframe().f_back.f_lineno
    properties['AssemblyName'] = "STAT_API"
    properties['MachineName'] = socket.gethostname()
    properties['current_request_id'] = request_id
    properties['func_name'] = func_name
    properties['module_name'] = module_name
    properties['lineno'] = lineno
    seqlog.set_global_log_properties(**properties)
    if level == "info":
        current_app.logger.info(msg, **kwargs)
    if level == "debug":
        current_app.logger.debug(msg, **kwargs)
    if level == "error":
        current_app.logger.error(msg, **kwargs)
    if level == "warning":
        current_app.logger.warning(msg, **kwargs)
    if level == "critical":
        current_app.logger.critical(msg, **kwargs)


def log_errors(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception as err:
            # logging.error(err.args[0], exc_info=True)
            send_log_to_seq(
                err.args[0], exc_info=True, level="error")
            return {'error': 'System error'}
    return wrapper

Config on start application:

seq_log_config = app.config['SEQ_LOG_CONF']
seqlog.log_to_seq(
    server_url=seq_log_config['server_url'],
    api_key=None,
    level=seq_log_config['level'],
    batch_size=seq_log_config['batch_size'],
    auto_flush_timeout=seq_log_config['auto_flush_timeout'],  # seconds
    override_root_logger=seq_log_config['override_root_logger'],
    json_encoder_class=app.json_encoder, )