Module panama.logging.logger

Classes

class PanamaAdapter (logger: logging.Logger,
config_run: ConfigRun)

Expand source code

class PanamaAdapter(logging.LoggerAdapter):
    """Panama adapter to add contextual information and create the methods start_task, end_task and log_step."""

    def __init__(self, logger: logging.Logger, config_run: ConfigRun):
        """Initialize adapter.

        Args:
            logger (logging.Logger): underlying logger object.
            config_run (ConfigRun): config run, with default data.
        """
        self.config_run = config_run
        super().__init__(logger=logger, extra=config_run.widgets)

    def process(self, msg, kwargs):
        """Add to extra the required information. Deals with trace parameter.

        Args:
            msg (str): logged message content.
            kwargs (dict): log call arguments.

        Returns:
            (str, dict): msg and kwargs handled.
        """
        forbidden = ["exc_info", "stack_info", "logger"]
        extra = {k: str(v) for k, v in kwargs.items() if k not in forbidden}
        # add trace kwarg to extra
        if isinstance(kwargs.get("exc_info"), str):
            extra["trace"] = kwargs["exc_info"]
        kwargs["extra"] = extra
        # remove trace from kwargs, as it has been added to extras
        kwargs.pop("trace", None)
        return msg, kwargs

    def move_json_to_remote(self):
        """Method used to move logs to remote storage."""
        # get task name from config_run
        task_name = self.config_run.task_name
        # get logger_dict from manager
        logger_dict = self.logger.manager.loggerDict.items()
        # keep panama loggers related to the task
        loggers = [v for k, v in logger_dict if k.startswith(f"PANAMA_{task_name}")]
        # get file handlers for every logger and stash the file
        for l in loggers:
            for h in l.logger.handlers:
                if isinstance(h, PanamaFileHandler):
                    h.move_to_remote()

    def start_task(self, **kwargs):
        """Method used to open a task record."""
        self.log(level=99, msg="RUNNING", extra=kwargs, stack_info=False)

    def end_task(self, trace: Union[Exception, None] = None, move_file: bool = True):
        """Method used to close a task record."""
        if trace is None:
            self.log(level=100, msg="OK")
        else:
            self.log(level=100, msg="KO", exc_info=trace)
        # wait a bit if some files have not been closed
        time.sleep(10)
        if move_file is True:
            # move task json to remote
            self.move_json_to_remote()

    def log_step(self, step_id: str, level: int = logging.INFO, trace: Union[Exception, str, None] = None, **extras):
        """Method used to log a step."""
        self.log(level=level, msg=step_id, extra=extras, exc_info=trace)  # type: ignore

Panama adapter to add contextual information and create the methods start_task, end_task and log_step.

Initialize adapter.

Args

logger : logging.Logger

underlying logger object.

config_run : ConfigRun

config run, with default data.

Ancestors

• logging.LoggerAdapter

Methods

def end_task(self, trace: Exception | None = None, move_file: bool = True)

Expand source code

def end_task(self, trace: Union[Exception, None] = None, move_file: bool = True):
    """Method used to close a task record."""
    if trace is None:
        self.log(level=100, msg="OK")
    else:
        self.log(level=100, msg="KO", exc_info=trace)
    # wait a bit if some files have not been closed
    time.sleep(10)
    if move_file is True:
        # move task json to remote
        self.move_json_to_remote()

Method used to close a task record.

def log_step(self, step_id: str, level: int = 20, trace: Exception | str | None = None, **extras)

Expand source code

def log_step(self, step_id: str, level: int = logging.INFO, trace: Union[Exception, str, None] = None, **extras):
    """Method used to log a step."""
    self.log(level=level, msg=step_id, extra=extras, exc_info=trace)  # type: ignore

Method used to log a step.

def move_json_to_remote(self)

Expand source code

def move_json_to_remote(self):
    """Method used to move logs to remote storage."""
    # get task name from config_run
    task_name = self.config_run.task_name
    # get logger_dict from manager
    logger_dict = self.logger.manager.loggerDict.items()
    # keep panama loggers related to the task
    loggers = [v for k, v in logger_dict if k.startswith(f"PANAMA_{task_name}")]
    # get file handlers for every logger and stash the file
    for l in loggers:
        for h in l.logger.handlers:
            if isinstance(h, PanamaFileHandler):
                h.move_to_remote()

Method used to move logs to remote storage.

def process(self, msg, kwargs)

Expand source code

def process(self, msg, kwargs):
    """Add to extra the required information. Deals with trace parameter.

    Args:
        msg (str): logged message content.
        kwargs (dict): log call arguments.

    Returns:
        (str, dict): msg and kwargs handled.
    """
    forbidden = ["exc_info", "stack_info", "logger"]
    extra = {k: str(v) for k, v in kwargs.items() if k not in forbidden}
    # add trace kwarg to extra
    if isinstance(kwargs.get("exc_info"), str):
        extra["trace"] = kwargs["exc_info"]
    kwargs["extra"] = extra
    # remove trace from kwargs, as it has been added to extras
    kwargs.pop("trace", None)
    return msg, kwargs

Add to extra the required information. Deals with trace parameter.

Args

msg : str

logged message content.

kwargs : dict

log call arguments.

Returns

(str, dict): msg and kwargs handled.

def start_task(self, **kwargs)

Expand source code

def start_task(self, **kwargs):
    """Method used to open a task record."""
    self.log(level=99, msg="RUNNING", extra=kwargs, stack_info=False)

Method used to open a task record.