Skip to content

watchpost

Modules:

  • app –

    Watchpost application and Starlette integration.

  • cache –

    Caching utilities and pluggable storage backends.

  • check –

    Check definitions, execution helpers, and caching.

  • cli –
  • datasource –

    Datasource primitives and factory helpers.

  • discover_checks –

    Utilities for discovering checks defined in modules and packages.

  • environment –

    Execution environments and registry.

  • executor –

    Threaded check executor for Watchpost.

  • globals –

    Global state for Watchpost and a proxy to the active application instance.

  • hostname –

    Hostname resolution utilities and strategies.

  • http –

    Starlette HTTP routes for Watchpost.

  • result –

    Result types and builders for Watchpost checks.

  • scheduling_strategy –

    Scheduling strategies and validation.

  • utils –

    Utility helpers for Watchpost.

  • vendored –

Classes:

Functions:

  • build_result –

    Start building up a new check result that allows adding multiple

  • crit –

    Generates a CheckResult object indicating a CRIT check state.

  • ok –

    Generates a CheckResult object indicating an OK check state.

  • unknown –

    Generates a CheckResult object indicating an UNKNOWN check state.

  • warn –

    Generates a CheckResult object indicating a WARN check state.

Cache 

Cache(storage: Storage)

High-level caching API using a pluggable storage backend.

Use get and store to interact with the cache directly, or the memoize decorator to cache function results.

Methods:

  • get –

    Retrieve a value from the cache.

  • memoize –

    Memoize a function by caching its return value.

  • store –

    Store a value in the cache.

get 

get(
    key: Hashable,
    default: T | None = None,
    *,
    package: str | None = None,
    return_expired: bool = False,
) -> CacheEntry[T] | None

Retrieve a value from the cache.

Parameters:

  • key

    (Hashable) –

    The key to look up. Must be hashable and unique within the given package.

  • default

    (T | None, default: None ) –

    An optional default value to return when the key is not found.

  • package

    (str | None, default: None ) –

    The package namespace. If not provided, the package of the caller is used.

  • return_expired

    (bool, default: False ) –

    Whether to return an expired entry once before it is removed. (This behavior does depend on the storage backend supporting this.)

Returns:

  • CacheEntry[T] | None –

    The cache entry if found. If not found, returns None when no

  • CacheEntry[T] | None –

    default is given, otherwise a cache entry wrapping the default

  • CacheEntry[T] | None –

    value.

memoize 

memoize(
    *,
    key: Hashable | None = None,
    key_generator: Callable[P, Hashable] | None = None,
    package: str | None = None,
    return_expired: bool = False,
    ttl: timedelta | None = None,
) -> Callable[[Callable[P, R]], Callable[P, R]]

Memoize a function by caching its return value.

Use either a fixed key or a key_generator to compute the cache key from call arguments. If key contains format placeholders like "{arg}", they will be formatted from the function's bound arguments.

Parameters:

  • key

    (Hashable | None, default: None ) –

    A static key or a format string used to build the key. Mutually exclusive with key_generator.

  • key_generator

    (Callable[P, Hashable] | None, default: None ) –

    A callable that receives the function's arguments and returns a hashable key. Mutually exclusive with key.

  • package

    (str | None, default: None ) –

    The package namespace. If not provided, the package of the caller is used.

  • return_expired

    (bool, default: False ) –

    Whether to return an expired entry once before it is recomputed and overwritten.

  • ttl

    (timedelta | None, default: None ) –

    Optional time-to-live for stored values.

Returns:

  • Callable[[Callable[P, R]], Callable[P, R]] –

    A decorator that wraps the function and provides cached results.

Notes
  • Only one of key or key_generator must be provided.
  • If return_expired is True, an expired value may be returned once; it is then replaced with a fresh value.

store 

store(
    key: Hashable,
    value: T,
    *,
    package: str | None = None,
    ttl: timedelta | None = None,
) -> CacheEntry[T]

Store a value in the cache.

Parameters:

  • key

    (Hashable) –

    The key under which to store the value. Must be hashable and unique within the given package.

  • value

    (T) –

    The value to store.

  • package

    (str | None, default: None ) –

    The package namespace. If not provided, the package of the caller is used.

  • ttl

    (timedelta | None, default: None ) –

    Optional time-to-live for the entry. When omitted, the entry does not expire.

Returns:

  • CacheEntry[T] –

    The cache entry that was stored.

ChainedStorage 

ChainedStorage(storages: list[Storage])

Bases: Storage

A storage implementation that chains multiple storage backends together.

On retrieval, it tries each storage in order until it finds a hit. On storage, it stores the value in all storage backends.

Parameters:

  • storages

    (list[Storage]) –

    Storage backends to chain. The order defines lookup priority; the first backend with a hit will be used.

CheckResult dataclass 

CheckResult(
    check_state: CheckState,
    summary: str,
    details: Details | None = None,
    name_suffix: str | None = None,
    metrics: list[Metric] | None = None,
    hostname: HostnameInput | None = None,
)

Represents the result of a check performed on a system or component.

This class encapsulates the result of a check operation, providing details such as the state of the check, a summary, optional details, a name suffix, metrics, and hostname information. It is designed to store and convey the outcome of a check operation in a structured format.

Attributes:

  • check_state (CheckState) –

    The state of the check, indicating whether it was successful, warning,

  • details (str | None) –

    A detailed output of the check result, providing additional information

  • hostname (HostnameInput | None) –

    An optional hostname that overrides the hostname that would have been used

  • metrics (list[Metric] | None) –

    An optional list of metrics to associate with the check.

  • name_suffix (str | None) –

    A suffix to add to the check name as defined in the @check decorator.

  • summary (str) –

    A summary of the check result, indicating the outcome of the check.

check_state instance-attribute 

check_state: CheckState = check_state

The state of the check, indicating whether it was successful, warning, critical, or unknown.

details class-attribute instance-attribute 

details: str | None = normalize_details(details)

A detailed output of the check result, providing additional information beyond the summary.

Checkmk will show this detailed output when you are viewing a specific service.

hostname class-attribute instance-attribute 

hostname: HostnameInput | None = hostname

An optional hostname that overrides the hostname that would have been used otherwise.

metrics class-attribute instance-attribute 

metrics: list[Metric] | None = metrics

An optional list of metrics to associate with the check.

name_suffix class-attribute instance-attribute 

name_suffix: str | None = name_suffix

A suffix to add to the check name as defined in the @check decorator.

This enables a single check function to return multiple results that create multiple services on the Checkmk side.

summary instance-attribute 

summary: str = summary

A summary of the check result, indicating the outcome of the check.

Checkmk will show this summary on pages like a host overview.

Datasource

Bases: ABC

Base class for datasources that provide input to checks.

Subclass this to integrate systems your checks read from (APIs, files, services, etc.). Datasources can influence where and when checks are run by exposing scheduling strategies.

Attributes:

scheduling_strategies class-attribute instance-attribute 

scheduling_strategies: (
    tuple[SchedulingStrategy, ...] | EllipsisType | None
) = ...

Optional scheduling strategies that constrain where a check may run.

  • Ellipsis (...) means "unspecified". If a datasource created by a factory leaves this as Ellipsis, Watchpost will fall back to the factory's own scheduling_strategies if any.
  • A tuple provides explicit strategies for this datasource and takes precedence over any factory strategies.
  • None or an empty tuple means "no constraints".

DatasourceFactory

Bases: Protocol

Protocol for factories that construct datasources for checks.

A factory centralizes configuration and creation logic for a datasource. Watchpost can use factory-level scheduling_strategies when the created datasource does not define its own.

Attributes:

new class-attribute 

new: Callable[..., Datasource]

Function that creates and returns a datasource instance.

Watchpost forwards the positional and keyword arguments declared via FromFactory(..., *args, **kwargs) to this callable.

scheduling_strategies class-attribute instance-attribute 

scheduling_strategies: (
    tuple[SchedulingStrategy, ...] | EllipsisType | None
) = ...

Optional scheduling strategies applied to datasources created by this factory.

A tuple sets explicit constraints; None or an empty tuple means "no constraints".

DatasourceUnavailable

Bases: Exception

Raised when a datasource cannot be created, accessed, or used.

Watchpost raises this error when a check requests data, but the underlying system is not reachable, misconfigured, or otherwise unavailable.

DiskStorage 

DiskStorage(directory: str)

Bases: Storage

A disk-backed storage backend using Pickle-serialized CacheEntry objects.

Entries are stored under a versioned directory and sharded by a hash prefix.

Parameters:

  • directory

    (str) –

    Directory root where cache files are written.

Environment 

Environment(
    name: str,
    *,
    hostname: HostnameInput | None = None,
    **metadata: Hashable,
)

Represents a logical environment in which checks run.

An environment typically maps to how or where checks are executed (for example: dev, stage, prod, monitoring). It can carry a hostname resolution strategy and arbitrary metadata that checks and data sources may use to alter behavior.

Parameters:

  • name

    (str) –

    The name of the environment (e.g., "dev", "stage", "prod", "monitoring").

  • hostname

    (HostnameInput | None, default: None ) –

    Input used to determine the hostname when running checks in this environment. Accepts the same forms as HostnameInput such as a template string, a callable, or a strategy instance. If omitted, the application may fall back to its default hostname generation.

  • metadata

    (Hashable, default: {} ) –

    Arbitrary, hashable metadata attached to the environment. Checks and data sources may consult these values to change behavior.

Notes

Hostname handling integrates with the hostname resolution system in watchpost.hostname. The chosen strategy ultimately influences the Checkmk host a service is associated with.

EnvironmentRegistry 

EnvironmentRegistry()

Simple in-memory registry of Environment objects.

The registry provides dictionary-like access and helpers to create and manage environments used by a Watchpost application.

Methods:

  • add –

    Add or replace an environment by its name.

  • get –

    Return the environment for the given name, or a default if missing.

  • new –

    Create a new environment, add it to the registry, and return it.

add 

add(environment: Environment) -> None

Add or replace an environment by its name.

Parameters:

  • environment

    (Environment) –

    The environment to register. If another environment with the same name exists, it will be overwritten.

get 

get(
    name: str, default: Environment | None = None
) -> Environment | None

Return the environment for the given name, or a default if missing.

Parameters:

  • name

    (str) –

    The environment name to look up.

  • default

    (Environment | None, default: None ) –

    Value to return when the name is not present.

Returns:

  • Environment | None –

    The matching Environment, or the provided default.

new 

new(
    name: str,
    *,
    hostname: HostnameInput | None = None,
    **metadata: Hashable,
) -> Environment

Create a new environment, add it to the registry, and return it.

Parameters:

  • name

    (str) –

    The name of the environment to create.

  • hostname

    (HostnameInput | None, default: None ) –

    Input used to determine the hostname for this environment. See Environment.__init__ for supported forms.

  • metadata

    (Hashable, default: {} ) –

    Arbitrary, hashable metadata to attach to the environment.

Returns:

  • Environment –

    The created and registered Environment instance.

FromFactory 

FromFactory(
    factory: type[DatasourceFactory],
    *args: Any,
    **kwargs: Any,
)

FromFactory(*args: Any, **kwargs: Any)

FromFactory(
    factory: type[DatasourceFactory] | Any = None,
    *args: Any,
    **kwargs: Any,
)

Marker used with typing.Annotated to request a datasource from a factory.

Use in a check parameter annotation as:

Annotated[MyDatasource, FromFactory(MyFactory, "service")]

You may omit the factory type when the parameter type implements the factory protocol:

Annotated[MyDatasourceWithFactory, FromFactory("service")]

Parameters are forwarded to the factory's new callable. Watchpost caches the constructed datasource per factory type and argument set.

Parameters:

  • factory

    (type[DatasourceFactory] | Any, default: None ) –

    Optional factory type used to create the datasource. If omitted, Watchpost infers the factory from the annotated parameter's type. If the first positional argument is not a type, it is treated as a factory argument and the factory is inferred.

  • args

    (Any, default: () ) –

    Positional arguments forwarded to the factory new callable.

  • kwargs

    (Any, default: {} ) –

    Keyword arguments forwarded to the factory new callable.

Methods:

  • cache_key –

    Generate a stable cache key for this factory invocation.

cache_key 

cache_key(
    type_key: type[DatasourceFactory] | None,
) -> tuple[type[DatasourceFactory] | None, int, int]

Generate a stable cache key for this factory invocation.

Parameters:

  • type_key

    (type[DatasourceFactory] | None) –

    Fallback factory type to use when this instance did not specify a factory explicitly (i.e., it will be inferred from the annotated parameter type).

Returns:

  • type[DatasourceFactory] | None –

    A tuple that identifies the factory type and the provided arguments,

  • int –

    suitable for use as a cache key.

InMemoryStorage 

InMemoryStorage()

Bases: Storage

A simple in-memory storage backend.

Useful for tests or ephemeral caching within a single process. Entries are lost when the process exits.

Metric dataclass 

Metric(
    name: str,
    value: int | float,
    levels: Thresholds | None = None,
    boundaries: Thresholds | None = None,
)

Represents a measurable metric with an optional threshold configuration.

This maps to the Checkmk concept of metrics exactly.

MustRunAgainstGivenTargetEnvironmentStrategy 

MustRunAgainstGivenTargetEnvironmentStrategy(
    *environments: Environment,
)

Bases: SchedulingStrategy

Restrict which target environments the check may run against.

Use this to allow only specific targets (for example, only "stage"). If the requested target is not in the allowed set, the decision is DONT_SCHEDULE.

This strategy does not affect from where the check executes, i.e. the execution environment.

    One or more target environments that the check may run against.

MustRunInGivenExecutionEnvironmentStrategy 

MustRunInGivenExecutionEnvironmentStrategy(
    *environments: Environment,
)

Bases: SchedulingStrategy

Require that the check executes from one of the given execution environments.

Use this when a datasource or check can only run from specific locations (for example, from a monitoring environment). If the current execution environment is not in the allowed set, the decision is DONT_SCHEDULE.

    One or more execution environments from which the check may run.

MustRunInTargetEnvironmentStrategy

Bases: SchedulingStrategy

Require that the current execution environment equals the target environment.

This models "run in-environment" behavior where the check must execute inside the environment it is checking. If current != target, the decision is DONT_SCHEDULE.

RedisStorage 

RedisStorage(
    redis_client: Redis,
    *,
    use_redis_ttl: bool = True,
    redis_key_infix: str | None = None,
)

Bases: Storage

A Redis-backed storage backend.

When use_redis_ttl is True, Redis key expiration is used so expired entries are never returned by get.

Parameters:

  • redis_client

    (Redis) –

    The Redis client instance.

  • use_redis_ttl

    (bool, default: True ) –

    Whether to use Redis TTL for automatic expiry. When True, expired entries are not returned.

  • redis_key_infix

    (str | None, default: None ) –

    Optional infix to namespace keys to avoid collisions between multiple caches or watchposts.

Methods:

  • get –

    Retrieve a cache entry from Redis.

  • store –

    Store a cache entry in Redis.

get 

get(
    cache_key: CacheKey, return_expired: bool = False
) -> CacheEntry[T] | None

Retrieve a cache entry from Redis.

Parameters:

  • cache_key

    (CacheKey) –

    The key to retrieve the value for.

  • return_expired

    (bool, default: False ) –

    Whether to return an expired entry once before it is removed. (If Redis's TTL is used, an expired entry is never returned and this has no effect.)

Returns:

  • CacheEntry[T] | None –

    The cache entry if found, otherwise None.

store 

store(entry: CacheEntry) -> None

Store a cache entry in Redis.

Parameters:

  • entry

    (CacheEntry) –

    The cache entry to store.

Thresholds dataclass 

Thresholds(warning: int | float, critical: int | float)

Represents threshold values with warning and critical levels.

This maps to the Checkmk concept of thresholds exactly and thus directly relates to metrics, see the Metric class.

Watchpost 

Watchpost(
    *,
    checks: list[Check | ModuleType],
    execution_environment: Environment,
    version: str = "unknown",
    max_workers: int | None = None,
    executor: CheckExecutor[list[ExecutionResult]]
    | None = None,
    check_cache_storage: Storage | None = None,
    default_scheduling_strategies: list[SchedulingStrategy]
    | None = None,
    hostname: HostnameInput | None = None,
    hostname_fallback_to_default_hostname_generation: bool = True,
    hostname_coerce_into_valid_hostname: bool = True,
)

Main Watchpost application and ASGI app.

Watchpost discovers and runs checks across environments, coordinates datasources, applies scheduling strategies, and generates Checkmk-compatible output. A Watchpost instance is also an ASGI application backed by Starlette and exposes operational endpoints.

Notes

The instance manages a per-check result cache and a key-aware executor to run checks without blocking. It also verifies check configuration and hostname generation at application startup.

Parameters:

  • checks

    (list[Check | ModuleType]) –

    A list of Check objects or Python modules to scan for checks. Modules are discovered recursively.

  • execution_environment

    (Environment) –

    The environment in which this Watchpost instance runs checks.

  • version

    (str, default: 'unknown' ) –

    Version string included in the Checkmk agent header output.

  • max_workers

    (int | None, default: None ) –

    Maximum number of worker threads for the internal executor. Used when no custom executor is provided.

  • executor

    (CheckExecutor[list[ExecutionResult]] | None, default: None ) –

    Optional custom CheckExecutor to use instead of the default.

  • check_cache_storage

    (Storage | None, default: None ) –

    Optional storage backend for the per-check result cache. Defaults to in-memory storage.

  • default_scheduling_strategies

    (list[SchedulingStrategy] | None, default: None ) –

    Default strategies applied to all checks in addition to those specified by checks and datasources. If not provided, the DetectImpossibleCombinationStrategy will be applied by default.

  • hostname

    (HostnameInput | None, default: None ) –

    Optional Watchpost-level hostname strategy or value used to resolve the piggyback host for results.

  • hostname_fallback_to_default_hostname_generation

    (bool, default: True ) –

    Whether to fall back to the default hostname generation "{service_name}-{environment.name}" when no strategy resolves a hostname.

  • hostname_coerce_into_valid_hostname

    (bool, default: True ) –

    Whether to coerce a non-compliant hostname into RFC1123 format during hostname resolution. If False, a non-compliant hostname will result in an error.

Methods:

app_context 

app_context() -> Generator[Watchpost]

Provide a context where the current Watchpost instance is active.

This sets the global context variable so helper utilities can access the current application instance during check execution.

Returns:

  • Generator[Watchpost] –

    The current Watchpost instance via a context manager.

register_datasource 

register_datasource(
    datasource_type: type[_D], **kwargs: dict[str, Any]
) -> None

Register a datasource type and its constructor arguments.

Parameters:

  • datasource_type

    (type[_D]) –

    The Datasource subclass to register.

  • kwargs

    (dict[str, Any], default: {} ) –

    Keyword arguments to use when instantiating the datasource.

register_datasource_factory 

register_datasource_factory(
    factory_type: type[_DF],
) -> None

Register a DatasourceFactory that can produce datasources.

Parameters:

  • factory_type

    (type[_DF]) –

    The factory type to register.

run_check 

run_check(
    check: Check,
    *,
    custom_executor: CheckExecutor[list[ExecutionResult]]
    | None = None,
    use_cache: bool = True,
) -> Generator[ExecutionResult]

Run a single check across all its target environments.

Parameters:

  • check

    (Check) –

    The Check to run.

  • custom_executor

    (CheckExecutor[list[ExecutionResult]] | None, default: None ) –

    Optional executor used only for this call.

  • use_cache

    (bool, default: True ) –

    Whether to use and update the per-check cache.

Yields:

  • Generator[ExecutionResult] –

    ExecutionResult objects produced by the check for each environment.

run_checks 

run_checks(act_as_agent: bool = True) -> Generator[bytes]

Run all checks and produce a Checkmk-compatible output stream.

This yields the agent header, the serialized results of all checks, and synthetic sections.

Parameters:

  • act_as_agent

    (bool, default: True ) –

    If Watchpost should act as a full Checkmk agent, including the <<<checkmk>>> preamble. This should be true if Watchpost is queried by the Checkmk site directly (via HTTP), but can be set to false if you have the Checkmk agent invoke Watchpost for you, for example.

Yields:

  • Generator[bytes] –

    Bytes in Checkmk agent format.

run_checks_once 

run_checks_once(act_as_agent: bool = True) -> None

Run all the checks once and write the output stream to stdout.

This is a convenience method primarily intended for CLI usage.

Parameters:

  • act_as_agent

    (bool, default: True ) –

    If Watchpost should act as a full Checkmk agent, including the <<<checkmk>>> preamble. This should be true if Watchpost is queried by the Checkmk site directly (via HTTP), but can be set to false if you have the Checkmk agent invoke Watchpost for you, for example.

verify_check_scheduling 

verify_check_scheduling(force: bool = False) -> None

Validate that checks can be scheduled and invoked correctly.

This verifies argument provisioning for each check and evaluates each strategy's configuration checks. It aggregates errors across all checks.

Parameters:

  • force

    (bool, default: False ) –

    Run verification even if it has already completed successfully.

Raises:

  • ExceptionGroup –

    If one or more checks are misconfigured.

verify_hostname_generation 

verify_hostname_generation(force: bool = False) -> None

Validate that hostnames can be resolved for all checks and environments.

Parameters:

  • force

    (bool, default: False ) –

    Run verification even if it has already completed successfully.

Raises:

  • ExceptionGroup –

    If hostname resolution fails for any check/environment.

build_result 

build_result(
    ok_summary: str,
    fail_summary: str,
    base_details: Details | None = None,
    name_suffix: str | None = None,
    metrics: list[Metric] | None = None,
    alternative_hostname: HostnameInput | None = None,
) -> OngoingCheckResult

Start building up a new check result that allows adding multiple OK/WARN/UNKNOWN/CRIT results, called "partials", eventually resulting in a regular check result that holds the worst check state provided by the individual results.

Use of this builder greatly simplifies scenarios where your check function validates multiple aspects of a single system or component but only returns a single check result: by allowing you to provide the status of each aspect as you check it through simple function calls on the builder instead of having to manually combine the results into a single check result, you can reduce the complexity of your check function and improve its readability.

Parameters:

  • ok_summary

    (str) –

    Overall check summary that will be shown if the final result is OK.

  • fail_summary

    (str) –

    Overall check summary that will be shown if the final result is WARN, UNKNOWN, or CRIT.

  • base_details

    (Details | None, default: None ) –

    Optional base details for the check result. They will always be included and then enriched by the details provided by partial results.

  • name_suffix

    (str | None, default: None ) –

    Optional suffix for the check result name.

  • metrics

    (list[Metric] | None, default: None ) –

    Optional list of metrics associated with the check.

  • alternative_hostname

    (HostnameInput | None, default: None ) –

    Optional alternative hostname input for the check.

Returns:

crit 

crit(
    summary: str,
    details: Details | None = None,
    name_suffix: str | None = None,
    metrics: list[Metric] | None = None,
    alternative_hostname: HostnameInput | None = None,
) -> CheckResult

Generates a CheckResult object indicating a CRIT check state.

This function creates and returns a CheckResult indicating the system or component is in a CRIT state. It allows for passing additional information such as a summary, details, name suffix, metrics, or an alternative hostname.

Parameters:

  • summary

    (str) –

    A summary of the check result, indicating the outcome of the check. Checkmk will show this summary on pages like a host overview.

  • details

    (Details | None, default: None ) –

    A detailed output of the check result, providing additional information beyond the summary. Checkmk will show this detailed output when you are viewing a specific service.

  • name_suffix

    (str | None, default: None ) –

    A suffix to add to the check name as defined in the @check decorator. This enables a single check function to return multiple results that create multiple services on the Checkmk side.

  • metrics

    (list[Metric] | None, default: None ) –

    An optional list of metrics to associate with the check.

  • alternative_hostname

    (HostnameInput | None, default: None ) –

    An optional hostname that overrides the hostname that would have been used otherwise.

Returns:

  • CheckResult –

    A CheckResult object representing the CRIT check result.

ok 

ok(
    summary: str,
    details: Details | None = None,
    name_suffix: str | None = None,
    metrics: list[Metric] | None = None,
    alternative_hostname: HostnameInput | None = None,
) -> CheckResult

Generates a CheckResult object indicating an OK check state.

This function creates and returns a CheckResult indicating the system or component is in an OK state. It allows for passing additional information such as a summary, details, name suffix, metrics, or an alternative hostname.

Parameters:

  • summary

    (str) –

    A summary of the check result, indicating the outcome of the check. Checkmk will show this summary on pages like a host overview.

  • details

    (Details | None, default: None ) –

    A detailed output of the check result, providing additional information beyond the summary. Checkmk will show this detailed output when you are viewing a specific service.

  • name_suffix

    (str | None, default: None ) –

    A suffix to add to the check name as defined in the @check decorator. This enables a single check function to return multiple results that create multiple services on the Checkmk side.

  • metrics

    (list[Metric] | None, default: None ) –

    An optional list of metrics to associate with the check.

  • alternative_hostname

    (HostnameInput | None, default: None ) –

    An optional hostname that overrides the hostname that would have been used otherwise.

Returns:

  • CheckResult –

    A CheckResult object representing the OK check result.

unknown 

unknown(
    summary: str,
    details: Details | None = None,
    name_suffix: str | None = None,
    metrics: list[Metric] | None = None,
    alternative_hostname: HostnameInput | None = None,
) -> CheckResult

Generates a CheckResult object indicating an UNKNOWN check state.

This function creates and returns a CheckResult indicating the system or component is in an UNKNOWN state. It allows for passing additional information such as a summary, details, name suffix, metrics, or an alternative hostname.

Parameters:

  • summary

    (str) –

    A summary of the check result, indicating the outcome of the check. Checkmk will show this summary on pages like a host overview.

  • details

    (Details | None, default: None ) –

    A detailed output of the check result, providing additional information beyond the summary. Checkmk will show this detailed output when you are viewing a specific service.

  • name_suffix

    (str | None, default: None ) –

    A suffix to add to the check name as defined in the @check decorator. This enables a single check function to return multiple results that create multiple services on the Checkmk side.

  • metrics

    (list[Metric] | None, default: None ) –

    An optional list of metrics to associate with the check.

  • alternative_hostname

    (HostnameInput | None, default: None ) –

    An optional hostname that overrides the hostname that would have been used otherwise.

Returns:

  • CheckResult –

    A CheckResult object representing the UNKNOWN check result.

warn 

warn(
    summary: str,
    details: Details | None = None,
    name_suffix: str | None = None,
    metrics: list[Metric] | None = None,
    alternative_hostname: HostnameInput | None = None,
) -> CheckResult

Generates a CheckResult object indicating a WARN check state.

This function creates and returns a CheckResult indicating the system or component is in a WARN state. It allows for passing additional information such as a summary, details, name suffix, metrics, or an alternative hostname.

Parameters:

  • summary

    (str) –

    A summary of the check result, indicating the outcome of the check. Checkmk will show this summary on pages like a host overview.

  • details

    (Details | None, default: None ) –

    A detailed output of the check result, providing additional information beyond the summary. Checkmk will show this detailed output when you are viewing a specific service.

  • name_suffix

    (str | None, default: None ) –

    A suffix to add to the check name as defined in the @check decorator. This enables a single check function to return multiple results that create multiple services on the Checkmk side.

  • metrics

    (list[Metric] | None, default: None ) –

    An optional list of metrics to associate with the check.

  • alternative_hostname

    (HostnameInput | None, default: None ) –

    An optional hostname that overrides the hostname that would have been used otherwise.

Returns:

  • CheckResult –

    A CheckResult object representing the WARN check result.

app

Watchpost application and Starlette integration.

This module provides the Watchpost ASGI application that discovers and runs checks, manages datasources, applies scheduling strategies, resolves hostnames, coordinates execution via an internal executor, and exposes HTTP endpoints.

Notes
  • The app streams Checkmk-compatible output from /.
  • Operational endpoints expose executor statistics and errors.
  • Global app context is available via current_app in globals.py.

Classes:

  • Watchpost –

    Main Watchpost application and ASGI app.

Watchpost 

Watchpost(
    *,
    checks: list[Check | ModuleType],
    execution_environment: Environment,
    version: str = "unknown",
    max_workers: int | None = None,
    executor: CheckExecutor[list[ExecutionResult]]
    | None = None,
    check_cache_storage: Storage | None = None,
    default_scheduling_strategies: list[SchedulingStrategy]
    | None = None,
    hostname: HostnameInput | None = None,
    hostname_fallback_to_default_hostname_generation: bool = True,
    hostname_coerce_into_valid_hostname: bool = True,
)

Main Watchpost application and ASGI app.

Watchpost discovers and runs checks across environments, coordinates datasources, applies scheduling strategies, and generates Checkmk-compatible output. A Watchpost instance is also an ASGI application backed by Starlette and exposes operational endpoints.

Notes

The instance manages a per-check result cache and a key-aware executor to run checks without blocking. It also verifies check configuration and hostname generation at application startup.

Parameters:

  • checks

    (list[Check | ModuleType]) –

    A list of Check objects or Python modules to scan for checks. Modules are discovered recursively.

  • execution_environment

    (Environment) –

    The environment in which this Watchpost instance runs checks.

  • version

    (str, default: 'unknown' ) –

    Version string included in the Checkmk agent header output.

  • max_workers

    (int | None, default: None ) –

    Maximum number of worker threads for the internal executor. Used when no custom executor is provided.

  • executor

    (CheckExecutor[list[ExecutionResult]] | None, default: None ) –

    Optional custom CheckExecutor to use instead of the default.

  • check_cache_storage

    (Storage | None, default: None ) –

    Optional storage backend for the per-check result cache. Defaults to in-memory storage.

  • default_scheduling_strategies

    (list[SchedulingStrategy] | None, default: None ) –

    Default strategies applied to all checks in addition to those specified by checks and datasources. If not provided, the DetectImpossibleCombinationStrategy will be applied by default.

  • hostname

    (HostnameInput | None, default: None ) –

    Optional Watchpost-level hostname strategy or value used to resolve the piggyback host for results.

  • hostname_fallback_to_default_hostname_generation

    (bool, default: True ) –

    Whether to fall back to the default hostname generation "{service_name}-{environment.name}" when no strategy resolves a hostname.

  • hostname_coerce_into_valid_hostname

    (bool, default: True ) –

    Whether to coerce a non-compliant hostname into RFC1123 format during hostname resolution. If False, a non-compliant hostname will result in an error.

Methods:

app_context 

app_context() -> Generator[Watchpost]

Provide a context where the current Watchpost instance is active.

This sets the global context variable so helper utilities can access the current application instance during check execution.

Returns:

  • Generator[Watchpost] –

    The current Watchpost instance via a context manager.

register_datasource 

register_datasource(
    datasource_type: type[_D], **kwargs: dict[str, Any]
) -> None

Register a datasource type and its constructor arguments.

Parameters:

  • datasource_type
    (type[_D]) –

    The Datasource subclass to register.

  • kwargs
    (dict[str, Any], default: {} ) –

    Keyword arguments to use when instantiating the datasource.

register_datasource_factory 

register_datasource_factory(
    factory_type: type[_DF],
) -> None

Register a DatasourceFactory that can produce datasources.

Parameters:

  • factory_type
    (type[_DF]) –

    The factory type to register.

run_check 

run_check(
    check: Check,
    *,
    custom_executor: CheckExecutor[list[ExecutionResult]]
    | None = None,
    use_cache: bool = True,
) -> Generator[ExecutionResult]

Run a single check across all its target environments.

Parameters:

  • check
    (Check) –

    The Check to run.

  • custom_executor
    (CheckExecutor[list[ExecutionResult]] | None, default: None ) –

    Optional executor used only for this call.

  • use_cache
    (bool, default: True ) –

    Whether to use and update the per-check cache.

Yields:

  • Generator[ExecutionResult] –

    ExecutionResult objects produced by the check for each environment.

run_checks 

run_checks(act_as_agent: bool = True) -> Generator[bytes]

Run all checks and produce a Checkmk-compatible output stream.

This yields the agent header, the serialized results of all checks, and synthetic sections.

Parameters:

  • act_as_agent
    (bool, default: True ) –

    If Watchpost should act as a full Checkmk agent, including the <<<checkmk>>> preamble. This should be true if Watchpost is queried by the Checkmk site directly (via HTTP), but can be set to false if you have the Checkmk agent invoke Watchpost for you, for example.

Yields:

  • Generator[bytes] –

    Bytes in Checkmk agent format.

run_checks_once 

run_checks_once(act_as_agent: bool = True) -> None

Run all the checks once and write the output stream to stdout.

This is a convenience method primarily intended for CLI usage.

Parameters:

  • act_as_agent
    (bool, default: True ) –

    If Watchpost should act as a full Checkmk agent, including the <<<checkmk>>> preamble. This should be true if Watchpost is queried by the Checkmk site directly (via HTTP), but can be set to false if you have the Checkmk agent invoke Watchpost for you, for example.

verify_check_scheduling 

verify_check_scheduling(force: bool = False) -> None

Validate that checks can be scheduled and invoked correctly.

This verifies argument provisioning for each check and evaluates each strategy's configuration checks. It aggregates errors across all checks.

Parameters:

  • force
    (bool, default: False ) –

    Run verification even if it has already completed successfully.

Raises:

  • ExceptionGroup –

    If one or more checks are misconfigured.

verify_hostname_generation 

verify_hostname_generation(force: bool = False) -> None

Validate that hostnames can be resolved for all checks and environments.

Parameters:

  • force
    (bool, default: False ) –

    Run verification even if it has already completed successfully.

Raises:

  • ExceptionGroup –

    If hostname resolution fails for any check/environment.

cache

Caching utilities and pluggable storage backends.

This module provides a small, consistent caching API (Cache) and a set of storage backends (Storage) for in-memory, disk, and Redis persistence. It also includes a memoization decorator for function-level caching.

Classes:

  • Cache –

    High-level caching API using a pluggable storage backend.

  • CacheEntry –

    Represents a cached value together with metadata.

  • CacheKey –

    A compound key that uniquely identifies a cache entry within a package.

  • ChainedStorage –

    A storage implementation that chains multiple storage backends together.

  • DiskStorage –

    A disk-backed storage backend using Pickle-serialized CacheEntry objects.

  • InMemoryStorage –

    A simple in-memory storage backend.

  • RedisStorage –

    A Redis-backed storage backend.

  • Storage –

    Base interface for cache storage backends.

Functions:

Cache 

Cache(storage: Storage)

High-level caching API using a pluggable storage backend.

Use get and store to interact with the cache directly, or the memoize decorator to cache function results.

Methods:

  • get –

    Retrieve a value from the cache.

  • memoize –

    Memoize a function by caching its return value.

  • store –

    Store a value in the cache.

get 

get(
    key: Hashable,
    default: T | None = None,
    *,
    package: str | None = None,
    return_expired: bool = False,
) -> CacheEntry[T] | None

Retrieve a value from the cache.

Parameters:

  • key
    (Hashable) –

    The key to look up. Must be hashable and unique within the given package.

  • default
    (T | None, default: None ) –

    An optional default value to return when the key is not found.

  • package
    (str | None, default: None ) –

    The package namespace. If not provided, the package of the caller is used.

  • return_expired
    (bool, default: False ) –

    Whether to return an expired entry once before it is removed. (This behavior does depend on the storage backend supporting this.)

Returns:

  • CacheEntry[T] | None –

    The cache entry if found. If not found, returns None when no

  • CacheEntry[T] | None –

    default is given, otherwise a cache entry wrapping the default

  • CacheEntry[T] | None –

    value.

memoize 

memoize(
    *,
    key: Hashable | None = None,
    key_generator: Callable[P, Hashable] | None = None,
    package: str | None = None,
    return_expired: bool = False,
    ttl: timedelta | None = None,
) -> Callable[[Callable[P, R]], Callable[P, R]]

Memoize a function by caching its return value.

Use either a fixed key or a key_generator to compute the cache key from call arguments. If key contains format placeholders like "{arg}", they will be formatted from the function's bound arguments.

Parameters:

  • key
    (Hashable | None, default: None ) –

    A static key or a format string used to build the key. Mutually exclusive with key_generator.

  • key_generator
    (Callable[P, Hashable] | None, default: None ) –

    A callable that receives the function's arguments and returns a hashable key. Mutually exclusive with key.

  • package
    (str | None, default: None ) –

    The package namespace. If not provided, the package of the caller is used.

  • return_expired
    (bool, default: False ) –

    Whether to return an expired entry once before it is recomputed and overwritten.

  • ttl
    (timedelta | None, default: None ) –

    Optional time-to-live for stored values.

Returns:

  • Callable[[Callable[P, R]], Callable[P, R]] –

    A decorator that wraps the function and provides cached results.

Notes
  • Only one of key or key_generator must be provided.
  • If return_expired is True, an expired value may be returned once; it is then replaced with a fresh value.

store 

store(
    key: Hashable,
    value: T,
    *,
    package: str | None = None,
    ttl: timedelta | None = None,
) -> CacheEntry[T]

Store a value in the cache.

Parameters:

  • key
    (Hashable) –

    The key under which to store the value. Must be hashable and unique within the given package.

  • value
    (T) –

    The value to store.

  • package
    (str | None, default: None ) –

    The package namespace. If not provided, the package of the caller is used.

  • ttl
    (timedelta | None, default: None ) –

    Optional time-to-live for the entry. When omitted, the entry does not expire.

Returns:

  • CacheEntry[T] –

    The cache entry that was stored.

CacheEntry dataclass 

CacheEntry(
    cache_key: CacheKey,
    value: T,
    added_at: datetime | None,
    ttl: timedelta | None,
)

Represents a cached value together with metadata.

This object stores the value and all information needed by backends to determine freshness and validity.

Methods:

  • is_expired –

    Determine whether the cache entry has expired.

Attributes:

  • added_at (datetime | None) –

    The timestamp when this entry was stored. None means the value is a

  • cache_key (CacheKey) –

    The globally unique cache key for this entry.

  • ttl (timedelta | None) –

    The time-to-live for this entry. None means the value does not expire.

  • value (T) –

    The cached value.

added_at instance-attribute 

added_at: datetime | None

The timestamp when this entry was stored. None means the value is a default and not persisted yet.

cache_key instance-attribute 

cache_key: CacheKey

The globally unique cache key for this entry.

ttl instance-attribute 

ttl: timedelta | None

The time-to-live for this entry. None means the value does not expire.

value instance-attribute 

value: T

The cached value.

is_expired 

is_expired() -> bool

Determine whether the cache entry has expired.

Returns:

  • bool –

    True if the entry is expired; otherwise False.

CacheKey dataclass 

CacheKey(key: Hashable, package: str)

A compound key that uniquely identifies a cache entry within a package.

The combination of key and package forms the globally unique key used by storage backends.

Attributes:

  • key (Hashable) –

    The user-provided key. It is unique within the given package.

  • package (str) –

    The package namespace used to separate keys across applications or checks.

key instance-attribute 

key: Hashable

The user-provided key. It is unique within the given package.

package instance-attribute 

package: str

The package namespace used to separate keys across applications or checks.

ChainedStorage 

ChainedStorage(storages: list[Storage])

Bases: Storage

A storage implementation that chains multiple storage backends together.

On retrieval, it tries each storage in order until it finds a hit. On storage, it stores the value in all storage backends.

Parameters:

  • storages

    (list[Storage]) –

    Storage backends to chain. The order defines lookup priority; the first backend with a hit will be used.

DiskStorage 

DiskStorage(directory: str)

Bases: Storage

A disk-backed storage backend using Pickle-serialized CacheEntry objects.

Entries are stored under a versioned directory and sharded by a hash prefix.

Parameters:

  • directory

    (str) –

    Directory root where cache files are written.

InMemoryStorage 

InMemoryStorage()

Bases: Storage

A simple in-memory storage backend.

Useful for tests or ephemeral caching within a single process. Entries are lost when the process exits.

RedisStorage 

RedisStorage(
    redis_client: Redis,
    *,
    use_redis_ttl: bool = True,
    redis_key_infix: str | None = None,
)

Bases: Storage

A Redis-backed storage backend.

When use_redis_ttl is True, Redis key expiration is used so expired entries are never returned by get.

Parameters:

  • redis_client

    (Redis) –

    The Redis client instance.

  • use_redis_ttl

    (bool, default: True ) –

    Whether to use Redis TTL for automatic expiry. When True, expired entries are not returned.

  • redis_key_infix

    (str | None, default: None ) –

    Optional infix to namespace keys to avoid collisions between multiple caches or watchposts.

Methods:

  • get –

    Retrieve a cache entry from Redis.

  • store –

    Store a cache entry in Redis.

get 

get(
    cache_key: CacheKey, return_expired: bool = False
) -> CacheEntry[T] | None

Retrieve a cache entry from Redis.

Parameters:

  • cache_key
    (CacheKey) –

    The key to retrieve the value for.

  • return_expired
    (bool, default: False ) –

    Whether to return an expired entry once before it is removed. (If Redis's TTL is used, an expired entry is never returned and this has no effect.)

Returns:

  • CacheEntry[T] | None –

    The cache entry if found, otherwise None.

store 

store(entry: CacheEntry) -> None

Store a cache entry in Redis.

Parameters:

  • entry
    (CacheEntry) –

    The cache entry to store.

Storage

Bases: ABC

Base interface for cache storage backends.

Implementations persist and retrieve CacheEntry objects.

Methods:

  • get –

    Retrieve a cache entry by key.

  • store –

    Persist a cache entry.

get abstractmethod 

get(
    cache_key: CacheKey, return_expired: bool = False
) -> CacheEntry[T] | None

Retrieve a cache entry by key.

Parameters:

  • cache_key
    (CacheKey) –

    The cache key to look up.

  • return_expired
    (bool, default: False ) –

    Whether to return an expired entry once before it is removed.

Returns:

  • CacheEntry[T] | None –

    The cache entry if found, otherwise None.

store abstractmethod 

store(entry: CacheEntry) -> None

Persist a cache entry.

Parameters:

  • entry
    (CacheEntry) –

    The cache entry to store.

get_caller_package 

get_caller_package() -> str

Return the caller's package name.

This helper inspects the call stack to determine the package of the code invoking the cache. It falls back to the module name when no package is set.

Returns:

  • str –

    The package name of the caller.

check

Check definitions, execution helpers, and caching.

This module provides the core pieces for defining and running monitoring checks:

  • Check: a lightweight wrapper around a user-defined check function plus metadata such as the service name, labels, environments, and hostname strategy.
  • check: a decorator to declare checks in a Pythonic way.
  • CheckCache: a small helper that stores and retrieves executed check results.

Notes:

  • A Check maps to one or more Checkmk services. The application turns results produced by a check into Checkmk-compatible output later. Service labels map to Checkmk service labels.
  • Both synchronous and asynchronous check functions are supported.
  • Hostname resolution (the Checkmk piggyback host) is delegated to the hostname strategy; see watchpost.hostname for details.

Classes:

  • Check –

    Represents a monitoring check definition.

  • CheckCache –

    Caches executed check results per check and environment.

  • ErrorHandler –

    A strategy for transforming error results to match a check's service

Functions:

  • check –

    Decorator to define a Watchpost check function.

Check dataclass 

Check(
    check_function: CheckFunction,
    service_name: str,
    service_labels: dict[str, Any],
    environments: list[Environment],
    cache_for: timedelta | None,
    invocation_information: InvocationInformation
    | None = None,
    scheduling_strategies: list[SchedulingStrategy]
    | None = None,
    hostname_strategy: HostnameStrategy | None = None,
    error_handlers: list[ErrorHandler] | None = None,
)

Represents a monitoring check definition.

A Check wraps a user-defined check function together with metadata such as the service name, labels, environments, caching, scheduling, and hostname strategy. A Check maps to one or more Checkmk services; each execution produces one or more results that are later rendered as Checkmk-compatible output.

Methods:

  • get_function_kwargs –

    Build the keyword arguments to call the check function.

  • run_async –

    Execute the check function asynchronously and return ExecutionResults.

  • run_sync –

    Execute the check function synchronously and return ExecutionResults.

Attributes:

  • cache_for (timedelta | None) –

    Optional time-to-live for caching check results. If set, results are cached

  • check_function (CheckFunction) –

    The user-provided function that implements the check. It may be synchronous

  • environments (list[Environment]) –

    The environments against which this check is executed.

  • error_handlers (list[ErrorHandler] | None) –

    Optional error handlers that are called to adapt the results when the check

  • hostname_strategy (HostnameStrategy | None) –

    Strategy to resolve the piggyback host (hostname) for results of this

  • invocation_information (InvocationInformation | None) –

    Source location information for the check definition (file and line).

  • is_async (bool) –

    Indicates whether the check function is asynchronous.

  • name (str) –

    Returns the fully qualified name of the check function.

  • scheduling_strategies (list[SchedulingStrategy] | None) –

    Optional scheduling strategies that influence when, where and how the check

  • service_labels (dict[str, Any]) –

    Labels attached to the service. This maps to Checkmk service labels exactly.

  • service_name (str) –

    The base service name used for the Checkmk service created by this check.

  • signature (Signature) –

    Returns the cached inspect.Signature of the check function.

  • type_hints (dict[str, Any]) –

    Returns the resolved type hints for the check function's parameters.

cache_for instance-attribute 

cache_for: timedelta | None

Optional time-to-live for caching check results. If set, results are cached for this duration, otherwise they are always re-executed after Checkmk picks up results.

check_function instance-attribute 

check_function: CheckFunction

The user-provided function that implements the check. It may be synchronous or asynchronous and may accept one or more Datasource instances and optionally an environment: Environment parameter.

environments instance-attribute 

environments: list[Environment]

The environments against which this check is executed.

error_handlers class-attribute instance-attribute 

error_handlers: list[ErrorHandler] | None = None

Optional error handlers that are called to adapt the results when the check function raises an exception.

hostname_strategy class-attribute instance-attribute 

hostname_strategy: HostnameStrategy | None = None

Strategy to resolve the piggyback host (hostname) for results of this check. If not set here, environment- or app-level strategies may apply.

invocation_information class-attribute instance-attribute 

invocation_information: InvocationInformation | None = None

Source location information for the check definition (file and line). Used for diagnostics and included in output.

is_async property 

is_async: bool

Indicates whether the check function is asynchronous.

Returns True if the function is a coroutine or a coroutine function, otherwise False.

name property 

name: str

Returns the fully qualified name of the check function.

This is used as a stable identifier in places like caching keys and diagnostics.

scheduling_strategies class-attribute instance-attribute 

scheduling_strategies: list[SchedulingStrategy] | None = (
    None
)

Optional scheduling strategies that influence when, where and how the check is run.

service_labels instance-attribute 

service_labels: dict[str, Any]

Labels attached to the service. This maps to Checkmk service labels exactly.

service_name instance-attribute 

service_name: str

The base service name used for the Checkmk service created by this check. A result may append a name_suffix to this name to create multiple services.

signature property 

signature: Signature

Returns the cached inspect.Signature of the check function.

type_hints property 

type_hints: dict[str, Any]

Returns the resolved type hints for the check function's parameters.

The mapping excludes the return annotation. If resolving forward references raises NameError, it falls back to a mapping derived from the function signature.

get_function_kwargs 

get_function_kwargs(
    *,
    environment: Environment,
    datasources: dict[str, Datasource],
) -> dict[str, Environment | Datasource]

Build the keyword arguments to call the check function.

Parameters:

  • environment
    (Environment) –

    The environment in which the check runs. Only included in the result if the check function declares an environment parameter.

  • datasources
    (dict[str, Datasource]) –

    A mapping of datasource names to instantiated Datasource objects.

Returns:

  • dict[str, Environment | Datasource] –

    A dictionary of keyword arguments to pass to the check function.

run_async async 

run_async(
    *,
    watchpost: Watchpost,
    environment: Environment,
    datasources: dict[str, Datasource],
) -> list[ExecutionResult]

Execute the check function asynchronously and return ExecutionResults.

Captures stdout/stderr during execution, then normalizes and materializes results. Raises TypeError if the check function is synchronous.

Parameters:

  • watchpost
    (Watchpost) –

    The Watchpost application providing execution context.

  • environment
    (Environment) –

    The environment in which the check should run.

  • datasources
    (dict[str, Datasource]) –

    Instantiated Datasource objects passed as keyword arguments.

Returns:

  • list[ExecutionResult] –

    A list of ExecutionResult instances created from the check output.

Raises:

  • TypeError –

    If the check is synchronous. Use run_sync instead.

run_sync 

run_sync(
    *,
    watchpost: Watchpost,
    environment: Environment,
    datasources: dict[str, Datasource],
) -> list[ExecutionResult]

Execute the check function synchronously and return ExecutionResults.

Captures stdout/stderr during execution, then normalizes and materializes results. Raises TypeError if the check function is asynchronous (use run_async instead).

Parameters:

  • watchpost
    (Watchpost) –

    The Watchpost application providing execution context.

  • environment
    (Environment) –

    The environment in which the check should run.

  • datasources
    (dict[str, Datasource]) –

    Instantiated Datasource objects passed as keyword arguments.

Returns:

  • list[ExecutionResult] –

    A list of ExecutionResult instances created from the check output.

Raises:

  • TypeError –

    If the check is asynchronous. Use run_async instead.

CheckCache 

CheckCache(storage: Storage)

Caches executed check results per check and environment.

This class is a thin wrapper around Cache that uses a composite key based on the check definition and environment. Values are lists of ExecutionResult objects, stored with a TTL derived from the Check.

Parameters:

  • storage

    (Storage) –

    The storage backend used to persist cache entries.

Methods:

get_check_results_cache_entry 

get_check_results_cache_entry(
    check: Check,
    environment: Environment,
    return_expired: bool = False,
) -> CacheEntry[list[ExecutionResult]] | None

Retrieve cached results for a given check and environment.

Parameters:

  • check
    (Check) –

    The check whose results should be retrieved.

  • environment
    (Environment) –

    The environment associated with the cached results.

  • return_expired
    (bool, default: False ) –

    Whether to return an entry even if it has expired. An expired entry is returned at most once.

Returns:

store_check_results 

store_check_results(
    check: Check,
    environment: Environment,
    results: list[ExecutionResult],
    override_cache_for: timedelta | None = None,
) -> None

Store the results of a check execution.

Parameters:

  • check
    (Check) –

    The check that produced the results.

  • environment
    (Environment) –

    The environment in which the check ran.

  • results
    (list[ExecutionResult]) –

    The list of ExecutionResult instances to cache.

  • override_cache_for
    (timedelta | None, default: None ) –

    Override the time to cache the result for.

Notes

This is a no-op if check.cache_for is None (caching disabled).

ErrorHandler

Bases: Protocol

A strategy for transforming error results to match a check's service pattern.

When a check function fails to execute successfully (e.g., raises an exception, datasource is unavailable, or scheduling prevents execution), Watchpost needs to create synthetic monitoring results. An ErrorHandler transforms the initial error result(s) to match the services that the check would normally create.

For example, if a check normally produces results for multiple hostnames or with different name suffixes, an ErrorHandler can expand a single error result into multiple results with the appropriate hostname and service name variations.

Multiple ErrorHandlers can be attached to a single check. They are applied sequentially, with each handler receiving the output of the previous one. This allows composition of different expansion strategies (e.g., first expand by hostname, then by name suffix).

See expand_by_hostname() and expand_by_name_suffix() for built-in implementations that handle common expansion patterns.

check 

check(
    *,
    name: str,
    service_labels: dict[str, Any],
    environments: list[Environment],
    cache_for: timedelta | str | None,
    hostname: HostnameInput | None = None,
    scheduling_strategies: list[SchedulingStrategy]
    | None = None,
    error_handlers: list[ErrorHandler] | None = None,
) -> Callable[[CheckFunction], Check]

Decorator to define a Watchpost check function.

Use this decorator to declare a check and attach metadata such as the service name, labels, environments, caching, and an optional hostname strategy. The decorated function may return a single result or result-builder, a list of results or result-builders, or yield results or result-builders as a generator.

This decorator supports both sync and async functions.

Parameters:

  • name

    (str) –

    The Checkmk service name for this check.

  • service_labels

    (dict[str, Any]) –

    Labels to attach to the service. This maps to Checkmk service labels exactly.

  • environments

    (list[Environment]) –

    The environments in which this check will be executed.

  • cache_for

    (timedelta | str | None) –

    Optional cache duration. Accepts a timedelta or a string supported by normalize_to_timedelta. If provided, results are cached for this duration.

  • hostname

    (HostnameInput | None, default: None ) –

    An optional hostname input or strategy that controls piggyback host resolution for results of this check.

  • scheduling_strategies

    (list[SchedulingStrategy] | None, default: None ) –

    Optional list of scheduling strategies to apply to this check.

  • error_handlers

    (list[ErrorHandler] | None, default: None ) –

    Optional list of error handlers that are called to adapt the results when the check function raises an exception.

Returns:

  • Callable[[CheckFunction], Check] –

    A Check instance wrapping the decorated function and provided

  • Callable[[CheckFunction], Check] –

    metadata.

Notes
  • A CheckResult may set a name_suffix to create multiple services from a single check function.
  • See watchpost.hostname for the available hostname strategy inputs.

cli

Modules:

loader

Classes:

  • AppNotFound –

    Custom exception for when the app cannot be found.

Functions:

  • find_app –

    Finds and loads the Watchpost app instance.

AppNotFound

Bases: Exception

Custom exception for when the app cannot be found.

find_app 

find_app(app_str: str | None) -> Watchpost

Finds and loads the Watchpost app instance.

The search order is: 1. The app_str argument if provided (e.g., 'my_module:app'). 2. Convention: look for a Watchpost instance named app in watchpost.py, then app.py, then main.py in the current directory.

datasource

Datasource primitives and factory helpers.

This module defines the Datasource base class, the DatasourceFactory protocol for constructing datasources, a FromFactory marker to declare factory-created dependencies in check parameters, and a DatasourceUnavailable exception to signal that a datasource is not available or otherwise unusable.

Notes

Datasources and factories can declare scheduling_strategies that Watchpost uses to decide where a check may run.

Classes:

  • Datasource –

    Base class for datasources that provide input to checks.

  • DatasourceFactory –

    Protocol for factories that construct datasources for checks.

  • DatasourceUnavailable –

    Raised when a datasource cannot be created, accessed, or used.

  • FromFactory –

    Marker used with typing.Annotated to request a datasource from a factory.

Datasource

Bases: ABC

Base class for datasources that provide input to checks.

Subclass this to integrate systems your checks read from (APIs, files, services, etc.). Datasources can influence where and when checks are run by exposing scheduling strategies.

Attributes:

scheduling_strategies class-attribute instance-attribute 

scheduling_strategies: (
    tuple[SchedulingStrategy, ...] | EllipsisType | None
) = ...

Optional scheduling strategies that constrain where a check may run.

  • Ellipsis (...) means "unspecified". If a datasource created by a factory leaves this as Ellipsis, Watchpost will fall back to the factory's own scheduling_strategies if any.
  • A tuple provides explicit strategies for this datasource and takes precedence over any factory strategies.
  • None or an empty tuple means "no constraints".

DatasourceFactory

Bases: Protocol

Protocol for factories that construct datasources for checks.

A factory centralizes configuration and creation logic for a datasource. Watchpost can use factory-level scheduling_strategies when the created datasource does not define its own.

Attributes:

new class-attribute 

new: Callable[..., Datasource]

Function that creates and returns a datasource instance.

Watchpost forwards the positional and keyword arguments declared via FromFactory(..., *args, **kwargs) to this callable.

scheduling_strategies class-attribute instance-attribute 

scheduling_strategies: (
    tuple[SchedulingStrategy, ...] | EllipsisType | None
) = ...

Optional scheduling strategies applied to datasources created by this factory.

A tuple sets explicit constraints; None or an empty tuple means "no constraints".

DatasourceUnavailable

Bases: Exception

Raised when a datasource cannot be created, accessed, or used.

Watchpost raises this error when a check requests data, but the underlying system is not reachable, misconfigured, or otherwise unavailable.

FromFactory 

FromFactory(
    factory: type[DatasourceFactory],
    *args: Any,
    **kwargs: Any,
)

FromFactory(*args: Any, **kwargs: Any)

FromFactory(
    factory: type[DatasourceFactory] | Any = None,
    *args: Any,
    **kwargs: Any,
)

Marker used with typing.Annotated to request a datasource from a factory.

Use in a check parameter annotation as:

Annotated[MyDatasource, FromFactory(MyFactory, "service")]

You may omit the factory type when the parameter type implements the factory protocol:

Annotated[MyDatasourceWithFactory, FromFactory("service")]

Parameters are forwarded to the factory's new callable. Watchpost caches the constructed datasource per factory type and argument set.

Parameters:

  • factory

    (type[DatasourceFactory] | Any, default: None ) –

    Optional factory type used to create the datasource. If omitted, Watchpost infers the factory from the annotated parameter's type. If the first positional argument is not a type, it is treated as a factory argument and the factory is inferred.

  • args

    (Any, default: () ) –

    Positional arguments forwarded to the factory new callable.

  • kwargs

    (Any, default: {} ) –

    Keyword arguments forwarded to the factory new callable.

Methods:

  • cache_key –

    Generate a stable cache key for this factory invocation.

cache_key 

cache_key(
    type_key: type[DatasourceFactory] | None,
) -> tuple[type[DatasourceFactory] | None, int, int]

Generate a stable cache key for this factory invocation.

Parameters:

  • type_key
    (type[DatasourceFactory] | None) –

    Fallback factory type to use when this instance did not specify a factory explicitly (i.e., it will be inferred from the annotated parameter type).

Returns:

  • type[DatasourceFactory] | None –

    A tuple that identifies the factory type and the provided arguments,

  • int –

    suitable for use as a cache key.

discover_checks

Utilities for discovering checks defined in modules and packages.

This module scans Python modules for global Check instances and, when requested, traverses packages to find checks across submodules. It offers predicates to include or exclude modules, an optional per-check filter, and configurable error handling for import failures.

Classes:

  • DiscoveryError –

    Raised when a submodule import fails during discovery and strict error

Functions:

  • discover_checks –

    Discover Check instances defined as global names in a module or package.

DiscoveryError

Bases: Exception

Raised when a submodule import fails during discovery and strict error handling is enabled.

This exception is raised only when raise_on_import_error=True is passed to discover_checks() and a submodule cannot be imported while traversing a package.

discover_checks 

discover_checks(
    module: ModuleType | str,
    *,
    recursive: bool = True,
    include_module: ModulePredicate | None = None,
    exclude_module: ModulePredicate | None = None,
    check_filter: CheckPredicate | None = None,
    raise_on_import_error: bool = False,
) -> list[Check]

Discover Check instances defined as global names in a module or package.

This function scans the given module for global Check objects and, when recursive is enabled, traverses subpackages to discover checks across the package tree. It supports module-level include/exclude predicates, a per-check filter, and optional strict error handling for imports.

Parameters:

  • module

    (ModuleType | str) –

    The root module object or dotted module name to scan for checks.

  • recursive

    (bool, default: True ) –

    Indicates whether to traverse subpackages when the module is a package.

  • include_module

    (ModulePredicate | None, default: None ) –

    Predicate receiving the imported module. Return True to include the module in scanning; return False to skip it. Use this to constrain traversal (for example, skip test modules or expensive imports).

  • exclude_module

    (ModulePredicate | None, default: None ) –

    Predicate receiving the imported module. If it returns True, the module is skipped. This is applied after include_module.

  • check_filter

    (CheckPredicate | None, default: None ) –

    Predicate receiving (check, module, name) that decides whether a discovered Check should be included. Return True to keep it.

  • raise_on_import_error

    (bool, default: False ) –

    If True, raise a DiscoveryError when a submodule import fails while traversing packages. If False (default), such modules are skipped.

Returns:

  • list[Check] –

    A list of discovered Check instances. If the same Check object is

  • list[Check] –

    re-exported from multiple modules, it appears only once.

Raises:

  • DiscoveryError –

    If a submodule import fails and raise_on_import_error is True.

Notes

Checks are discovered by scanning module globals and selecting objects that are instances of Check. Identity de-duplicates duplicate objects to avoid listing the same Check multiple times when re-exported.

environment

Execution environments and registry.

This module defines Environment, representing where a check runs or targets, and EnvironmentRegistry, a simple container to manage named environments.

Notes

Environments can carry hostname configuration used during result output.

Classes:

Environment 

Environment(
    name: str,
    *,
    hostname: HostnameInput | None = None,
    **metadata: Hashable,
)

Represents a logical environment in which checks run.

An environment typically maps to how or where checks are executed (for example: dev, stage, prod, monitoring). It can carry a hostname resolution strategy and arbitrary metadata that checks and data sources may use to alter behavior.

Parameters:

  • name

    (str) –

    The name of the environment (e.g., "dev", "stage", "prod", "monitoring").

  • hostname

    (HostnameInput | None, default: None ) –

    Input used to determine the hostname when running checks in this environment. Accepts the same forms as HostnameInput such as a template string, a callable, or a strategy instance. If omitted, the application may fall back to its default hostname generation.

  • metadata

    (Hashable, default: {} ) –

    Arbitrary, hashable metadata attached to the environment. Checks and data sources may consult these values to change behavior.

Notes

Hostname handling integrates with the hostname resolution system in watchpost.hostname. The chosen strategy ultimately influences the Checkmk host a service is associated with.

EnvironmentRegistry 

EnvironmentRegistry()

Simple in-memory registry of Environment objects.

The registry provides dictionary-like access and helpers to create and manage environments used by a Watchpost application.

Methods:

  • add –

    Add or replace an environment by its name.

  • get –

    Return the environment for the given name, or a default if missing.

  • new –

    Create a new environment, add it to the registry, and return it.

add 

add(environment: Environment) -> None

Add or replace an environment by its name.

Parameters:

  • environment
    (Environment) –

    The environment to register. If another environment with the same name exists, it will be overwritten.

get 

get(
    name: str, default: Environment | None = None
) -> Environment | None

Return the environment for the given name, or a default if missing.

Parameters:

  • name
    (str) –

    The environment name to look up.

  • default
    (Environment | None, default: None ) –

    Value to return when the name is not present.

Returns:

  • Environment | None –

    The matching Environment, or the provided default.

new 

new(
    name: str,
    *,
    hostname: HostnameInput | None = None,
    **metadata: Hashable,
) -> Environment

Create a new environment, add it to the registry, and return it.

Parameters:

  • name
    (str) –

    The name of the environment to create.

  • hostname
    (HostnameInput | None, default: None ) –

    Input used to determine the hostname for this environment. See Environment.__init__ for supported forms.

  • metadata
    (Hashable, default: {} ) –

    Arbitrary, hashable metadata to attach to the environment.

Returns:

  • Environment –

    The created and registered Environment instance.

executor

Threaded check executor for Watchpost.

Provides a non-blocking, key-aware execution engine that de-duplicates work per key and can run both synchronous and asynchronous check functions. It exposes lightweight statistics used by the HTTP endpoints and tests.

Classes:

AsyncioLoopThread 

AsyncioLoopThread(*args: Any, **kwargs: Any)

Bases: Thread

Run an asyncio event loop in a dedicated background thread.

The thread creates its own event loop and runs it forever until stopped. CheckExecutor uses this to execute coroutine functions without blocking the worker threads in the thread pool.

BlockingCheckExecutor 

BlockingCheckExecutor(max_workers: int | None = 1)

Bases: CheckExecutor[T]

Execute checks while blocking until results are available.

This variant waits for all futures of a key to complete when result() is called. It is useful for tests and simple CLI usage where non-blocking behavior is not required.

Notes

Prefer the default CheckExecutor for production use. This class is provided for special cases (such as the CLI or in unit-tests) where synchronous behavior is desired.

Classes:

  • Statistics –

    Summary statistics of the executor state.

Methods:

  • errored –

    Return a mapping of keys to stringified exceptions for errored jobs.

  • result –

    Wait for all active futures of the key and then return the next result.

  • shutdown –

    Shut down the executor and stop the background event loop.

  • statistics –

    Compute executor statistics.

  • submit –

    Submit a function to run for a key, deduplicating concurrent work.

Attributes:

  • asyncio_loop (AbstractEventLoop) –

    Return the background asyncio event loop, starting it on first access.

asyncio_loop property 

asyncio_loop: AbstractEventLoop

Return the background asyncio event loop, starting it on first access.

Returns:

  • AbstractEventLoop –

    The event loop used to run coroutine functions submitted to this

  • AbstractEventLoop –

    executor.

Statistics dataclass 

Statistics(
    total: int,
    completed: int,
    errored: int,
    running: int,
    awaiting_pickup: int,
)

Summary statistics of the executor state.

These values feed monitoring endpoints and tests to provide visibility into how many jobs are running, finished, or awaiting pickup.

Attributes:

  • awaiting_pickup (int) –

    Total number of finished futures (completed + errored) that have not

  • completed (int) –

    Number of successfully completed futures awaiting pickup.

  • errored (int) –

    Number of futures that completed with an exception and await pickup.

  • running (int) –

    Number of futures currently executing (not yet completed).

  • total (int) –

    Number of active futures across all keys (running + awaiting pickup).

awaiting_pickup instance-attribute 
awaiting_pickup: int

Total number of finished futures (completed + errored) that have not yet been retrieved via result().

completed instance-attribute 
completed: int

Number of successfully completed futures awaiting pickup.

errored instance-attribute 
errored: int

Number of futures that completed with an exception and await pickup.

running instance-attribute 
running: int

Number of futures currently executing (not yet completed).

total instance-attribute 
total: int

Number of active futures across all keys (running + awaiting pickup).

errored 

errored() -> dict[str, str]

Return a mapping of keys to stringified exceptions for errored jobs.

Returns:

  • dict[str, str] –

    A dict mapping the string form of each key to the corresponding

  • dict[str, str] –

    exception message of futures that completed with an error and have

  • dict[str, str] –

    not yet been picked up via result().

result 

result(key: Hashable) -> T | None

Wait for all active futures of the key and then return the next result.

Parameters:

  • key
    (Hashable) –

    The key used when submitting the job(s).

Returns:

  • T | None –

    The completed result value, or None if no finished results are

  • T | None –

    queued after waiting.

shutdown 

shutdown(wait: bool = False) -> None

Shut down the executor and stop the background event loop.

Parameters:

  • wait
    (bool, default: False ) –

    If true, waits for all running futures to finish before returning. This is passed through to the underlying ThreadPoolExecutor.shutdown().

statistics 

statistics() -> Statistics

Compute executor statistics.

Returns:

  • Statistics –

    A CheckExecutor.Statistics instance summarizing total, completed,

  • Statistics –

    errored, running, and awaiting pickup futures.

submit 

submit(
    key: Hashable,
    func: Callable[P, T | Awaitable[T]],
    *args: P.args,
    resubmit: bool = False,
    **kwargs: P.kwargs,
) -> Future

Submit a function to run for a key, deduplicating concurrent work.

If another job with the same key is already running and resubmit is false, this returns the existing future instead of starting a new one. Coroutine functions are scheduled on the background asyncio loop.

Parameters:

  • key
    (Hashable) –

    The deduplication key. Only one active job per key is started unless resubmit=True is given.

  • func
    (Callable[P, T | Awaitable[T]]) –

    The callable to execute. May be synchronous or a coroutine function.

  • *args
    (P.args, default: () ) –

    Positional arguments passed to the callable.

  • resubmit
    (bool, default: False ) –

    When true, always schedules a new job even if one with the same key is already running.

  • **kwargs
    (P.kwargs, default: {} ) –

    Keyword arguments passed to the callable.

Returns:

  • Future –

    A Future representing the running or already existing job.

CheckExecutor 

CheckExecutor(max_workers: int | None = None)

Execute checks concurrently while avoiding duplicate work per key.

This executor wraps a ThreadPoolExecutor and adds key-aware submission: if a job for a key is already running, later submissions with the same key return the existing future unless resubmit=True is passed. The executor can also run coroutine functions by scheduling them on a single background asyncio event loop.

Classes:

  • Statistics –

    Summary statistics of the executor state.

Methods:

  • errored –

    Return a mapping of keys to stringified exceptions for errored jobs.

  • result –

    Retrieve the next finished result for a key.

  • shutdown –

    Shut down the executor and stop the background event loop.

  • statistics –

    Compute executor statistics.

  • submit –

    Submit a function to run for a key, deduplicating concurrent work.

Attributes:

  • asyncio_loop (AbstractEventLoop) –

    Return the background asyncio event loop, starting it on first access.

asyncio_loop property 

asyncio_loop: AbstractEventLoop

Return the background asyncio event loop, starting it on first access.

Returns:

  • AbstractEventLoop –

    The event loop used to run coroutine functions submitted to this

  • AbstractEventLoop –

    executor.

Statistics dataclass 

Statistics(
    total: int,
    completed: int,
    errored: int,
    running: int,
    awaiting_pickup: int,
)

Summary statistics of the executor state.

These values feed monitoring endpoints and tests to provide visibility into how many jobs are running, finished, or awaiting pickup.

Attributes:

  • awaiting_pickup (int) –

    Total number of finished futures (completed + errored) that have not

  • completed (int) –

    Number of successfully completed futures awaiting pickup.

  • errored (int) –

    Number of futures that completed with an exception and await pickup.

  • running (int) –

    Number of futures currently executing (not yet completed).

  • total (int) –

    Number of active futures across all keys (running + awaiting pickup).

awaiting_pickup instance-attribute 
awaiting_pickup: int

Total number of finished futures (completed + errored) that have not yet been retrieved via result().

completed instance-attribute 
completed: int

Number of successfully completed futures awaiting pickup.

errored instance-attribute 
errored: int

Number of futures that completed with an exception and await pickup.

running instance-attribute 
running: int

Number of futures currently executing (not yet completed).

total instance-attribute 
total: int

Number of active futures across all keys (running + awaiting pickup).

errored 

errored() -> dict[str, str]

Return a mapping of keys to stringified exceptions for errored jobs.

Returns:

  • dict[str, str] –

    A dict mapping the string form of each key to the corresponding

  • dict[str, str] –

    exception message of futures that completed with an error and have

  • dict[str, str] –

    not yet been picked up via result().

result 

result(key: Hashable) -> T | None

Retrieve the next finished result for a key.

If no job for the key exists, raises a KeyError. If jobs exist but none has completed yet, returns None. When the last active future for a key is consumed, the internal state for the key is cleaned up.

Parameters:

  • key
    (Hashable) –

    The key used when submitting the job(s).

Returns:

  • T | None –

    The completed result value or None if the job is still running.

Raises:

  • KeyError –

    If no job for the given key has been submitted.

shutdown 

shutdown(wait: bool = False) -> None

Shut down the executor and stop the background event loop.

Parameters:

  • wait
    (bool, default: False ) –

    If true, waits for all running futures to finish before returning. This is passed through to the underlying ThreadPoolExecutor.shutdown().

statistics 

statistics() -> Statistics

Compute executor statistics.

Returns:

  • Statistics –

    A CheckExecutor.Statistics instance summarizing total, completed,

  • Statistics –

    errored, running, and awaiting pickup futures.

submit 

submit(
    key: Hashable,
    func: Callable[P, T | Awaitable[T]],
    *args: P.args,
    resubmit: bool = False,
    **kwargs: P.kwargs,
) -> Future

Submit a function to run for a key, deduplicating concurrent work.

If another job with the same key is already running and resubmit is false, this returns the existing future instead of starting a new one. Coroutine functions are scheduled on the background asyncio loop.

Parameters:

  • key
    (Hashable) –

    The deduplication key. Only one active job per key is started unless resubmit=True is given.

  • func
    (Callable[P, T | Awaitable[T]]) –

    The callable to execute. May be synchronous or a coroutine function.

  • *args
    (P.args, default: () ) –

    Positional arguments passed to the callable.

  • resubmit
    (bool, default: False ) –

    When true, always schedules a new job even if one with the same key is already running.

  • **kwargs
    (P.kwargs, default: {} ) –

    Keyword arguments passed to the callable.

Returns:

  • Future –

    A Future representing the running or already existing job.

globals

Global state for Watchpost and a proxy to the active application instance.

This module exposes current_app, a lightweight LocalProxy that resolves to the running Watchpost instance. It uses a ContextVar, so access is bound to the current async/task context rather than a process-wide global. This makes it safe to use from request handlers, check execution, and any code that runs under Watchpost.app_context().

Notes:

  • Watchpost.app_context() sets and resets the context variable around operations that need access to the active application. The ASGI integration also enters this context automatically for incoming requests.
  • Accessing current_app outside an active context raises a RuntimeError with a helpful message. Ensure your code runs inside Watchpost.app_context() or via the ASGI app when using current_app.

hostname

Hostname resolution utilities and strategies.

Provides a small strategy protocol to compute the target hostname for a check result, helpers to validate or coerce hostnames to RFC1123, and the orchestration function resolve_hostname that applies precedence across result, check, environment, and the Watchpost application.

Notes

Hostnames must follow RFC1123: ASCII letters, digits, and hyphens in dot-separated labels, each label 1-63 characters, total length up to 253 characters. Labels must start and end with an alphanumeric character.

Classes:

Functions:

  • coerce_to_rfc1123 –

    Normalize and coerce an arbitrary string into an RFC1123-compatible

  • is_rfc1123_hostname –

    Determine whether the given value is a valid RFC1123 hostname.

  • resolve_hostname –

    Resolve the final hostname for a check execution.

  • to_strategy –

    Convert user-facing hostname input into a HostnameStrategy.

CoercingStrategy 

CoercingStrategy(inner: HostnameStrategy)

Bases: HostnameStrategy

Wraps another strategy and coerces its result to RFC1123 if present.

If the inner strategy returns None, no coercion happens and None is returned. If a non-empty hostname is returned, it is passed through coerce_to_rfc1123.

Parameters:

  • inner

    (HostnameStrategy) –

    The inner strategy whose output should be coerced.

CompositeStrategy 

CompositeStrategy(*strategies: HostnameStrategy)

Bases: HostnameStrategy

Tries multiple strategies in order and returns the first non-empty result.

Parameters:

  • strategies

    (HostnameStrategy, default: () ) –

    One or more strategies to evaluate in order.

FunctionStrategy 

FunctionStrategy(
    fn: Callable[
        [HostnameContext], str | NoPiggybackHost | None
    ],
)

Bases: HostnameStrategy

Calls a function with the HostnameContext to compute a hostname.

Returning None indicates no decision and allows fallthrough to later strategies.

Parameters:

  • fn

    (Callable[[HostnameContext], str | NoPiggybackHost | None]) –

    A callable that receives a HostnameContext and returns a hostname or None.

HostnameContext dataclass 

HostnameContext(
    check: Check,
    environment: Environment,
    service_name: str,
    service_labels: dict[str, Any],
    result: CheckResult | None = None,
)

Carries context available when resolving a hostname for a check execution.

This object is passed to hostname strategies so they can decide on a hostname based on the check, environment, service information, and an optional result.

Methods:

  • new –

    Create a HostnameContext with sensible defaults from the given inputs.

Attributes:

  • check (Check) –

    The current check being executed.

  • environment (Environment) –

    The current environment of the execution.

  • result (CheckResult | None) –

    The optional result that the check returned. If present, it may include a

  • service_labels (dict[str, Any]) –

    Labels attached to the service. Useful for templating or custom strategies.

  • service_name (str) –

    The service name as defined by the check (@check). Checkmk will use this

check instance-attribute 

check: Check

The current check being executed.

environment instance-attribute 

environment: Environment

The current environment of the execution.

result class-attribute instance-attribute 

result: CheckResult | None = None

The optional result that the check returned. If present, it may include a hostname override on the result which takes precedence during resolution.

service_labels instance-attribute 

service_labels: dict[str, Any]

Labels attached to the service. Useful for templating or custom strategies.

service_name instance-attribute 

service_name: str

The service name as defined by the check (@check). Checkmk will use this name to identify the service.

new classmethod 

new(
    *,
    check: Check,
    environment: Environment,
    result: CheckResult | None = None,
    service_name: str | None = None,
    service_labels: dict[str, Any] | None = None,
) -> HostnameContext

Create a HostnameContext with sensible defaults from the given inputs.

Parameters:

  • check
    (Check) –

    The check for which a hostname is being resolved.

  • environment
    (Environment) –

    The environment in which the check runs.

  • result
    (CheckResult | None, default: None ) –

    Optional result produced by the check. When given, used to support per-result hostname overrides.

  • service_name
    (str | None, default: None ) –

    Optional service name. Defaults to the check's service name.

  • service_labels
    (dict[str, Any] | None, default: None ) –

    Optional service labels. Defaults to the check's service labels.

Returns:

HostnameResolutionError

Bases: Exception

Raised when a hostname cannot be resolved or validated.

This error wraps exceptions thrown by strategies and signals failure to produce a usable hostname for a check/environment combination.

HostnameStrategy

Bases: Protocol

Strategy protocol for resolving a hostname from a HostnameContext.

Implementations return a hostname string or None if they cannot decide. Returning None allows composition where later strategies may provide a hostname.

NoPiggybackHostStrategy

Bases: HostnameStrategy

Explicitly disables the setting of a piggyback hostname. Whatever results a check has created will be associated with the host the Watchpost application ran on.

Compared to simply providing an empty string, using this strategy will never result in either a fallback hostname being generated or a HostnameResolutionError to be raised.

StaticHostnameStrategy 

StaticHostnameStrategy(hostname: str)

Bases: HostnameStrategy

Always returns the given hostname literal.

Parameters:

  • hostname

    (str) –

    The hostname to return for any context.

TemplateStrategy 

TemplateStrategy(template: str)

Bases: HostnameStrategy

Formats a string template using fields from the HostnameContext.

The template is processed with str.format(**asdict(ctx)). You can access context fields, including nested attributes on objects such as {environment.name} or {check.service_name}.

Parameters:

  • template

    (str) –

    A format string for str.format using keys from the context.

coerce_to_rfc1123 

coerce_to_rfc1123(value: str | NoPiggybackHost) -> str

Normalize and coerce an arbitrary string into an RFC1123-compatible hostname.

The process lowercases the input, removes unsupported characters, replaces invalid characters with hyphens, normalizes repeated separators, trims leading/trailing hyphens from labels, and truncates labels and the overall hostname to RFC1123 limits. If the value cannot be coerced into a valid hostname, a ValueError is raised.

Parameters:

  • value

    (str | NoPiggybackHost) –

    The input string to coerce into a hostname.

Returns:

  • str –

    A string that complies with RFC1123 hostname rules.

Raises:

  • ValueError –

    If the input is empty or cannot be transformed into a valid hostname.

is_rfc1123_hostname 

is_rfc1123_hostname(value: str) -> bool

Determine whether the given value is a valid RFC1123 hostname.

Parameters:

  • value

    (str) –

    The hostname candidate to validate.

Returns:

  • bool –

    True if the value conforms to RFC1123; otherwise, False.

Notes

Enforces ASCII letters, digits, and hyphens in dot-separated labels; each label is 1-63 characters and must start and end with an alphanumeric character. The full hostname must not exceed 253 characters.

resolve_hostname 

resolve_hostname(
    *,
    watchpost: Watchpost,
    check: Check,
    environment: Environment,
    result: CheckResult | None,
    fallback_to_default_hostname_generation: bool = True,
    coerce_into_valid_hostname: bool = True,
    explicit_hostname: HostnameInput | None = None,
) -> str

Resolve the final hostname for a check execution.

Precedence:

  1. Explicit hostname (if present).
  2. Per-result override on the CheckResult (if present).
  3. Check-level strategy on the Check.
  4. Environment-level strategy on the Environment.
  5. Watchpost-level strategy on the application.
  6. Optional fallback to "{service_name}-{environment.name}" if enabled.

The resolved hostname must conform to RFC1123. If it does not and coerce_into_valid_hostname is True, the hostname is coerced using coerce_to_rfc1123. Otherwise, a HostnameResolutionError is raised.

Parameters:

  • watchpost

    (Watchpost) –

    The Watchpost application providing a default strategy.

  • check

    (Check) –

    The check for which a hostname is being resolved.

  • environment

    (Environment) –

    The environment in which the check runs.

  • result

    (CheckResult | None) –

    Optional result that may contain a per-result hostname override.

  • fallback_to_default_hostname_generation

    (bool, default: True ) –

    Whether to fall back to "{service_name}-{environment.name}" when no strategy produced a hostname.

  • coerce_into_valid_hostname

    (bool, default: True ) –

    Whether to coerce a non-compliant hostname into RFC1123 format.

  • explicit_hostname

    (HostnameInput | None, default: None ) –

    An explicit hostname override, if any.

Returns:

  • str –

    The resolved hostname.

Raises:

  • HostnameResolutionError –

    If no hostname can be resolved, a strategy fails, or a non-compliant hostname cannot be coerced.

to_strategy 

to_strategy(
    value: HostnameInput | None,
) -> HostnameStrategy | None

Convert user-facing hostname input into a HostnameStrategy.

Accepts one of the following inputs:

  • A string interpreted as a template for TemplateStrategy.
  • A callable taking HostnameContext and returning a hostname or None.
  • An object implementing the HostnameStrategy protocol.

Parameters:

  • value

    (HostnameInput | None) –

    The value to convert into a strategy, or None to indicate no strategy.

Returns:

  • HostnameStrategy | None –

    A HostnameStrategy instance or None if the input was None.

Raises:

  • TypeError –

    If the input type is not supported.

http

Starlette HTTP routes for Watchpost.

This module exposes operational endpoints and the streaming root endpoint:

  • /healthcheck returns 204 for liveness checks.
  • /executor/statistics returns executor statistics as JSON.
  • /executor/errored returns a list of errored checks as JSON.
  • / streams Checkmk-compatible output from running checks.

result

Result types and builders for Watchpost checks.

Classes:

  • CheckResult –

    Represents the result of a check performed on a system or component.

  • CheckState –

    Represents the state of a check.

  • ExecutionResult –

    This is an internal type that represents the final execution result of a

  • Metric –

    Represents a measurable metric with an optional threshold configuration.

  • OngoingCheckResult –

    An "ongoing check result" represents a builder that allows you to build up a

  • Thresholds –

    Represents threshold values with warning and critical levels.

Functions:

  • build_result –

    Start building up a new check result that allows adding multiple

  • crit –

    Generates a CheckResult object indicating a CRIT check state.

  • normalize_details –

    Normalize and format the given details into a string.

  • ok –

    Generates a CheckResult object indicating an OK check state.

  • unknown –

    Generates a CheckResult object indicating an UNKNOWN check state.

  • warn –

    Generates a CheckResult object indicating a WARN check state.

CheckResult dataclass 

CheckResult(
    check_state: CheckState,
    summary: str,
    details: Details | None = None,
    name_suffix: str | None = None,
    metrics: list[Metric] | None = None,
    hostname: HostnameInput | None = None,
)

Represents the result of a check performed on a system or component.

This class encapsulates the result of a check operation, providing details such as the state of the check, a summary, optional details, a name suffix, metrics, and hostname information. It is designed to store and convey the outcome of a check operation in a structured format.

Attributes:

  • check_state (CheckState) –

    The state of the check, indicating whether it was successful, warning,

  • details (str | None) –

    A detailed output of the check result, providing additional information

  • hostname (HostnameInput | None) –

    An optional hostname that overrides the hostname that would have been used

  • metrics (list[Metric] | None) –

    An optional list of metrics to associate with the check.

  • name_suffix (str | None) –

    A suffix to add to the check name as defined in the @check decorator.

  • summary (str) –

    A summary of the check result, indicating the outcome of the check.

check_state instance-attribute 

check_state: CheckState = check_state

The state of the check, indicating whether it was successful, warning, critical, or unknown.

details class-attribute instance-attribute 

details: str | None = normalize_details(details)

A detailed output of the check result, providing additional information beyond the summary.

Checkmk will show this detailed output when you are viewing a specific service.

hostname class-attribute instance-attribute 

hostname: HostnameInput | None = hostname

An optional hostname that overrides the hostname that would have been used otherwise.

metrics class-attribute instance-attribute 

metrics: list[Metric] | None = metrics

An optional list of metrics to associate with the check.

name_suffix class-attribute instance-attribute 

name_suffix: str | None = name_suffix

A suffix to add to the check name as defined in the @check decorator.

This enables a single check function to return multiple results that create multiple services on the Checkmk side.

summary instance-attribute 

summary: str = summary

A summary of the check result, indicating the outcome of the check.

Checkmk will show this summary on pages like a host overview.

CheckState

Bases: Enum

Represents the state of a check.

This maps to the Checkmk concept of states exactly.

Attributes:

  • check_function (Callable[..., CheckResult]) –

    Determines the appropriate check function based on the current check

  • state_marker (str) –

    Returns a Checkmk output compatible state marker (such as (!!) for

check_function property 

check_function: Callable[..., CheckResult]

Determines the appropriate check function based on the current check state.

This property evaluates the current CheckState of the instance and matches it to a corresponding check function (ok, warn, crit, or unknown).

Returns:

  • Callable[..., CheckResult] –

    The function corresponding to the current state of the check.

state_marker property 

state_marker: str

Returns a Checkmk output compatible state marker (such as (!!) for CRIT) which Checkmk will automatically render as a visual status label.

ExecutionResult dataclass 

ExecutionResult(
    piggyback_host: str,
    service_name: str,
    service_labels: dict[str, str],
    environment_name: str,
    check_state: CheckState,
    summary: str,
    details: str | None = None,
    metrics: list[Metric] | None = None,
    check_definition: InvocationInformation | None = None,
)

This is an internal type that represents the final execution result of a check, containing everything required to turn it into Checkmk-compatible output.

Metric dataclass 

Metric(
    name: str,
    value: int | float,
    levels: Thresholds | None = None,
    boundaries: Thresholds | None = None,
)

Represents a measurable metric with an optional threshold configuration.

This maps to the Checkmk concept of metrics exactly.

OngoingCheckResult 

OngoingCheckResult(
    ok_summary: str,
    fail_summary: str,
    base_details: Details | None = None,
    name_suffix: str | None = None,
    metrics: list[Metric] | None = None,
    alternative_hostname: HostnameInput | None = None,
)

An "ongoing check result" represents a builder that allows you to build up a new check result by adding multiple OK/WARN/UNKNOWN/CRIT results, called "partials", which will eventually result in a regular check result that holds the worst check state provided by the individual results.

Use of this builder greatly simplifies scenarios where your check function validates multiple aspects of a single system or component but only returns a single check result: by allowing you to provide the status of each aspect as you check it through simple function calls on the builder instead of having to manually combine the results into a single check result, you can reduce the complexity of your check function and improve its readability.

Constructing this builder is recommended through the top-level build_result available in this module.

Classes:

  • Partial –

    The internal type representing a partial result of a check.

Methods:

  • add_check_result –

    Add a check result to the builder as a partial result.

  • crit –

    Add a CRIT result to the builder as a partial result.

  • ok –

    Add an OK result to the builder as a partial result.

  • to_check_result –

    Finalize this builder by turning it into a check result.

  • unknown –

    Add an UNKNOWN result to the builder as a partial result.

  • warn –

    Add a WARN result to the builder as a partial result.

Attributes:

check_state property 

check_state: CheckState

The overall check state of the builder, calculated based on the worst check state of the individual results.

Partial dataclass 

Partial(
    check_state: CheckState,
    summary: str,
    details: Details | None,
)

The internal type representing a partial result of a check.

add_check_result 

add_check_result(check_result: CheckResult) -> None

Add a check result to the builder as a partial result.

For most use cases you probably want to use the higher-level ok, warn, crit, and unknown functions instead.

Parameters:

  • check_result
    (CheckResult) –

    The check result to add as a partial result.

crit 

crit(summary: str, details: Details | None = None) -> None

Add a CRIT result to the builder as a partial result.

Parameters:

  • summary
    (str) –

    A summary of the check result, indicating the outcome of the check. Checkmk will show this summary on pages like a host overview.

  • details
    (Details | None, default: None ) –

    A detailed output of the check result, providing additional information beyond the summary. Checkmk will show this detailed output when you are viewing a specific service.

ok 

ok(summary: str, details: Details | None = None) -> None

Add an OK result to the builder as a partial result.

Parameters:

  • summary
    (str) –

    A summary of the check result, indicating the outcome of the check. Checkmk will show this summary on pages like a host overview.

  • details
    (Details | None, default: None ) –

    A detailed output of the check result, providing additional information beyond the summary. Checkmk will show this detailed output when you are viewing a specific service.

to_check_result 

to_check_result() -> CheckResult

Finalize this builder by turning it into a check result.

The created result holds the cumulative data of all partial results, with the worst check state as determined by the check state of the individual results.

unknown 

unknown(
    summary: str, details: Details | None = None
) -> None

Add an UNKNOWN result to the builder as a partial result.

Parameters:

  • summary
    (str) –

    A summary of the check result, indicating the outcome of the check. Checkmk will show this summary on pages like a host overview.

  • details
    (Details | None, default: None ) –

    A detailed output of the check result, providing additional information beyond the summary. Checkmk will show this detailed output when you are viewing a specific service.

warn 

warn(summary: str, details: Details | None = None) -> None

Add a WARN result to the builder as a partial result.

Parameters:

  • summary
    (str) –

    A summary of the check result, indicating the outcome of the check. Checkmk will show this summary on pages like a host overview.

  • details
    (Details | None, default: None ) –

    A detailed output of the check result, providing additional information beyond the summary. Checkmk will show this detailed output when you are viewing a specific service.

Thresholds dataclass 

Thresholds(warning: int | float, critical: int | float)

Represents threshold values with warning and critical levels.

This maps to the Checkmk concept of thresholds exactly and thus directly relates to metrics, see the Metric class.

build_result 

build_result(
    ok_summary: str,
    fail_summary: str,
    base_details: Details | None = None,
    name_suffix: str | None = None,
    metrics: list[Metric] | None = None,
    alternative_hostname: HostnameInput | None = None,
) -> OngoingCheckResult

Start building up a new check result that allows adding multiple OK/WARN/UNKNOWN/CRIT results, called "partials", eventually resulting in a regular check result that holds the worst check state provided by the individual results.

Use of this builder greatly simplifies scenarios where your check function validates multiple aspects of a single system or component but only returns a single check result: by allowing you to provide the status of each aspect as you check it through simple function calls on the builder instead of having to manually combine the results into a single check result, you can reduce the complexity of your check function and improve its readability.

Parameters:

  • ok_summary

    (str) –

    Overall check summary that will be shown if the final result is OK.

  • fail_summary

    (str) –

    Overall check summary that will be shown if the final result is WARN, UNKNOWN, or CRIT.

  • base_details

    (Details | None, default: None ) –

    Optional base details for the check result. They will always be included and then enriched by the details provided by partial results.

  • name_suffix

    (str | None, default: None ) –

    Optional suffix for the check result name.

  • metrics

    (list[Metric] | None, default: None ) –

    Optional list of metrics associated with the check.

  • alternative_hostname

    (HostnameInput | None, default: None ) –

    Optional alternative hostname input for the check.

Returns:

crit 

crit(
    summary: str,
    details: Details | None = None,
    name_suffix: str | None = None,
    metrics: list[Metric] | None = None,
    alternative_hostname: HostnameInput | None = None,
) -> CheckResult

Generates a CheckResult object indicating a CRIT check state.

This function creates and returns a CheckResult indicating the system or component is in a CRIT state. It allows for passing additional information such as a summary, details, name suffix, metrics, or an alternative hostname.

Parameters:

  • summary

    (str) –

    A summary of the check result, indicating the outcome of the check. Checkmk will show this summary on pages like a host overview.

  • details

    (Details | None, default: None ) –

    A detailed output of the check result, providing additional information beyond the summary. Checkmk will show this detailed output when you are viewing a specific service.

  • name_suffix

    (str | None, default: None ) –

    A suffix to add to the check name as defined in the @check decorator. This enables a single check function to return multiple results that create multiple services on the Checkmk side.

  • metrics

    (list[Metric] | None, default: None ) –

    An optional list of metrics to associate with the check.

  • alternative_hostname

    (HostnameInput | None, default: None ) –

    An optional hostname that overrides the hostname that would have been used otherwise.

Returns:

  • CheckResult –

    A CheckResult object representing the CRIT check result.

normalize_details 

normalize_details(details: Details | None) -> str | None

Normalize and format the given details into a string.

This function processes the input details parameter and normalizes it into a string representation based on its type. If the input is None, it returns None. For Exception objects, it extracts and formats the exception details. For dictionaries, it formats each key-value pair as a newline-separated string. Empty strings are normalized to None. Other string inputs are returned unchanged.

Parameters:

  • details

    (Details | None) –

    The input details to normalize. It can be of type Details (e.g., exception, string, dictionary), or None.

Returns:

  • str | None –

    A formatted string representation of the input details, or None if the

  • str | None –

    input is None or an empty string.

ok 

ok(
    summary: str,
    details: Details | None = None,
    name_suffix: str | None = None,
    metrics: list[Metric] | None = None,
    alternative_hostname: HostnameInput | None = None,
) -> CheckResult

Generates a CheckResult object indicating an OK check state.

This function creates and returns a CheckResult indicating the system or component is in an OK state. It allows for passing additional information such as a summary, details, name suffix, metrics, or an alternative hostname.

Parameters:

  • summary

    (str) –

    A summary of the check result, indicating the outcome of the check. Checkmk will show this summary on pages like a host overview.

  • details

    (Details | None, default: None ) –

    A detailed output of the check result, providing additional information beyond the summary. Checkmk will show this detailed output when you are viewing a specific service.

  • name_suffix

    (str | None, default: None ) –

    A suffix to add to the check name as defined in the @check decorator. This enables a single check function to return multiple results that create multiple services on the Checkmk side.

  • metrics

    (list[Metric] | None, default: None ) –

    An optional list of metrics to associate with the check.

  • alternative_hostname

    (HostnameInput | None, default: None ) –

    An optional hostname that overrides the hostname that would have been used otherwise.

Returns:

  • CheckResult –

    A CheckResult object representing the OK check result.

unknown 

unknown(
    summary: str,
    details: Details | None = None,
    name_suffix: str | None = None,
    metrics: list[Metric] | None = None,
    alternative_hostname: HostnameInput | None = None,
) -> CheckResult

Generates a CheckResult object indicating an UNKNOWN check state.

This function creates and returns a CheckResult indicating the system or component is in an UNKNOWN state. It allows for passing additional information such as a summary, details, name suffix, metrics, or an alternative hostname.

Parameters:

  • summary

    (str) –

    A summary of the check result, indicating the outcome of the check. Checkmk will show this summary on pages like a host overview.

  • details

    (Details | None, default: None ) –

    A detailed output of the check result, providing additional information beyond the summary. Checkmk will show this detailed output when you are viewing a specific service.

  • name_suffix

    (str | None, default: None ) –

    A suffix to add to the check name as defined in the @check decorator. This enables a single check function to return multiple results that create multiple services on the Checkmk side.

  • metrics

    (list[Metric] | None, default: None ) –

    An optional list of metrics to associate with the check.

  • alternative_hostname

    (HostnameInput | None, default: None ) –

    An optional hostname that overrides the hostname that would have been used otherwise.

Returns:

  • CheckResult –

    A CheckResult object representing the UNKNOWN check result.

warn 

warn(
    summary: str,
    details: Details | None = None,
    name_suffix: str | None = None,
    metrics: list[Metric] | None = None,
    alternative_hostname: HostnameInput | None = None,
) -> CheckResult

Generates a CheckResult object indicating a WARN check state.

This function creates and returns a CheckResult indicating the system or component is in a WARN state. It allows for passing additional information such as a summary, details, name suffix, metrics, or an alternative hostname.

Parameters:

  • summary

    (str) –

    A summary of the check result, indicating the outcome of the check. Checkmk will show this summary on pages like a host overview.

  • details

    (Details | None, default: None ) –

    A detailed output of the check result, providing additional information beyond the summary. Checkmk will show this detailed output when you are viewing a specific service.

  • name_suffix

    (str | None, default: None ) –

    A suffix to add to the check name as defined in the @check decorator. This enables a single check function to return multiple results that create multiple services on the Checkmk side.

  • metrics

    (list[Metric] | None, default: None ) –

    An optional list of metrics to associate with the check.

  • alternative_hostname

    (HostnameInput | None, default: None ) –

    An optional hostname that overrides the hostname that would have been used otherwise.

Returns:

  • CheckResult –

    A CheckResult object representing the WARN check result.

scheduling_strategy

Scheduling strategies and validation.

Defines how Watchpost decides whether a check should run from the current execution environment against a target environment. Strategies encapsulate constraints (where a check must execute from, which targets it may run against, or that current and target must match) and combine into a single scheduling decision.

Main concepts:

  • SchedulingStrategy: Protocol implemented by concrete strategies.
  • SchedulingDecision: Outcome used by the executor (SCHEDULE, SKIP, DONT_SCHEDULE).
  • DetectImpossibleCombinationStrategy: Validation strategy that detects and raises InvalidCheckConfiguration when multiple strategies are mutually incompatible, so a check could never be scheduled.

Notes:

  • These rules affect what Checkmk will eventually see: if a check is never scheduled in a given environment, no service is reported from that node.
  • "Current == Target" refers to running a check from the same environment that it targets (common for in-cluster checks). When this is required, the allowed execution environments and allowed target environments must overlap; otherwise the configuration is impossible.

Example:

from watchpost.environment import Environment from watchpost.datasource import Datasource from watchpost.scheduling_strategy import ( ... MustRunInGivenExecutionEnvironmentStrategy, ... MustRunAgainstGivenTargetEnvironmentStrategy, ... )

Monitoring = Environment("Monitoring") Preprod = Environment("Preprod")

class LogSystem(Datasource): ... # Run from Monitoring, but may target Monitoring or Preprod ... scheduling_strategies = ( ... MustRunInGivenExecutionEnvironmentStrategy(Monitoring), ... MustRunAgainstGivenTargetEnvironmentStrategy(Monitoring, Preprod), ... )

The Watchpost app collects the strategies required by involved datasources and applies them to each check to derive the final scheduling decision.

Classes:

DetectImpossibleCombinationStrategy

Bases: SchedulingStrategy

Detect mutually incompatible scheduling constraints and raise an error.

This strategy does not itself restrict scheduling. Instead, it validates the active strategies for a check and raises InvalidCheckConfiguration when constraints cannot be satisfied together.

Examples:

  • Conflicting required execution environments with no overlap (for example, one datasource requires "Monitoring" while another requires "Preprod").
  • Allowed target environments that do not include all environments declared in the @check decorator.
  • A "current == target" requirement while the allowed execution and target sets do not overlap, so the check can never run.

InvalidCheckConfiguration 

InvalidCheckConfiguration(
    check: Check,
    reason: str,
    cause: Exception | None = None,
)

Bases: Exception

Should be raised when a check's configuration is contradictory or otherwise invalid and cannot be scheduled anywhere until corrected.

This can happen if, for example, two datasources with conflicting scheduling requirements are used in a single check.

MustRunAgainstGivenTargetEnvironmentStrategy 

MustRunAgainstGivenTargetEnvironmentStrategy(
    *environments: Environment,
)

Bases: SchedulingStrategy

Restrict which target environments the check may run against.

Use this to allow only specific targets (for example, only "stage"). If the requested target is not in the allowed set, the decision is DONT_SCHEDULE.

This strategy does not affect from where the check executes, i.e. the execution environment.

    One or more target environments that the check may run against.

MustRunInGivenExecutionEnvironmentStrategy 

MustRunInGivenExecutionEnvironmentStrategy(
    *environments: Environment,
)

Bases: SchedulingStrategy

Require that the check executes from one of the given execution environments.

Use this when a datasource or check can only run from specific locations (for example, from a monitoring environment). If the current execution environment is not in the allowed set, the decision is DONT_SCHEDULE.

    One or more execution environments from which the check may run.

MustRunInTargetEnvironmentStrategy

Bases: SchedulingStrategy

Require that the current execution environment equals the target environment.

This models "run in-environment" behavior where the check must execute inside the environment it is checking. If current != target, the decision is DONT_SCHEDULE.

SchedulingDecision

Bases: IntEnum

Scheduling decision returned by a SchedulingStrategy.

The value communicates what the executor should do with a given check in the current situation. Use this to express whether a check should be run now, temporarily skipped (reusing a previously cached result if available), or never be scheduled on this node.

If you determine that the check's configuration is invalid and can never be scheduled until fixed, raise InvalidCheckConfiguration instead!

Notes on semantics: - SCHEDULE: The check is eligible to run now. - SKIP: Conditions are temporarily unfavorable; the last known result will be reused if available. The strategy expects that conditions might change later (e.g., outside a maintenance window, or when a datasource becomes reachable again). If no cached result is available, the check will be marked as UNKNOWN. - DONT_SCHEDULE: This node/environment should never run the check (e.g., the check is pinned to another environment). Different from SKIP in that this is a permanent decision for the current environment, not reporting on the check at all.

Attributes:

  • DONT_SCHEDULE –

    The check should not run at all.

  • SCHEDULE –

    The check is eligible to run now.

  • SKIP –

    The check should be skipped temporarily and any prior results, if available,

DONT_SCHEDULE class-attribute instance-attribute 

DONT_SCHEDULE = 2

The check should not run at all.

This node/environment should never run the check (e.g., the check is pinned to another environment). Different from SKIP in that this is a permanent decision for the current environment, not reporting on the check at all.

This primarily applies for checks that have to run from a certain environment, and the current environment is not suitable.

SCHEDULE class-attribute instance-attribute 

SCHEDULE = 0

The check is eligible to run now.

SKIP class-attribute instance-attribute 

SKIP = 1

The check should be skipped temporarily and any prior results, if available, should be reused.

This usually means that the conditions are temporarily unfavorable; the last known result will be reused if available. The strategy expects that conditions might change later (e.g., outside a maintenance window, or when a datasource becomes reachable again). If no cached result is available, the check will be marked as UNKNOWN.

SchedulingStrategy

Bases: Protocol

Protocol for pluggable scheduling logic.

A scheduling strategy decides if and how a check should be executed in the current context. Implementations can consider the check definition, the environment that is attempting to execute the check (current execution environment), and the target environment the check is meant to observe.

Typical concerns handled by a strategy include: - Aligning check execution with datasource availability windows - Respecting environment pinning (e.g., only run from a specific location) - Handling temporary outages by skipping while reusing previous results - Detecting invalid configurations that can never be scheduled

Methods:

  • schedule –

    Decide whether and how the given check should be executed.

schedule 

schedule(
    check: Check,
    current_execution_environment: Environment,
    target_environment: Environment,
) -> SchedulingDecision

Decide whether and how the given check should be executed.

Parameters:

  • check
    (Check) –

    The check definition whose execution is being considered.

  • current_execution_environment
    (Environment) –

    The environment that the check would execute within.

  • target_environment
    (Environment) –

    The environment that the check is intended to observe or act upon.

Returns:

Raises:

  • InvalidCheckConfiguration –

    The check's configuration is contradictory or otherwise invalid and cannot be scheduled anywhere until corrected.

utils

Utility helpers for Watchpost.

Classes:

Functions:

InvocationInformation dataclass 

InvocationInformation(relative_path: str, line_number: int)

Call-site information for an invocation.

Holds the relative path of the module and the line number where the invocation occurred.

get_invocation_information 

get_invocation_information() -> (
    InvocationInformation | None
)

Return call-site information for the caller's caller.

Inspects the call stack to locate the module and line number of the function that invoked the caller. The module path is made relative to the project root inferred from the watchpost package. If any information cannot be determined, returns None.

Returns:

normalize_to_timedelta 

normalize_to_timedelta(
    value: timedelta | str | None,
) -> timedelta | None

Normalize a value to a datetime.timedelta.

  • If value is None, return None.
  • If value is already a timedelta, return it unchanged.
  • If value is a string, parse it with timelength (e.g., "5m", "2h30m") and return the corresponding timedelta.

Parameters:

  • value

    (timedelta | str | None) –

    A timedelta, a string time expression, or None.

Returns:

  • timedelta | None –

    A timedelta corresponding to the input, or None.

vendored

Modules:

local_proxy

Classes:

  • LocalProxy –

    A proxy to the object bound to a context-local object. All

LocalProxy 

LocalProxy(
    local: ContextVar[T] | Callable[[], T],
    name: str | None = None,
    *,
    unbound_message: str | None = None,
)

A proxy to the object bound to a context-local object. All operations on the proxy are forwarded to the bound object. If no object is bound, a RuntimeError is raised.

:param local: The context-local object that provides the proxied object. :param name: Proxy this attribute from the proxied object. :param unbound_message: The error message to show if the context-local object is unbound.

Proxy a :class:~contextvars.ContextVar to make it easier to access. Pass a name to proxy that attribute.

.. code-block:: python

_request_var = ContextVar("request")
request = LocalProxy(_request_var)
session = LocalProxy(_request_var, "session")

Proxy an attribute on a :class:Local namespace by calling the local with the attribute name:

.. code-block:: python

data = Local()
user = data("user")

Proxy the top item on a :class:LocalStack by calling the local. Pass a name to proxy that attribute.

.. code-block::

app_stack = LocalStack()
current_app = app_stack()
g = app_stack("g")

Pass a function to proxy the return value from that function. This was previously used to access attributes of local objects before that was supported directly.

.. code-block:: python

session = LocalProxy(lambda: request.session)

__repr__ and __class__ are proxied, so repr(x) and isinstance(x, cls) will look like the proxied object. Use issubclass(type(x), LocalProxy) to check if an object is a proxy.

.. code-block:: python

repr(user)  # <User admin>
isinstance(user, User)  # True
issubclass(type(user), LocalProxy)  # True

.. versionchanged:: 2.2.2 __wrapped__ is set when wrapping an object, not only when wrapping a function, to prevent doctest from failing.

.. versionchanged:: 2.2 Can proxy a ContextVar or LocalStack directly.

.. versionchanged:: 2.2 The name parameter can be used with any proxied object, not only Local.

.. versionchanged:: 2.2 Added the unbound_message parameter.

.. versionchanged:: 2.0 Updated proxied attributes and methods to reflect the current data model.

.. versionchanged:: 0.6.1 The class can be instantiated with a callable.