Source code for smonitor
from __future__ import annotations
# The build writes `_version.py`, so an installed distribution reports the same
# string either way — but reading it costs nothing, while `importlib.metadata`
# drags in `email.message`, `zipfile` and `inspect` and was over a third of the
# time to import this package. Since every library in the ecosystem pays that
# before doing any work, the cheap source is tried first and the metadata
# machinery is imported only when there is no build in this tree.
try:
from ._version import __version__
except ImportError:
try:
from importlib.metadata import version
__version__ = version("smonitor")
except Exception:
# Nothing here may keep the package from importing.
__version__ = "0.0.0+unknown"
from pathlib import Path
from typing import Optional
from . import integrations
from .bundle import (
collect_bundle,
compare_bundles,
export_bundle,
format_bundle_comparison_markdown,
)
from .config import (
build_effective_config,
extract_codes,
extract_policy,
extract_signals,
load_env_config,
load_project_config,
validate_project_config,
)
from .core.decorator import signal
from .core.manager import CONFIGURE_PARAMETERS, get_manager
from .handlers.console import ConsoleHandler, RichConsoleHandler
#: What a configuration file or environment variable may contribute to the
#: manager. `strict_config` is included because it survives in `effective` until
#: it is read and popped below; it governs validation, not the manager.
_CONFIG_FILE_KEYS = CONFIGURE_PARAMETERS | {"strict_config"}
__all__ = [
"configure",
"emit",
"resolve",
"report",
"signal",
"get_manager",
"export_bundle",
"collect_bundle",
"compare_bundles",
"format_bundle_comparison_markdown",
"integrations",
]
[docs]
def configure(**kwargs):
manager = get_manager()
config_path: Optional[Path] = kwargs.pop("config_path", None)
codes_override = kwargs.pop("codes", None)
signals_override = kwargs.pop("signals", None)
routes_override = kwargs.pop("routes", None)
filters_override = kwargs.pop("filters", None)
strict_config = kwargs.pop("strict_config", None)
project_cfg = load_project_config(config_path or Path.cwd())
env_cfg = load_env_config()
effective = build_effective_config(project_cfg, env_cfg)
# `_smonitor.py` and the environment are data, not a call signature. A key
# the manager does not accept must not reach it as a keyword argument,
# because `ensure_configured()` runs during the host library's import and a
# `TypeError` there takes the whole library down over a typo in a config
# file. `validate_project_config` names such keys, and `strict_config`
# still raises on them below.
#
# Keys passed directly to this function are not filtered: there a typo is
# the caller's own, on the line they are looking at, and an immediate
# `TypeError` is the right answer.
for key in [key for key in effective if key not in _CONFIG_FILE_KEYS]:
effective.pop(key)
effective.update({k: v for k, v in kwargs.items() if v is not None})
if strict_config is None:
strict_config = effective.get("strict_config")
if "strict_config" in effective:
effective.pop("strict_config")
errors = validate_project_config(project_cfg)
if errors and strict_config:
raise ValueError("Invalid _smonitor.py: " + "; ".join(errors))
policy = extract_policy(project_cfg)
if routes_override is not None:
policy["routes"] = routes_override
if filters_override is not None:
policy["filters"] = filters_override
if codes_override is not None:
codes = codes_override
else:
project_codes = extract_codes(project_cfg)
if project_codes:
codes = {**manager.get_codes(), **project_codes}
else:
codes = None
if signals_override is not None:
signals = signals_override
else:
project_signals = extract_signals(project_cfg)
if project_signals:
signals = {**manager.get_signals(), **project_signals}
else:
signals = None
if "handlers" not in kwargs or kwargs["handlers"] is None:
# Default to a console handler if none provided
if not manager._handlers:
theme = effective.get("theme", "plain")
if theme == "rich":
manager.add_handler(RichConsoleHandler())
else:
manager.add_handler(ConsoleHandler())
manager.configure(**effective, **policy, codes=codes, signals=signals)
return manager
[docs]
def emit(level, message, **kwargs):
manager = get_manager()
if not manager._handlers:
manager.add_handler(ConsoleHandler())
return manager.emit(level, message, **kwargs)
[docs]
def resolve(message=None, **kwargs):
manager = get_manager()
return manager.resolve(message, **kwargs)