API Reference

Core Library

Domain Model

Domain Models

Domain models for checks, results, and services.

class nyxmon.domain.models.Check(*, check_id: int, service_id: int, name: str = '', check_type: str, url: str, check_interval: int = 300, next_check_time: int = 0, processing_started_at: int = 0, status: Literal['idle', 'processing'] = 'idle', disabled: bool = False, data: dict)

Bases: object

execute() → None

schedule_next_check() → None

Schedule the next execution of this check.

class nyxmon.domain.models.CheckResult(check: Check, result: Result)

Bases: object

property passed: bool

property should_notify: bool

class nyxmon.domain.models.CheckStatus

Bases: object

IDLE: Literal['idle'] = 'idle'

PROCESSING: Literal['processing'] = 'processing'

class nyxmon.domain.models.CheckType

Bases: object

DNS: Literal['dns'] = 'dns'

HTTP: Literal['http'] = 'http'

IMAP: Literal['imap'] = 'imap'

JSON_HTTP: Literal['json-http'] = 'json-http'

JSON_METRICS: Literal['json-metrics'] = 'json-metrics'

PING: Literal['ping'] = 'ping'

SMTP: Literal['smtp'] = 'smtp'

TCP: Literal['tcp'] = 'tcp'

class nyxmon.domain.models.Result(*, result_id: int | None = None, check_id: int, status: Literal['ok', 'warning', 'error'], data: dict)

Bases: object

class nyxmon.domain.models.ResultStatus

Bases: object

ERROR: Literal['error'] = 'error'

OK: Literal['ok'] = 'ok'

WARNING: Literal['warning'] = 'warning'

class nyxmon.domain.models.Service(*, service_id: int, data: dict)

Bases: object

class nyxmon.domain.models.StatusChoices

Bases: object

FAILED: Literal['failed'] = 'failed'

PASSED: Literal['passed'] = 'passed'

RECOVERING: Literal['recovering'] = 'recovering'

UNKNOWN: Literal['unknown'] = 'unknown'

WARNING: Literal['warning'] = 'warning'

classmethod get_css_class(status: str) → str

Message Bus

Central event dispatcher for commands and events.

class nyxmon.service_layer.message_bus.MessageBus(uow: UnitOfWork, event_handlers: dict[type[Event], list[Callable]], command_handlers: dict[type[Command], Callable])

Bases: object

This is a simple implementation of a message bus that can handle messages of type Event and Command using event_handlers and command_handlers. It uses a UnitOfWork to collect new events.

handle(message: Event | Command)

handle_command(command: Command)

handle_event(event: Event)

Check Executors

HTTP Check Executor

HTTP check executor implementation.

class nyxmon.adapters.runner.executors.http_executor.HttpCheckExecutor(client: AsyncClient | None = None)

Bases: object

Executor for HTTP checks.

Performs HTTP requests and validates responses. Can accept an external client or create its own.

async aclose() → None

Close the HTTP client if we created it.

Only closes the client if this executor created it. Externally provided clients are not closed.

async execute(check: Check) → Result

Execute an HTTP check and return a Result.

Parameters:

check – The HTTP check to execute

Returns:

Result with HTTP response information

DNS Check Executor

DNS check executor implementation.

class nyxmon.adapters.runner.executors.dns_executor.DnsCheckExecutor(resolver: DnsResolver | None = None)

Bases: object

Executor for DNS checks.

Performs DNS queries and validates results against expected IPs.

async aclose() → None

Close the executor.

DNS executor doesn’t manage any resources, so this is a no-op.

async execute(check: Check) → Result

Execute a DNS check and return a Result.

class nyxmon.adapters.runner.executors.dns_executor.DnsResolver(*args, **kwargs)

Bases: Protocol

Protocol describing the resolver interface expected by the executor.

async query(domain: str, config: DnsCheckConfig) → DnsResolverResult

Resolve domain according to config and return structured data.

class nyxmon.adapters.runner.executors.dns_executor.DnsResolverResult(records: List[str], metadata: dict[str, Any])

Bases: object

Structured result returned by DNS resolvers.

metadata: dict[str, Any]

records: List[str]

class nyxmon.adapters.runner.executors.dns_executor.DnspythonResolver

Bases: object

Resolver implementation backed by dnspython.

async query(domain: str, config: DnsCheckConfig) → DnsResolverResult

Repository Interfaces

class nyxmon.adapters.repositories.interface.CheckRepository(*args, **kwargs)

Bases: Protocol

A repository interface for storing and retrieving checks.

add(check) → None

Add a check to the repository.

get(check_id: int)

Get a check from the repository by ID.

list() → List[Check]

Get a list of all checks.

async list_async() → List[Check]

Get a list of all checks asynchronously.

seen: set

class nyxmon.adapters.repositories.interface.RepositoryStore(*args, **kwargs)

Bases: Protocol

A protocol for a collection of repositories.

checks: CheckRepository

list() → List[ResultRepository | CheckRepository | ServiceRepository]

Get a list of all repositories.

results: ResultRepository

services: ServiceRepository

class nyxmon.adapters.repositories.interface.ResultRepository(*args, **kwargs)

Bases: Protocol

A repository interface for storing and retrieving results.

add(result: Result) → None

Add a result to the repository.

get(result_id: int) → Result

Get a result from the repository by ID.

list() → List[Result]

Get a list of all results.

list_for_check(check_id: int, limit: int) → List[Result]

Get recent results for a check, newest first.

seen: set[Result]

class nyxmon.adapters.repositories.interface.ServiceRepository(*args, **kwargs)

Bases: Protocol

A repository interface for storing and retrieving services.

add(service) → None

Add a service to the repository.

get(service_id: int)

Get a service from the repository by ID.

list() → List[Service]

Get a list of all services.

seen: set

Runner

Async Check Runner

class nyxmon.adapters.runner.async_runner.AsyncCheckRunner(portal_provider: BlockingPortalProvider)

Bases: CheckRunner

run_all(checks: Iterable[Check], result_received: Callable) → None

Run all checks.

Notifications

class nyxmon.adapters.notification.AsyncTelegramNotifier(token: str | None = None, chat_id: str | None = None)

Bases: Notifier

async async_notify_check_failed(check: Check, result: Result) → None

Notify about a failed check asynchronously.

async async_notify_service_status_changed(service: Service, status: str) → None

Notify about a service status change asynchronously.

async async_send(text: str, high_priority: bool = False) → None

Send a message via Telegram asynchronously.

static escape_markdown_v2(text: str) → str

Escape special characters for Telegram’s MarkdownV2 format.

notify_check_failed(check: Check, result: Result) → None

Notify about a failed check.

notify_service_status_changed(service: Service, status: str) → None

Notify about a service status change.

set_portal_provider(portal_provider: BlockingPortalProvider) → None

Set the portal provider for async operations.

class nyxmon.adapters.notification.LoggingNotifier(*args, **kwargs)

Bases: Notifier

A simple notifier that logs messages to the console.

notify_check_failed(check: Check, result: Result) → None

Log a failed check notification.

notify_service_status_changed(service: Service, status: str) → None

Log a service status change notification.

set_portal_provider(portal_provider: BlockingPortalProvider) → None

Set the portal provider for async operations.

class nyxmon.adapters.notification.Notifier(*args, **kwargs)

Bases: Protocol

Interface for notification services.

notify_check_failed(check: Check, result: Result) → None

Notify about a failed check.

notify_service_status_changed(service: Service, status: str) → None

Notify about a service status change.