Source code for smonitor
from __future__ import annotations
from importlib.metadata import PackageNotFoundError, version
try:
__version__ = version("smonitor")
except PackageNotFoundError:
# Package is not installed
try:
from ._version import __version__
except ImportError:
__version__ = "0.0.0+unknown"
from pathlib import Path
from typing import Optional
from . import integrations
from .bundle import collect_bundle, export_bundle
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 get_manager
from .handlers.console import ConsoleHandler, RichConsoleHandler
__all__ = [
"configure",
"emit",
"resolve",
"report",
"signal",
"get_manager",
"export_bundle",
"collect_bundle",
"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)
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)