API Reference

Core Classes

class Probe(*args, **kwargs)[source]

Bases: Protocol

Protocol defining the interface for a probe.

property info: Dict[str, Any] | None

add_info(name, value)[source]

Return type:: None

async run_check()[source]

Return type:: ProbeResult

__init__(*args, **kwargs)

class ProbeBase(name=None, timeout=None, success_ttl=None, failed_ttl=None)[source]

Bases: object

Base implementation of a probe.

__init__(name=None, timeout=None, success_ttl=None, failed_ttl=None)[source]

Initialize the probe.

Parameters:

name (Optional[str]) – The name of the probe.
timeout (Optional[int]) – The timeout for the probe.
success_ttl (Union[int, timedelta, None]) – Cache duration for successful results. If None, successful results are not cached.
failed_ttl (Union[int, timedelta, None]) – Cache duration for failed results. If None, failed results are not cached.

add_info(name, value)[source]

Add information to the probe result.

Parameters:

name (str) – The name of the information.
value (Any) – The value of the information.

Return type:

None

property info: Dict[str, Any] | None

Get the information added to the probe result.

Returns:: The information added to the probe result.

async run_check()[source]

Run the check and return the result.

Return type:: ProbeResult
Returns:: The result of the check.

class CallableProbe(func, name=None, timeout=None, success_ttl=None, failed_ttl=None)[source]

Bases: ProbeBase

A probe that wraps a callable function.

__init__(func, name=None, timeout=None, success_ttl=None, failed_ttl=None)[source]

Initialize the probe.

Parameters:

name (Optional[str]) – The name of the probe.
timeout (Optional[int]) – The timeout for the probe.
success_ttl (Union[int, timedelta, None]) – Cache duration for successful results. If None, successful results are not cached.
failed_ttl (Union[int, timedelta, None]) – Cache duration for failed results. If None, failed results are not cached.

class Probirka(success_ttl=None, failed_ttl=None)[source]

Bases: object

Probirka is a health check manager that allows adding and running probes.

__init__(success_ttl=None, failed_ttl=None)[source]

Initialize the Probirka instance.

Parameters:

success_ttl (Union[int, timedelta, None]) – Default cache duration for successful results. If None, successful results are not cached.
failed_ttl (Union[int, timedelta, None]) – Default cache duration for failed results. If None, failed results are not cached.

add_info(name, value)[source]

Add information to the health check result.

Parameters:

name (str) – The name of the information.
value (Any) – The value of the information.

Return type:

None

property info: Dict[str, Any] | None

Get the information added to the health check result.

Returns:: The information added to the health check result.

add_probes(*probes, groups='')[source]

Add probes to the health check.

Parameters:

probes (Probe) – Probes to add
groups (Union[str, List[str]]) – Groups for optional probes. Probes without groups are required.

Return type:

None

add(name=None, timeout=None, groups='', success_ttl=None, failed_ttl=None)[source]

Decorator to add a callable as a probe.

Parameters:

name (Optional[str]) – Probe name
timeout (Optional[int]) – Probe timeout in seconds
groups (Union[str, List[str]]) – Groups for optional probes. Probes without groups are required.
success_ttl (Union[int, timedelta, None]) – Cache duration for successful results. If None, uses the global success_ttl setting.
failed_ttl (Union[int, timedelta, None]) – Cache duration for failed results. If None, uses the global failed_ttl setting.

Return type:

Callable

Returns:

Decorated function

async run(timeout=None, with_groups='', skip_required=False)[source]

Run health check and return results.

Parameters:

timeout (Optional[int]) – Overall timeout in seconds
with_groups (Union[str, List[str]]) – Groups to run. Required probes run unless skip_required=True
skip_required (bool) – Skip probes without groups

Return type:

ProbirkaResult

Returns:

Health check result

class ProbeResult(name, ok, cached, started_at, elapsed, info, error)[source]

Bases: object

name: str

ok: bool

cached: Optional[bool]

started_at: datetime

elapsed: timedelta

info: Optional[Dict[str, Any]]

error: Optional[str]

__init__(name, ok, cached, started_at, elapsed, info, error)

class ProbirkaResult(ok, started_at, elapsed, info, checks)[source]

Bases: object

ok: bool

started_at: datetime

elapsed: timedelta

info: Optional[Dict[str, Any]]

checks: Sequence[ProbeResult]

__init__(ok, started_at, elapsed, info, checks)

Integrations

FastAPI

make_fastapi_endpoint(probirka, timeout=None, with_groups='', skip_required=False, return_results=True, success_code=200, error_code=500)[source]

Create a FastAPI endpoint for a given Probirka instance.

Parameters:

probirka (Probirka) – The Probirka instance to run.
timeout (Optional[int]) – The timeout for the Probirka run.
with_groups (Union[str, List[str]]) – Groups to include in the Probirka run.
skip_required (bool) – Whether to skip required checks.
return_results (bool) – Whether to return the results in the response.
success_code (int) – The HTTP status code for a successful response.
error_code (int) – The HTTP status code for an error response.

Returns:

The FastAPI endpoint.

Return type:

Callable[[], Coroutine[Any, Any, Response]]

AIOHTTP

make_aiohttp_endpoint(probirka, timeout=None, with_groups='', skip_required=False, return_results=True, success_code=200, error_code=500)[source]

Create an aiohttp endpoint for a given Probirka instance.

Parameters:

probirka (Probirka) – The Probirka instance to run.
timeout (Optional[int]) – The timeout for the Probirka run.
with_groups (Union[str, List[str]]) – Groups to include in the Probirka run.
skip_required (bool) – Whether to skip required checks.
return_results (bool) – Whether to return the results in the response.
success_code (int) – The HTTP status code for a successful response.
error_code (int) – The HTTP status code for an error response.

Returns:

The aiohttp endpoint.

Return type:

Callable[[web.Request], Coroutine[Any, Any, web.Response]]