PyLogShield Logger¶

def info(self, msg: Any, *args: Any, mask: bool = False, **kwargs: Any) -> None:
    self._log_with_processing(logging.INFO, msg, *args, mask=mask, **kwargs)

def debug(self, msg: Any, *args: Any, mask: bool = False, **kwargs: Any) -> None:
    self._log_with_processing(logging.DEBUG, msg, *args, mask=mask, **kwargs)

def warning(self, msg: Any, *args: Any, mask: bool = False, **kwargs: Any) -> None:
    self._log_with_processing(logging.WARNING, msg, *args, mask=mask, **kwargs)

def error(self, msg: Any, *args: Any, mask: bool = False, **kwargs: Any) -> None:
    self._log_with_processing(logging.ERROR, msg, *args, mask=mask, **kwargs)

def critical(self, msg: Any, *args: Any, mask: bool = False, **kwargs: Any) -> None:
    self._log_with_processing(logging.CRITICAL, msg, *args, mask=mask, **kwargs)

def exception(  # type: ignore[override]
    self,
    msg: Any,
    *args: Any,
    mask: bool = False,
    exc_info: bool = True,
    **kwargs: Any,
) -> None:
    """Log an error with exception info.

    When ``mask=True``, string args of the active exception are scrubbed
    before the record is emitted.

    Note: traceback *locals* are not masked — use a traceback-filtering
    tool if local variable values must also be redacted.
    """
    self._log_with_processing(
        logging.ERROR, msg, *args, mask=mask, exc_info=exc_info, **kwargs
    )

def set_log_level(self, level: Union[LogLevel, str, int]) -> None:
    """Change the logger and handler levels at runtime.

    Parameters
    ----------
    level : LogLevel or str or int
        The new logging level to set.
    """
    resolved = self._resolve_log_level(level)
    self.setLevel(resolved)
    self.log_level = resolved
    for handler in self.handlers:
        handler.setLevel(resolved)

def get_metrics(self) -> Optional[Dict[str, Any]]:
    """Return logging metrics if metrics are enabled.

    Returns
    -------
    dict or None
        Dictionary with log counts and rates per level, total count,
        and elapsed time. Returns None if metrics are disabled.

    Examples
    --------
    >>> logger = get_logger("app", enable_metrics=True)
    >>> logger.info("test")
    >>> metrics = logger.get_metrics()
    >>> print(f"Total logs: {metrics['count']}")
    """
    if self.metrics_handler is not None:
        return self.metrics_handler.logs_per_second()
    return None

def context(self, **fields: Any):
    """Return a sync context manager that injects *fields* into all logs.

    Shorthand for :func:`~pylogshield.context.log_context`.  Requires the
    logger to have been created with ``enable_context=True``.

    Parameters
    ----------
    **fields : Any
        Key/value pairs to attach to every log record emitted inside the
        ``with`` block.

    Examples
    --------
    >>> with logger.context(request_id="abc", user_id=42):
    ...     logger.info("Processing order")
    """
    return log_context(**fields)

def async_context(self, **fields: Any):
    """Return an async context manager that injects *fields* into all logs.

    Shorthand for :func:`~pylogshield.context.async_log_context`.
    Requires the logger to have been created with ``enable_context=True``.

    Parameters
    ----------
    **fields : Any
        Key/value pairs to attach to every log record emitted inside the
        ``async with`` block.

    Examples
    --------
    >>> async with logger.async_context(request_id="xyz"):
    ...     logger.info("Async handler")
    """
    return async_log_context(**fields)

def shutdown(self) -> None:
    """Stop any background listener and clean up resources.

    This should be called when the logger is no longer needed to properly
    stop the background queue listener thread if async logging is enabled.
    """
    if self._queue_listener is not None:
        self._queue_listener.stop()
        self._queue_listener = None

@classmethod
def from_config(cls, name: str, config: Mapping[str, Any]) -> "PyLogShield":
    """Create a PyLogShield instance from a dictionary configuration.

    Parameters
    ----------
    name : str
        The name of the logger.
    config : Mapping[str, Any]
        Configuration dictionary with optional keys: level, enable_json,
        use_queue, use_rich, rate_limit_seconds, log_directory, log_file,
        rotate_file, rotate_max_bytes, rotate_backup_count, add_console,
        enable_metrics, log_filter, enable_context_scrubber, enable_context,
        queue_maxsize.

    Returns
    -------
    PyLogShield
        A new PyLogShield instance configured from the provided dictionary.

    Examples
    --------
    >>> config = {"level": "DEBUG", "enable_json": True, "rotate_file": True}
    >>> logger = PyLogShield.from_config("my_app", config)
    """
    # Optional log-filter construction from config
    lf = config.get("log_filter")
    lf_obj: Optional[logging.Filter] = None
    if isinstance(lf, dict) and "keywords" in lf:
        lf_obj = KeywordFilter(
            lf.get("keywords", []),
            include=bool(lf.get("include", True)),
            case_insensitive=bool(lf.get("case_insensitive", True)),
        )
    elif isinstance(lf, (list, tuple, set, frozenset)):
        lf_obj = KeywordFilter(list(lf))
    elif isinstance(lf, logging.Filter):
        lf_obj = lf

    return cls(
        name,
        log_level=LogLevel.parse(config.get("level", "INFO")),
        enable_json=bool(config.get("enable_json", False)),
        use_queue=bool(config.get("use_queue", False)),
        use_rich=bool(config.get("use_rich", False)),
        rate_limit_seconds=float(config.get("rate_limit_seconds", 0.0)),
        log_directory=config.get("log_directory"),
        log_file=config.get("log_file"),
        rotate_file=bool(config.get("rotate_file", False)),
        rotate_max_bytes=int(config.get("rotate_max_bytes", 5_000_000)),
        rotate_backup_count=int(config.get("rotate_backup_count", 5)),
        add_console=bool(config.get("add_console", True)),
        enable_metrics=bool(config.get("enable_metrics", False)),
        log_filter=lf_obj,
        enable_context_scrubber=bool(config.get("enable_context_scrubber", True)),
        enable_context=bool(config.get("enable_context", False)),
        queue_maxsize=int(config.get("queue_maxsize", 0)),
    )

@staticmethod
def add_sensitive_fields(fields: List[str]) -> None:
    """Add field names to the sensitive data redaction registry.

    Parameters
    ----------
    fields : list of str
        List of field names to add to the sensitive registry.

    Examples
    --------
    >>> PyLogShield.add_sensitive_fields(["ssn", "credit_card"])
    """
    cfg_add_sensitive_fields(fields)