Monitoring API

Complete API reference for the monitoring module.

Tip

Pair this reference with Monitoring for the feature guide.

Quick Overview

  • Monitoring: Simplified API with @collector decorators

  • MonitoringService: Lower-level orchestrator for collect/render pipeline

  • Render Types: StatusCell, MonitorTable, MonitorKV, MonitorList, MonitorMetric, MonitorImage

  • Delivery: FileDelivery (local files), MailDelivery (email via kstlib.mail)

  • Config: Load from kstlib.conf.yml or *.monitor.yml files

Simplified API

Monitoring

class kstlib.monitoring.monitoring.Monitoring(*, template=None, template_file=None, inline_css=True, fail_fast=False, delivery=None, name='monitoring')[source]

Bases: object

Simplified monitoring with decorator-based collectors.

This class provides a cleaner API than MonitoringService:

  • Collectors are registered via @mon.collector decorator

  • Config loaded from kstlib.conf.yml section monitoring:

  • Automatic template_file resolution

  • Integrated delivery

Parameters:

  • template (str | None) – Jinja2 template string (mutually exclusive with template_file).

  • template_file (str | Path | None) – Path to template file (mutually exclusive with template).

  • inline_css (bool) – Use inline CSS for email compatibility (default True).

  • fail_fast (bool) – Raise on first collector error (default False).

  • delivery (DeliveryBackend | _DeferredMailDelivery | None) – Optional delivery backend (FileDelivery or MailDelivery).

  • name (str) – Dashboard name (for delivery subject, default “monitoring”).

Examples

Direct instantiation:

>>> mon = Monitoring(template="<p>{{ msg }}</p>")
>>> @mon.collector
... def msg():
...     return "Hello"
>>> mon.run_sync().html
'<p>Hello</p>'

From config file:

>>> mon = Monitoring.from_config()
__init__(self, *, template: 'str | None' = None, template_file: 'str | Path | None' = None, inline_css: 'bool' = True, fail_fast: 'bool' = False, delivery: 'DeliveryBackend | _DeferredMailDelivery | None' = None, name: 'str' = 'monitoring') → 'None' -> None[source]

Initialize monitoring instance.

classmethod from_config(config_section: 'str' = 'monitoring', *, base_dir: 'Path | None' = None) → 'Monitoring' -> Monitoring[source]

Create Monitoring instance from kstlib.conf.yml.

Loads the monitoring: section from the config file.

Parameters:

  • config_section (str) – Config section name (default “monitoring”).

  • base_dir (Path | None) – Base directory for resolving template_file paths. Defaults to current working directory.

Returns:

Configured Monitoring instance.

Raises:

ValueError – If config section is missing or invalid.

Return type:

Monitoring

Examples

>>> mon = Monitoring.from_config()  
>>> @mon.collector  
... def data():  
...     return {"key": "value"}
property name: str

Return dashboard name.

property collector_names: list[str]

Return list of registered collector names.

collector(self, func: 'F') → 'F' -> F[source]

Register a collector function.

The function name becomes the template variable name.

Parameters:

func (F) – Function returning data for the template.

Returns:

The original function (unmodified).

Return type:

F

Examples

>>> mon = Monitoring(template="{{ status }}")
>>> @mon.collector
... def status():
...     return "OK"
>>> mon.run_sync().html
'OK'
add_collector(self, name: 'str', func: 'Collector') → 'Monitoring' -> Monitoring[source]

Add a collector with explicit name.

Use this when you need a different name than the function name.

Parameters:
Returns:

Self for chaining.

Return type:

Monitoring

async run(self, *, deliver: 'bool' = True) → 'MonitoringResult' -> MonitoringResult[source]

Collect data, render template, and optionally deliver.

Parameters:

deliver (bool) – If True and delivery is configured, send the result.

Returns:

MonitoringResult with HTML and metadata.

Return type:

MonitoringResult

run_sync(self, *, deliver: 'bool' = True) → 'MonitoringResult' -> MonitoringResult[source]

Run the monitoring pipeline synchronously.

Parameters:

deliver (bool) – If True and delivery is configured, send the result.

Returns:

MonitoringResult with HTML and metadata.

Return type:

MonitoringResult

Orchestration

MonitoringService

class kstlib.monitoring.service.MonitoringService(template, collectors=None, *, inline_css=True, fail_fast=True)[source]

Bases: object

Orchestrator for collecting, rendering, and delivering monitoring dashboards.

The service manages the full lifecycle of a monitoring dashboard:

  1. Collectors - Register data sources as sync or async callables

  2. Template - Jinja2 template with | render filter support

  3. Render - Generate HTML with automatic CSS handling

  4. Deliver - Optional email delivery via kstlib.mail transports

Parameters:

  • template (str) – Jinja2 template string for rendering the dashboard.

  • collectors (dict[str, Collector] | None) – Optional dict mapping names to collector callables.

  • inline_css (bool) – If True (default), use inline CSS for email compatibility.

  • fail_fast (bool) – If True, raise on first collector error. If False, continue and report errors in the result.

Examples

Simple dashboard with status cell:

>>> from kstlib.monitoring import MonitoringService, StatusCell, StatusLevel
>>> service = MonitoringService(
...     template="""<div>{{ status | render }}</div>""",
...     collectors={"status": lambda: StatusCell("OK", StatusLevel.OK)},
... )
>>> result = service.run_sync()
>>> "OK" in result.html
True

Adding collectors after construction (chainable):

>>> service = MonitoringService(template="<p>{{ msg }}</p>")
>>> service.add_collector("msg", lambda: "Hello").run_sync().html
'<p>Hello</p>'
__init__(self, template: 'str', collectors: 'dict[str, Collector] | None' = None, *, inline_css: 'bool' = True, fail_fast: 'bool' = True) → 'None' -> None[source]

Initialize the monitoring service.

Parameters:

  • template (str) – Jinja2 template string.

  • collectors (dict[str, Callable[[], Any] | Callable[[], Awaitable[Any]]] | None) – Optional initial collectors dict.

  • inline_css (bool) – Use inline CSS for email compatibility.

  • fail_fast (bool) – Raise immediately on collector errors.

property template: str

Return the template string.

property inline_css: bool

Return whether inline CSS is enabled.

property collector_names: list[str]

Return list of registered collector names.

add_collector(self, name: 'str', collector: 'Collector') → 'MonitoringService' -> MonitoringService[source]

Add a data collector.

Collectors are callables (sync or async) that return data to be passed to the template. The returned data can be any type including Renderable objects like StatusCell, MonitorTable, etc.

Parameters:

  • name (str) – Name to use in the template (e.g., “status” for {{ status }}).

  • collector (Callable[[], Any] | Callable[[], Awaitable[Any]]) – Callable returning the data. Can be sync or async.

Returns:

Self for method chaining.

Return type:

MonitoringService

Examples

>>> service = MonitoringService(template="<p>{{ x }} + {{ y }}</p>")
>>> service.add_collector("x", lambda: 1).add_collector("y", lambda: 2)
<kstlib.monitoring.service.MonitoringService object at ...>
remove_collector(self, name: 'str') → 'MonitoringService' -> MonitoringService[source]

Remove a collector by name.

Parameters:

name (str) – Name of the collector to remove.

Returns:

Self for method chaining.

Raises:

KeyError – If collector name not found.

Return type:

MonitoringService

async collect(self) → 'tuple[dict[str, Any], list[CollectorError]]' -> tuple[dict[str, Any], list[CollectorError]][source]

Run all collectors and gather data.

Collectors are run concurrently when possible. Async collectors are awaited, sync collectors are called directly.

Returns:

Tuple of (collected data dict, list of errors).

Raises:

CollectorError – If fail_fast=True and any collector fails.

Return type:

tuple[dict[str, Any], list[CollectorError]]

render(self, data: 'dict[str, Any]') → 'str' -> str[source]

Render the template with collected data.

Parameters:

data (dict[str, Any]) – Dictionary of data to pass to the template.

Returns:

Rendered HTML string.

Raises:

RenderError – If template rendering fails.

Return type:

str

async run(self) → 'MonitoringResult' -> MonitoringResult[source]

Collect data and render the dashboard.

This is the main entry point for async usage. It runs all collectors, renders the template, and returns a MonitoringResult.

Returns:

MonitoringResult with HTML, data, timestamps, and any errors.

Return type:

MonitoringResult

Examples

>>> import asyncio
>>> service = MonitoringService(
...     template="<p>{{ msg }}</p>",
...     collectors={"msg": lambda: "Hello"},
... )
>>> result = asyncio.run(service.run())
>>> "Hello" in result.html
True
run_sync(self) → 'MonitoringResult' -> MonitoringResult[source]

Run the monitoring service synchronously.

Convenience method for non-async contexts. Creates a new event loop if needed.

Returns:

MonitoringResult with HTML, data, timestamps, and any errors.

Return type:

MonitoringResult

Examples

>>> service = MonitoringService(
...     template="<p>{{ msg }}</p>",
...     collectors={"msg": lambda: "World"},
... )
>>> "World" in service.run_sync().html
True
async deliver(self, transport: 'MailTransport | AsyncMailTransport', message_builder: 'Callable[[str], EmailMessage]') → 'MonitoringResult' -> MonitoringResult[source]

Collect, render, and deliver via email transport.

This combines run() with email delivery. The message_builder callable receives the rendered HTML and should return a complete EmailMessage.

Parameters:

  • transport (MailTransport | AsyncMailTransport) – Mail transport (sync or async) from kstlib.mail.

  • message_builder (Callable[[str], EmailMessage]) – Callable that takes HTML and returns EmailMessage.

Returns:

MonitoringResult from the run.

Return type:

MonitoringResult

Examples

>>> from email.message import EmailMessage
>>> def build_message(html: str) -> EmailMessage:
...     msg = EmailMessage()
...     msg["From"] = "bot@example.com"
...     msg["To"] = "team@example.com"
...     msg["Subject"] = "Dashboard"
...     msg.set_content(html, subtype="html")
...     return msg

MonitoringResult

class kstlib.monitoring.service.MonitoringResult(html, data, collected_at, rendered_at, errors=<factory>)[source]

Bases: object

Result of a monitoring run.

html

The rendered HTML string.

Type:

str

data

The collected data dictionary.

Type:

dict[str, Any]

collected_at

Timestamp when data was collected.

Type:

datetime.datetime

rendered_at

Timestamp when HTML was rendered.

Type:

datetime.datetime

errors

List of collector errors (if fail_fast=False).

Type:

list[kstlib.monitoring.exceptions.CollectorError]

html: str
data: dict[str, Any]
collected_at: datetime
rendered_at: datetime
errors: list[CollectorError]
property success: bool

Return True if no collector errors occurred.

__init__(self, html: 'str', data: 'dict[str, Any]', collected_at: 'datetime', rendered_at: 'datetime', errors: 'list[CollectorError]' = <factory>) → None -> None

Render Types

Base Protocol

class kstlib.monitoring.types.Renderable(*args, **kwargs)[source]

Bases: Protocol

Protocol for objects that can render themselves as HTML.

render(self, *, inline_css: 'bool' = False) → 'str' -> str[source]

Render this object as an HTML string.

Parameters:

inline_css (bool) – If True, embed styles as inline style attributes instead of CSS class references. Useful for email rendering.

Returns:

HTML string representation.

Return type:

str

__init__(self, *args, **kwargs)

StatusLevel

class kstlib.monitoring.types.StatusLevel(value)[source]

Bases: IntEnum

Severity level for monitoring status indicators.

Values are ordered by severity so comparisons work naturally: StatusLevel.OK < StatusLevel.WARNING < StatusLevel.ERROR.

OK

Normal operation (#16A085 green).

WARNING

Degraded but functional (#F1C40F yellow).

ERROR

Service failure (#E85A4F red).

CRITICAL

Critical failure requiring immediate action (#c0392b dark red).

OK = 10
WARNING = 20
ERROR = 30
CRITICAL = 40

StatusCell

class kstlib.monitoring.cell.StatusCell(label, level)[source]

Bases: object

A colored status badge rendered as an HTML <span>.

label

Display text for the badge (e.g. “UP”, “DOWN”, “DEGRADED”).

Type:

str

level

Severity level controlling badge color.

Type:

StatusLevel

Examples

>>> from kstlib.monitoring.cell import StatusCell
>>> from kstlib.monitoring.types import StatusLevel
>>> cell = StatusCell("UP", StatusLevel.OK)
>>> "<span" in cell.render()
True
label: str
level: StatusLevel
render(self, *, inline_css: 'bool' = False) → 'str' -> str[source]

Render the status badge as an HTML <span>.

Parameters:

inline_css (bool) – If True, use inline styles instead of CSS classes.

Returns:

HTML <span> string.

Return type:

str

__init__(self, label: 'str', level: 'StatusLevel') → None -> None

MonitorTable

class kstlib.monitoring.table.MonitorTable(headers, title='')[source]

Bases: object

A table rendered as an HTML <table> with striped rows.

This is the only mutable render type: rows are added via add_row().

headers

Column headers.

Type:

list[str]

title

Optional caption rendered above the table.

Type:

str

Examples

>>> from kstlib.monitoring.table import MonitorTable
>>> t = MonitorTable(headers=["Service", "Status"])
>>> t.add_row(["API", "OK"])
>>> "<table" in t.render()
True
headers: list[str]
title: str
add_row(self, row: 'list[CellValue | StatusCell]') → 'None' -> None[source]

Append a row to the table.

Parameters:

row (list[CellValue | StatusCell]) – List of cell values matching the number of headers.

Raises:

RenderError – If the row length does not match the header count.

property row_count: int

Return the number of data rows.

render(self, *, inline_css: 'bool' = False) → 'str' -> str[source]

Render the table as an HTML <table>.

Parameters:

inline_css (bool) – If True, use inline styles instead of CSS classes.

Returns:

HTML <table> string.

Return type:

str

__init__(self, headers: 'list[str]', title: 'str' = '') → None -> None

MonitorKV

class kstlib.monitoring.kv.MonitorKV(items, title='')[source]

Bases: object

A key-value grid rendered as an HTML <dl>.

Values can be plain scalars or StatusCell objects for colored badges.

items

Ordered mapping of keys to values.

Type:

dict[str, CellValue | StatusCell]

title

Optional heading rendered above the list.

Type:

str

Examples

>>> from kstlib.monitoring.kv import MonitorKV
>>> kv = MonitorKV({"Host": "srv-01", "Port": 8080})
>>> "<dl" in kv.render()
True
items: dict[str, CellValue | StatusCell]
title: str
render(self, *, inline_css: 'bool' = False) → 'str' -> str[source]

Render the key-value pairs as an HTML <dl>.

Parameters:

inline_css (bool) – If True, use inline styles instead of CSS classes.

Returns:

HTML <dl> string, optionally preceded by an <h3> title.

Return type:

str

__init__(self, items: 'dict[str, CellValue | StatusCell]', title: 'str' = '') → None -> None

MonitorList

class kstlib.monitoring.list.MonitorList(items, ordered=False, title='')[source]

Bases: object

A list rendered as an HTML <ul> or <ol>.

Items can be plain scalars or StatusCell objects for colored badges.

items

Sequence of list items.

Type:

list[CellValue | StatusCell]

ordered

If True, render as <ol>; otherwise <ul>.

Type:

bool

title

Optional heading rendered above the list.

Type:

str

Examples

>>> from kstlib.monitoring.list import MonitorList
>>> ml = MonitorList(["Event A", "Event B"])
>>> "<ul>" in ml.render()
True
items: list[CellValue | StatusCell]
ordered: bool
title: str
render(self, *, inline_css: 'bool' = False) → 'str' -> str[source]

Render the list as an HTML <ul> or <ol>.

Parameters:

inline_css (bool) – If True, use inline styles instead of CSS classes.

Returns:

HTML list string, optionally preceded by an <h3> title.

Return type:

str

__init__(self, items: 'list[CellValue | StatusCell]', ordered: 'bool' = False, title: 'str' = '') → None -> None

MonitorMetric

class kstlib.monitoring.metric.MonitorMetric(value, label='', level=StatusLevel.OK, unit='')[source]

Bases: object

A hero-number metric rendered as an HTML <div>.

value

The metric value to display prominently.

Type:

CellValue

label

Optional descriptive label shown below the value.

Type:

str

level

Severity level controlling the value color.

Type:

StatusLevel

unit

Optional unit suffix appended to the value (e.g. “%”, “ms”).

Type:

str

Examples

>>> from kstlib.monitoring.metric import MonitorMetric
>>> from kstlib.monitoring.types import StatusLevel
>>> m = MonitorMetric(99.9, label="Uptime", level=StatusLevel.OK, unit="%")
>>> "99.9" in m.render()
True
value: CellValue
label: str
level: StatusLevel
unit: str
render(self, *, inline_css: 'bool' = False) → 'str' -> str[source]

Render the metric as an HTML <div>.

Parameters:

inline_css (bool) – If True, use inline styles instead of CSS classes.

Returns:

HTML <div> string.

Return type:

str

__init__(self, value: 'CellValue', label: 'str' = '', level: 'StatusLevel' = <StatusLevel.OK: 10>, unit: 'str' = '') → None -> None

MonitorImage

class kstlib.monitoring.image.MonitorImage(data=None, path=None, alt='', width=None, height=None)[source]

Bases: object

An image rendered as an HTML <img> with a base64 data URI.

The image data can be provided directly as bytes or loaded from a file path. Exactly one of data or path must be given.

data

Raw image bytes. Mutually exclusive with path.

Type:

bytes | None

path

Path to an image file. Mutually exclusive with data.

Type:

Path | None

alt

Alt text for the <img> tag (always HTML-escaped).

Type:

str

width

Optional width attribute (pixels).

Type:

int | None

height

Optional height attribute (pixels).

Type:

int | None

Raises:

RenderError – If both or neither of data/path are given, the image exceeds size limits, or the format is unsupported.

Examples

>>> from kstlib.monitoring.image import MonitorImage
>>> img = MonitorImage(b"\x89PNG\r\n\x1a\n" + b"\x00" * 50, alt="Logo")
>>> "<img" in img.render()
True
data: bytes | None
path: Path | None
alt: str
width: int | None
height: int | None
__post_init__(self) → 'None' -> None[source]

Validate inputs at construction time.

render(self, *, inline_css: 'bool' = False) → 'str' -> str[source]

Render the image as an HTML <img> with a data URI.

The inline_css parameter is accepted for protocol conformance but has no effect on image rendering (images are always inline).

Parameters:

inline_css (bool) – Accepted for Renderable protocol compatibility.

Returns:

HTML <img> string with base64 data URI.

Raises:

RenderError – If the image cannot be loaded, exceeds size limits, has an unsupported format, or (for SVG) contains dangerous content.

Return type:

str

__init__(self, data: 'bytes | None' = None, path: 'Path | None' = None, alt: 'str' = '', width: 'int | None' = None, height: 'int | None' = None) → None -> None

Jinja2 Renderer

render_template

kstlib.monitoring.renderer.render_template(source: 'str', context: 'dict[str, Any] | None' = None, *, inline_css: 'bool' = False) → 'str' -> str[source]

Render a Jinja2 template string with monitoring support.

This is a high-level convenience function that creates a temporary environment, compiles source as a template, and renders it with context.

When inline_css=False (default), the CSS class definitions from get_css_classes() are prepended to the output so that class-based rendering works out of the box.

Parameters:

  • source (str) – Jinja2 template source string.

  • context (dict[str, Any] | None) – Template variables. None is treated as an empty dict.

  • inline_css (bool) – If True, skip the <style> block prepend. Useful when styles are inlined into each element.

Returns:

Rendered HTML string.

Raises:

TypeError – If source is not a str or context is not a dict (or None).

Return type:

str

Examples

>>> from kstlib.monitoring.renderer import render_template
>>> render_template("Hello {{ name }}", {"name": "World"}, inline_css=True)
'Hello World'

render_html

kstlib.monitoring.renderer.render_html(value: 'Any', inline_css: 'bool' = False) → 'Markup' -> Markup[source]

Jinja2 filter that renders a value as safe HTML.

When value implements Renderable, its .render() method is called. Otherwise the value is converted to a string and HTML-escaped.

Register this filter under the name render on a Jinja2 environment so templates can use {{ data | render }}.

Parameters:

  • value (Any) – Any value to render. Renderable objects are dispatched to their own render() method; everything else is escaped.

  • inline_css (bool) – Forwarded to Renderable.render(inline_css=...).

Returns:

A jinja2.Markup instance (marked safe to prevent double-escaping by Jinja2 autoescape).

Return type:

Markup

Examples

>>> from kstlib.monitoring.renderer import render_html
>>> str(render_html("<b>bold</b>"))
'&lt;b&gt;bold&lt;/b&gt;'

create_environment

kstlib.monitoring.renderer.create_environment(**kwargs: 'Any') → 'Environment' -> Environment[source]

Create a Jinja2 Environment with monitoring filters.

The returned environment has:

  • autoescape=True by default (overridable via kwargs).

  • The render filter bound to render_html().

Parameters:

**kwargs (Any) – Forwarded to jinja2.Environment. Common options include loader, autoescape, trim_blocks.

Returns:

A configured jinja2.Environment.

Return type:

Environment

Examples

>>> from kstlib.monitoring.renderer import create_environment
>>> env = create_environment()
>>> "render" in env.filters
True

get_css_classes

kstlib.monitoring._styles.get_css_classes() → 'str' -> str[source]

Return a <style> block with all monitoring CSS classes.

This should be included once in the <head> of an HTML document when using class-based rendering (inline_css=False).

Returns:

Complete <style> element as a string.

Return type:

str

Examples

>>> from kstlib.monitoring._styles import get_css_classes
>>> css = get_css_classes()
>>> "<style>" in css
True
>>> ".status-ok" in css
True

Configuration

MonitoringConfig

class kstlib.monitoring.config.MonitoringConfig(name, template, collectors=<factory>, inline_css=True, fail_fast=True, source_path=None, metadata=<factory>)[source]

Bases: object

Parsed monitoring configuration.

name

Dashboard name (defaults to filename without extension).

Type:

str

template

Jinja2 template string for rendering.

Type:

str

collectors

List of collector configurations.

Type:

list[kstlib.monitoring.config.CollectorConfig]

inline_css

Whether to use inline CSS (default True).

Type:

bool

fail_fast

Whether to fail on first collector error (default True).

Type:

bool

source_path

Path to the source config file (if loaded from file).

Type:

pathlib.Path | None

metadata

Additional metadata from the config file.

Type:

dict[str, Any]

name: str
template: str
collectors: list[CollectorConfig]
inline_css: bool = True
fail_fast: bool = True
source_path: Path | None = None
metadata: dict[str, Any]
to_service(self) → 'MonitoringService' -> MonitoringService[source]

Create a MonitoringService from this configuration.

Returns:

Configured MonitoringService instance.

Raises:

MonitoringConfigCollectorError – If any collector cannot be created.

Return type:

MonitoringService

classmethod from_dict(data: 'dict[str, Any]', *, source_path: 'pathlib.Path | None' = None) → 'MonitoringConfig' -> MonitoringConfig[source]

Create a MonitoringConfig from a dictionary.

Parameters:

  • data (dict[str, Any]) – Configuration dictionary.

  • source_path (Path | None) – Optional path to source file.

Returns:

Parsed MonitoringConfig.

Raises:

MonitoringConfigFormatError – If required fields are missing.

Return type:

MonitoringConfig

__init__(self, name: 'str', template: 'str', collectors: 'list[CollectorConfig]' = <factory>, inline_css: 'bool' = True, fail_fast: 'bool' = True, source_path: 'pathlib.Path | None' = None, metadata: 'dict[str, Any]' = <factory>) → None -> None

CollectorConfig

class kstlib.monitoring.config.CollectorConfig(name, collector_type='static', value=None, module=None, function=None, env_var=None, default=None)[source]

Bases: object

Configuration for a single collector.

name

Name of the collector (used as template variable).

Type:

str

collector_type

Type of collector (“static”, “callable”, “env”).

Type:

str

value

Static value (for type=”static”).

Type:

Any

module

Module path (for type=”callable”).

Type:

str | None

function

Function name (for type=”callable”).

Type:

str | None

env_var

Environment variable name (for type=”env”).

Type:

str | None

default

Default value if env var not set (for type=”env”).

Type:

Any

name: str
collector_type: str = 'static'
value: Any = None
module: str | None = None
function: str | None = None
env_var: str | None = None
default: Any = None
to_collector(self) → 'Collector' -> Collector[source]

Convert config to a collector callable.

Returns:

Callable that can be used as a MonitoringService collector.

Raises:

MonitoringConfigCollectorError – If collector cannot be created.

Return type:

Callable[[], Any] | Callable[[], Awaitable[Any]]

__init__(self, name: 'str', collector_type: 'str' = 'static', value: 'Any' = None, module: 'str | None' = None, function: 'str | None' = None, env_var: 'str | None' = None, default: 'Any' = None) → None -> None

load_monitoring_config

kstlib.monitoring.config.load_monitoring_config(path: 'str | pathlib.Path', *, encoding: 'str' = 'utf-8') → 'MonitoringConfig' -> MonitoringConfig[source]

Load a monitoring configuration from a YAML file.

Parameters:

  • path (str | Path) – Path to the monitoring config file.

  • encoding (str) – File encoding (default UTF-8).

Returns:

Parsed MonitoringConfig.

Raises:

  • MonitoringConfigFileNotFoundError – If file does not exist.

  • MonitoringConfigFormatError – If file format is invalid.

Return type:

MonitoringConfig

Examples

>>> config = load_monitoring_config("dashboard.monitor.yml")  
>>> service = config.to_service()  
>>> result = service.run_sync()

discover_monitoring_configs

kstlib.monitoring.config.discover_monitoring_configs(directory: 'str | pathlib.Path', *, recursive: 'bool' = False, encoding: 'str' = 'utf-8') → 'dict[str, MonitoringConfig]' -> dict[str, MonitoringConfig][source]

Discover and load all monitoring configs in a directory.

Searches for files matching *.monitor.yml pattern.

Parameters:

  • directory (str | Path) – Directory to search.

  • recursive (bool) – If True, search subdirectories recursively.

  • encoding (str) – File encoding (default UTF-8).

Returns:

Dictionary mapping config names to MonitoringConfig objects.

Raises:

  • FileNotFoundError – If directory does not exist.

  • MonitoringConfigFormatError – If any config file is invalid.

Return type:

dict[str, MonitoringConfig]

Examples

>>> configs = discover_monitoring_configs("./monitoring")  
>>> for name, config in configs.items():  
...     print(f"Loaded: {name}")

create_services_from_directory

kstlib.monitoring.config.create_services_from_directory(directory: 'str | pathlib.Path', *, recursive: 'bool' = False, encoding: 'str' = 'utf-8') → 'dict[str, MonitoringService]' -> dict[str, MonitoringService][source]

Discover configs and create MonitoringService instances.

Convenience function that combines discover_monitoring_configs with to_service() for each config.

Parameters:

  • directory (str | Path) – Directory to search for *.monitor.yml files.

  • recursive (bool) – If True, search subdirectories recursively.

  • encoding (str) – File encoding (default UTF-8).

Returns:

Dictionary mapping config names to MonitoringService instances.

Return type:

dict[str, MonitoringService]

Examples

>>> services = create_services_from_directory("./monitoring")  
>>> for name, service in services.items():  
...     result = service.run_sync()  
...     print(f"{name}: {result.success}")

Delivery Backends

DeliveryBackend

class kstlib.monitoring.delivery.DeliveryBackend[source]

Bases: ABC

Abstract base class for delivery backends.

abstract async deliver(self, result: 'MonitoringResult', name: 'str') → 'DeliveryResult' -> DeliveryResult[source]

Deliver a monitoring result.

Parameters:

  • result (MonitoringResult) – The MonitoringResult to deliver.

  • name (str) – Name/subject for this delivery.

Returns:

DeliveryResult with success status and metadata.

Return type:

DeliveryResult

DeliveryResult

class kstlib.monitoring.delivery.DeliveryResult(success, timestamp, path=None, message_id=None, error=None, metadata=<factory>)[source]

Bases: object

Result of a delivery operation.

success

Whether delivery succeeded.

Type:

bool

timestamp

When delivery was attempted.

Type:

datetime.datetime

path

Output file path (for file delivery).

Type:

pathlib.Path | None

message_id

Email message ID (for mail delivery).

Type:

str | None

error

Error message if delivery failed.

Type:

str | None

metadata

Additional delivery metadata.

Type:

dict[str, Any]

success: bool
timestamp: datetime
path: Path | None
message_id: str | None
error: str | None
metadata: dict[str, Any]
__init__(self, success: 'bool', timestamp: 'datetime', path: 'pathlib.Path | None' = None, message_id: 'str | None' = None, error: 'str | None' = None, metadata: 'dict[str, Any]' = <factory>) → None -> None

FileDelivery

class kstlib.monitoring.delivery.FileDelivery(output_dir, *, filename_template='{name}_{timestamp}.html', create_dirs=True, max_files=100, encoding='utf-8')[source]

Bases: DeliveryBackend

Deliver monitoring results to local files.

Saves HTML output to files with automatic rotation and cleanup.

Parameters:

  • output_dir (str | pathlib.Path) – Directory to save files (str or Path).

  • filename_template (str) – Template for filenames.

  • create_dirs (bool) – Create output directory if missing.

  • max_files (int) – Maximum files to keep (oldest deleted when exceeded).

  • encoding (str) – File encoding.

Examples

>>> delivery = FileDelivery(output_dir="./reports")  
>>> result = await delivery.deliver(monitoring_result, "daily")  
>>> print(f"Saved to: {result.path}")

With rotation (keep last 7 files):

>>> delivery = FileDelivery(  
...     output_dir="./reports",
...     max_files=7,
... )
__init__(self, output_dir: 'str | pathlib.Path', *, filename_template: 'str' = '{name}_{timestamp}.html', create_dirs: 'bool' = True, max_files: 'int' = 100, encoding: 'str' = 'utf-8') → 'None' -> None[source]

Initialize file delivery backend.

property config: FileDeliveryConfig

Return the delivery configuration.

property last_result: DeliveryResult | None

Return the last delivery result.

async deliver(self, result: 'MonitoringResult', name: 'str') → 'DeliveryResult' -> DeliveryResult[source]

Save monitoring result HTML to a file.

Parameters:

  • result (MonitoringResult) – The MonitoringResult to save.

  • name (str) – Name for this report (used in filename).

Returns:

DeliveryResult with file path on success.

Raises:

  • DeliveryIOError – If file cannot be written.

  • DeliveryConfigError – If configuration is invalid.

Return type:

DeliveryResult

FileDeliveryConfig

class kstlib.monitoring.delivery.FileDeliveryConfig(output_dir, filename_template='{name}_{timestamp}.html', create_dirs=True, max_files=100, encoding='utf-8')[source]

Bases: object

Configuration for file delivery.

output_dir

Directory to save files.

Type:

str | pathlib.Path

filename_template

Template for filenames (supports {name}, {timestamp}).

Type:

str

create_dirs

Create output directory if missing.

Type:

bool

max_files

Maximum files to keep (oldest deleted, 0=unlimited).

Type:

int

encoding

File encoding.

Type:

str

output_dir: str | Path
filename_template: str = '{name}_{timestamp}.html'
create_dirs: bool = True
max_files: int = 100
encoding: str = 'utf-8'
__post_init__(self) → 'None' -> None[source]

Validate configuration after initialization.

__init__(self, output_dir: 'str | pathlib.Path', filename_template: 'str' = '{name}_{timestamp}.html', create_dirs: 'bool' = True, max_files: 'int' = 100, encoding: 'str' = 'utf-8') → None -> None

MailDelivery

class kstlib.monitoring.delivery.MailDelivery(transport, config)[source]

Bases: DeliveryBackend

Deliver monitoring results via email.

Wraps kstlib.mail transports for monitoring delivery.

Parameters:

  • transport (MailTransport | AsyncMailTransport) – Mail transport (sync or async).

  • sender – Sender email address.

  • recipients – List of recipient addresses.

  • cc – Optional CC addresses.

  • bcc – Optional BCC addresses.

  • subject_template – Subject template with {name} placeholder.

  • include_plain_text – Include plain text version of HTML.

Examples

>>> from kstlib.mail.transports.gmail import GmailTransport
>>> transport = GmailTransport(...)  
>>> delivery = MailDelivery(  
...     transport=transport,
...     sender="bot@example.com",
...     recipients=["team@example.com"],
... )
>>> result = await delivery.deliver(monitoring_result, "Daily Report")
__init__(self, transport: 'MailTransport | AsyncMailTransport', config: 'MailDeliveryConfig') → 'None' -> None[source]

Initialize mail delivery backend.

Parameters:

  • transport (MailTransport | AsyncMailTransport) – Mail transport (sync or async).

  • config (MailDeliveryConfig) – Mail delivery configuration.

classmethod create(transport: 'MailTransport | AsyncMailTransport', sender: 'str', recipients: 'list[str]', **kwargs: 'Any') → 'MailDelivery' -> MailDelivery[source]

Create a MailDelivery with configuration.

Convenience factory method that creates the config internally.

Parameters:

  • transport (MailTransport | AsyncMailTransport) – Mail transport (sync or async).

  • sender (str) – Sender email address.

  • recipients (list[str]) – List of recipient addresses.

  • **kwargs (Any) – Additional config options (cc, bcc, subject_template, etc.).

Returns:

Configured MailDelivery instance.

Return type:

MailDelivery

property config: MailDeliveryConfig

Return the delivery configuration.

property last_result: DeliveryResult | None

Return the last delivery result.

async deliver(self, result: 'MonitoringResult', name: 'str') → 'DeliveryResult' -> DeliveryResult[source]

Send monitoring result via email.

Parameters:

  • result (MonitoringResult) – The MonitoringResult to send.

  • name (str) – Name for this report (used in subject).

Returns:

DeliveryResult with message ID on success.

Raises:

DeliveryError – If email cannot be sent.

Return type:

DeliveryResult

MailDeliveryConfig

class kstlib.monitoring.delivery.MailDeliveryConfig(sender, recipients, cc=<factory>, bcc=<factory>, subject_template='Monitoring Report: {name}', include_plain_text=True)[source]

Bases: object

Configuration for mail delivery.

sender

Sender email address.

Type:

str

recipients

List of recipient addresses.

Type:

list[str]

cc

List of CC addresses.

Type:

list[str]

bcc

List of BCC addresses.

Type:

list[str]

subject_template

Subject template (supports {name}).

Type:

str

include_plain_text

Include plain text version.

Type:

bool

sender: str
recipients: list[str]
cc: list[str]
bcc: list[str]
subject_template: str = 'Monitoring Report: {name}'
include_plain_text: bool = True
__post_init__(self) → 'None' -> None[source]

Validate configuration after initialization.

__init__(self, sender: 'str', recipients: 'list[str]', cc: 'list[str]' = <factory>, bcc: 'list[str]' = <factory>, subject_template: 'str' = 'Monitoring Report: {name}', include_plain_text: 'bool' = True) → None -> None

Exceptions

See Monitoring Exceptions for detailed exception handling patterns.

Specialized exceptions raised by the kstlib.monitoring module.

Exception hierarchy:

KstlibError
    MonitoringError (base)
        CollectorError
        RenderError
        MonitoringConfigError
exception kstlib.monitoring.exceptions.CollectorError(collector_name, cause)[source]

Bases: MonitoringError

Error during data collection.

Raised when a collector callable fails during MonitoringService.collect().

collector_name

Name of the failed collector.

cause

The underlying exception that caused the failure.

__init__(self, collector_name: 'str', cause: 'Exception') → 'None' -> None[source]

Initialize with collector name and underlying cause.

Parameters:

  • collector_name (str) – Name of the failed collector.

  • cause (Exception) – The exception that caused the failure.

exception kstlib.monitoring.exceptions.MonitoringConfigError[source]

Bases: MonitoringError

Base exception for monitoring configuration errors.

Raised when a monitoring configuration file cannot be loaded or parsed.

exception kstlib.monitoring.exceptions.MonitoringError[source]

Bases: KstlibError

Base exception for all monitoring errors.

exception kstlib.monitoring.exceptions.RenderError[source]

Bases: MonitoringError, ValueError

HTML rendering failed.

Raised when a renderable object cannot produce valid HTML output, for example due to inconsistent data dimensions or template errors.

Config Exceptions

class kstlib.monitoring.config.MonitoringConfigFileNotFoundError[source]

Bases: MonitoringConfigError, FileNotFoundError

Monitoring configuration file not found.

class kstlib.monitoring.config.MonitoringConfigFormatError[source]

Bases: MonitoringConfigError, ValueError

Invalid monitoring configuration format.

class kstlib.monitoring.config.MonitoringConfigCollectorError[source]

Bases: MonitoringConfigError

Error loading a collector from configuration.

Delivery Exceptions

class kstlib.monitoring.delivery.DeliveryError[source]

Bases: MonitoringError

Base exception for delivery errors.

class kstlib.monitoring.delivery.DeliveryConfigError[source]

Bases: DeliveryError, ValueError

Invalid delivery configuration.

class kstlib.monitoring.delivery.DeliveryIOError[source]

Bases: DeliveryError, OSError

I/O error during delivery.

Module Exports

The following are available directly from kstlib.monitoring:

from kstlib.monitoring import (
    # Simplified API
    Monitoring,

    # Orchestration
    MonitoringService,
    MonitoringResult,

    # Render types
    StatusCell,
    MonitorTable,
    MonitorKV,
    MonitorList,
    MonitorMetric,
    MonitorImage,

    # Enums & Protocol
    StatusLevel,
    Renderable,
    CellValue,

    # Renderer
    render_template,
    render_html,
    create_environment,
    get_css_classes,

    # Config
    MonitoringConfig,
    CollectorConfig,
    load_monitoring_config,
    discover_monitoring_configs,
    create_services_from_directory,

    # Delivery
    DeliveryBackend,
    DeliveryResult,
    FileDelivery,
    FileDeliveryConfig,
    MailDelivery,
    MailDeliveryConfig,

    # Exceptions
    MonitoringError,
    CollectorError,
    RenderError,
    MonitoringConfigError,
    MonitoringConfigFileNotFoundError,
    MonitoringConfigFormatError,
    MonitoringConfigCollectorError,
    DeliveryError,
    DeliveryConfigError,
    DeliveryIOError,
)