API Reference

Core Classes

The core a probe is built on: the protocol, the base implementation, the callable wrapper, the failure.

ProbeCallable

Zero-argument sync or async callable returning True/None on success, False on failure.

alias of Callable[[], bool | Awaitable[bool | None] | None]

format_error(exc)[source]

Format an exception for the error field of a result.

Always includes the exception type so that exceptions with an empty message (TimeoutError, NotImplementedError, …) stay informative.

Parameters:: exc (BaseException) – The exception to format.
Return type:: str
Returns:: 'ExcType: message' or just 'ExcType' when the message is empty.

exception ProbeFailure[source]

Bases: Exception

Raised by a probe when the dependency answered, but not the way a healthy one should.

Examples: Redis replied to PING with something other than True, an HTTP endpoint returned an unexpected status code. The message ends up in ProbeResult.error.

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

Bases: Protocol

Protocol defining the interface for a probe.

property name: str: Probe name used in results.

property info: dict[str, Any] | None: Metadata attached to the probe, None if nothing was added.

add_info(name, value)[source]

Attach a key/value pair to the probe result.

Return type:: None

async run_check()[source]

Run the check and return its result; must not raise.

Return type:: ProbeResult

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

Bases: ABC

Base implementation of a probe.

Initialize the probe.

Parameters:

name (str | None) – The name of the probe.
timeout (int | None) – The timeout for the probe.
success_ttl (int | timedelta | None) – Cache duration for successful results. If None, successful results are not cached.
failed_ttl (int | timedelta | None) – Cache duration for failed results. If None, failed results are not cached.
allow_failure (bool) – If True, a failure of this probe does not affect the overall ProbirkaResult.ok; the failure is still visible in the probe’s own result.

property name: str

Get the name of the probe.

Returns:: The name of the probe.

property allow_failure: bool

Whether this probe may fail without affecting the overall result.

Returns:: The allow_failure setting of the probe.

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, allow_failure=False)[source]

Bases: ProbeBase

A probe that wraps a callable function.

Coroutine functions are awaited directly. Plain functions are executed in the event loop’s default executor so they neither block the loop nor escape the probe timeout. If a plain callable returns an awaitable (for example an object with an async def __call__ or a function returning a coroutine), that awaitable is awaited too. Note that on timeout the worker thread keeps running until the function returns; only the wait is cancelled.

Initialize the probe.

Parameters:

func (Callable[[], bool | Awaitable[bool | None] | None]) – Sync or async callable returning True/None on success, False on failure.
name (str | None) – The name of the probe. Defaults to func.__name__, falling back to the class name for callables without one (functools.partial, objects with __call__).
timeout (int | None) – The timeout for the probe.
success_ttl (int | timedelta | None) – Cache duration for successful results. If None, successful results are not cached.
failed_ttl (int | timedelta | None) – Cache duration for failed results. If None, failed results are not cached.
allow_failure (bool) – If True, a failure of this probe does not affect the overall result.

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

Bases: object

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

Initialize the Probirka instance.

Parameters:

success_ttl (int | timedelta | None) – Default cache duration for successful results. If None, successful results are not cached.
failed_ttl (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='', allow_failure=None)[source]

Add probes to the health check.

Parameters:

probes (Probe) – Probes to add
groups (str | Sequence[str]) – Groups for optional probes. Probes without groups are required.
allow_failure (bool | None) – Group-level override of the probes’ own allow_failure. None (default) leaves the group as is, so every probe keeps its own setting. True/False applies to all probes of the given groups, including ones added earlier, and replaces a value set by a previous call. A probe that is run through several sources (required list, several groups) is allowed to fail only if every source allows it. Requires groups.

Raises:

ValueError – if allow_failure is given without groups

Return type:

None

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

Register the decorated callable as a probe.

Parameters:

name (str | None) – Probe name
timeout (int | None) – Probe timeout in seconds
groups (str | Sequence[str]) – Groups for optional probes. Probes without groups are required.
success_ttl (int | timedelta | None) – Cache duration for successful results. If None, uses the global success_ttl setting. Pass 0 to disable caching of successful results for this probe.
failed_ttl (int | timedelta | None) – Cache duration for failed results. If None, uses the global failed_ttl setting. Pass 0 to disable caching of failed results for this probe.
allow_failure (bool) – If True, a failure of this probe does not affect the overall ok. A group-level allow_failure passed to add_probes() overrides this.

Return type:

Returns:

Decorated function

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

Run health check and return results.

The overall timeout never raises: probes that did not finish in time are reported as failed and error is set. ok is False only if a probe without allow_failure failed or did not finish in time.

Parameters:

timeout (int | None) – Overall timeout in seconds
with_groups (str | Sequence[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, allow_failure=False)[source]

Bases: object

Outcome of a single probe run.

Parameters:

name (str) – Probe name.
ok (bool) – Whether the check passed.
cached (bool | None) – True if served from cache, False if freshly computed with caching enabled, None if the probe has no TTL configured.
started_at (datetime) – When the check started, timezone-aware in the local zone of the host.
elapsed (timedelta) – How long the check took, measured with a monotonic clock.
info (dict[str, Any] | None) – Metadata added via ProbeBase.add_info().
error (str | None) – 'ExcType: message' when the check failed with an exception or timed out. Values the probe registered as secrets (the password of its connection string, its request headers) are already replaced with '***'.
allow_failure (bool) – True if this probe is allowed to fail without affecting ProbirkaResult.ok. Effective value for the run: the probe’s own setting, overridden by the group it was run in when that group sets allow_failure.

name: str

ok: bool

cached: bool | None

started_at: datetime

elapsed: timedelta

info: dict[str, Any] | None

error: str | None

allow_failure: bool = False

to_dict(*, redact=True)[source]

JSON-compatible representation of the result.

started_at is an ISO 8601 string carrying the UTC offset, elapsed is the duration in seconds.

Parameters:: redact (bool) – Mask secrets on the way out (the default): values of info under keys like password or api_key become '***', and passwords inside URLs and sensitive query parameters are masked in info strings and in error. Pass False to get info and error exactly as the probe produced them.
Return type:: dict[str, Any]

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

Bases: object

Aggregated outcome of a Probirka.run() call.

Parameters:

ok (bool) – True if every probe without allow_failure passed. Probes with allow_failure may fail or time out without affecting it.
started_at (datetime) – When the run started, timezone-aware in the local zone of the host.
elapsed (timedelta) – How long the whole run took, measured with a monotonic clock.
info (dict[str, Any] | None) – Metadata added via Probirka.add_info().
checks (Sequence[ProbeResult]) – Per-probe results in registration order.
error (str | None) – Timeout message when the overall timeout was hit (even if only probes with allow_failure did not finish), otherwise None.

ok: bool

started_at: datetime

elapsed: timedelta

info: dict[str, Any] | None

checks: Sequence[ProbeResult]

error: str | None = None

to_dict(*, redact=True)[source]

JSON-compatible representation of the result, shared by all framework integrations.

started_at is an ISO 8601 string carrying the UTC offset, elapsed is the duration in seconds, checks is a list of ProbeResult.to_dict().

Parameters:: redact (bool) – Mask secrets in info and in every check, see ProbeResult.to_dict(). True by default; the framework integrations always leave it on.
Return type:: dict[str, Any]

Lazy resolution of the public names that need a third-party package (private).

The package root binds module_getattr() and module_dir() as its PEP 562 hooks, so that import probirka never imports a client library or a framework: a probe or an adapter is imported on first access, and MissingDependencyError tells which package to install when that fails.

ROOT = 'probirka': The package whose namespace caches the resolved names (probirka).

exception MissingDependencyError[source]

Bases: ImportError

Raised when a lazily loaded name needs a package that is not installed.

Deliberately an ImportError rather than an AttributeError: from probirka import X turns an AttributeError into a bare “cannot import name”, dropping the install hint. The consequence is that hasattr(probirka, 'RedisProbe') raises on a bare install; use importlib.util.find_spec to feature-detect a client library instead.

installed(import_name)[source]

Report whether a package can be imported, without importing it.

Parameters:: import_name (str) – The top-level import name of the package.
Return type:: bool
Returns:: True if an import spec is found.

available()[source]

List the lazily resolved names whose package is installed.

The package root appends them to __all__, so from probirka import * brings the adapters and probes you can actually use and still works with nothing but probirka installed.

Return type:: list[str]
Returns:: The names, sorted.

module_getattr(name)[source]

Resolve a lazily loaded name of the package root (its PEP 562 __getattr__).

Parameters:

name (str) – The attribute looked up on the package.

Return type:

Any

Returns:

The probe class or adapter factory.

Raises:

AttributeError – If the name is not a lazily loaded one.
MissingDependencyError – If the package the name needs is not installed.

module_dir()[source]

List the attributes of the package root, lazily loaded names included (its PEP 562 __dir__).

Return type:: list[str]
Returns:: The attribute names, sorted.

Secrets

MASK, mask_url and redact_value are importable from the probirka package.

Keep secrets out of health check output.

Two complementary techniques, both stdlib-only:

known values — a probe registers the password of its connection string and the values of its request headers, and redact_secrets() replaces them wherever they show up in a message from the client library (the no_log approach of Ansible);
shape-based — redact_value() walks the info metadata and masks values under suspicious keys (password, token, api_key, …) and passwords inside URL-like strings (the approach of Django’s SafeExceptionReporterFilter and Sentry’s EventScrubber).

Only the password part of a URL is masked, never the user name, host or database: those are what makes an error message useful, and it is what SQLAlchemy and Celery do as well.

MASK = '***': What a secret is replaced with.

MIN_SECRET_LENGTH = 4

masking them would garble unrelated text.

Type:: Shorter values are never treated as secrets

mask_url(url, mask='***')[source]

Return url with its password replaced by mask.

Everything else is kept, so postgresql://app:hunter2@db:5432/app becomes postgresql://app:***@db:5432/app. A URL without a password is returned unchanged.

Parameters:

url (str) – The URL or connection string.
mask (str) – The replacement for the password.

Return type:

str

Returns:

The masked URL.

secrets_from_url(url)[source]

Return the password of a URL in every form a client library may echo it.

As written in the URL (p%40ss), decoded (p@ss), and as the HTTP basic auth credentials HTTP clients derive from the user info (Basic + base64 of user:password): httpx and aiohttp move user:pass@ out of the URL into an Authorization header.

Parameters:: url (str | None) – The URL or connection string, None when the probe was given a client instead.
Return type:: tuple[str, ...]
Returns:: Strings to register with ProbeBase._register_secrets().

secrets_from_headers(headers)[source]

Return the values of request headers, whole and without an auth scheme prefix.

Every header value is treated as a secret: a probe sends only what is needed to be let in, and a Bearer token is a secret even under a custom header name. For Authorization: Bearer abc... both Bearer abc... and abc... are returned.

Parameters:: headers (Mapping[str, str] | None) – The request headers, None when there are none.
Return type:: tuple[str, ...]
Returns:: Strings to register with ProbeBase._register_secrets().

redact_secrets(text, secrets)[source]

Replace every occurrence of the known secrets in text with MASK.

Longer secrets are replaced first, so a secret that contains another one does not leave a fragment behind.

Parameters:

text (str) – The text to clean, typically an exception message.
secrets (Iterable[str]) – Known secret values; empty and too short ones are ignored.

Return type:

str

Returns:

The cleaned text.

redact_string(text)[source]

Mask passwords in URLs and sensitive query parameters found in text.

A heuristic for text whose origin is unknown, like a message from a client library or a value of info: amqp://guest:guest@mq/ becomes amqp://guest:***@mq/ and ?token=abc becomes ?token=***.

Parameters:: text (str) – The text to clean.
Return type:: str
Returns:: The cleaned text.

redact_value(value)[source]

Recursively clean a JSON-like value, as found in info.

Mappings get the value under a sensitive-looking key (password, api_key, Authorization, …) replaced with MASK whatever it is; other values are cleaned recursively. Lists and tuples are cleaned element-wise, strings go through redact_string(). Anything else — numbers, None, arbitrary objects — is returned as is; the framework adapters clean the str() of an object when they serialize it.

Parameters:: value (Any) – The value to clean.
Return type:: Any
Returns:: A cleaned copy, or the same object when there was nothing to clean.

Ready-made Probes

All probes below are importable from the probirka package, e.g. from probirka import RedisProbe, and so are HttpProbePolicy and HttpProbePolicyViolation.

ClientOrFactory

Either a ready client instance or a zero-argument function (lambda, functools.partial) returning one.

alias of T | Callable[[], T]

resolve(client_or_factory)[source]

Return the client itself, or call the factory to obtain it.

Lets a probe be constructed before the connection pool exists (for example in a FastAPI lifespan): pass lambda: app.state.pool instead of the pool.

Only plain functions, methods, lambdas and functools.partial objects are treated as factories; a client object that happens to define __call__ is returned as is.

Parameters:: client_or_factory (TypeVar(T) | Callable[[], TypeVar(T)]) – A client instance or a zero-argument function returning one.
Return type:: TypeVar(T)
Returns:: The client instance.

require_exactly_one(**kwargs)[source]

Ensure exactly one of the given keyword arguments is not None.

Parameters:: kwargs (Any) – Mutually exclusive constructor arguments (client=..., dsn=...).
Raises:: ValueError – If none or more than one of them is set.
Return type:: None

class ClientProbeBase(name=None, timeout=None, success_ttl=None, failed_ttl=None, allow_failure=False)[source]

Bases: ProbeBase

Shared lifecycle for probes that talk to a service through a client object.

Subclasses store the user’s client (or factory) in _client and implement two hooks: _temporary_client(), an async context manager that opens and closes a short-lived client from the connection string, and _check_client(), the actual check. An existing client is never closed by the probe.

Initialize the probe.

Parameters:

name (str | None) – The name of the probe.
timeout (int | None) – The timeout for the probe.
success_ttl (int | timedelta | None) – Cache duration for successful results. If None, successful results are not cached.
failed_ttl (int | timedelta | None) – Cache duration for failed results. If None, failed results are not cached.
allow_failure (bool) – If True, a failure of this probe does not affect the overall ProbirkaResult.ok; the failure is still visible in the probe’s own result.

class TcpProbe(host, port, name=None, timeout=None, success_ttl=None, failed_ttl=None, allow_failure=False)[source]

Bases: ProbeBase

Check that a TCP connection to host:port can be established.

Uses only the standard library, so it works without any extra packages. Connection errors (ConnectionRefusedError, socket.gaierror, …) are reported in ProbeResult.error.

Initialize the probe.

Parameters:

host (str) – Host name or IP address to connect to.
port (int) – TCP port.
name (str | None) – The name of the probe. Defaults to 'TcpProbe'.
timeout (int | None) – The timeout for the probe.
success_ttl (int | timedelta | None) – Cache duration for successful results.
failed_ttl (int | timedelta | None) – Cache duration for failed results.
allow_failure (bool) – If True, a failure of this probe does not affect the overall result.

class PostgresAsyncpgProbe(client=None, *, dsn=None, name=None, timeout=None, success_ttl=None, failed_ttl=None, allow_failure=False)[source]

Bases: ClientProbeBase

Check PostgreSQL availability with asyncpg.

Runs SELECT 1 on an existing asyncpg.Pool / asyncpg.Connection (or a callable returning one), or on a short-lived connection opened from dsn.

Initialize the probe.

Parameters:

client (Any | Callable[[], Any] | None) – asyncpg.Pool, asyncpg.Connection or a zero-argument callable returning one.
dsn (str | None) – PostgreSQL DSN, e.g. postgresql://user:pass@host:5432/db. Mutually exclusive with client.
name (str | None) – The name of the probe. Defaults to the class name.
timeout (int | None) – The timeout for the probe.
success_ttl (int | timedelta | None) – Cache duration for successful results.
failed_ttl (int | timedelta | None) – Cache duration for failed results.
allow_failure (bool) – If True, a failure of this probe does not affect the overall result.

class RedisProbe(client=None, *, url=None, name=None, timeout=None, success_ttl=None, failed_ttl=None, allow_failure=False)[source]

Bases: ClientProbeBase

Check Redis availability with redis-py (redis.asyncio).

Sends PING to an existing redis.asyncio.Redis client (or a callable returning one), or to a short-lived client created from url.

Initialize the probe.

Parameters:

client (Any | Callable[[], Any] | None) – redis.asyncio.Redis (or cluster/sentinel client with ping) or a callable returning one.
url (str | None) – Redis URL, e.g. redis://localhost:6379/0. Mutually exclusive with client.
name (str | None) – The name of the probe. Defaults to the class name.
timeout (int | None) – The timeout for the probe.
success_ttl (int | timedelta | None) – Cache duration for successful results.
failed_ttl (int | timedelta | None) – Cache duration for failed results.
allow_failure (bool) – If True, a failure of this probe does not affect the overall result.

Security policy for the HTTP probes: which URLs a probe may request, which redirects it may follow.

A health endpoint requests whatever URL it was configured with. When that URL, or a redirect the target answers with, points at an internal host or a cloud metadata service, the probe becomes a server-side request forgery (SSRF) primitive. HttpProbePolicy limits what a probe will contact. Its defaults are deliberately soft: any http or https URL is fine and redirects are not followed. Everything stricter is opt-in.

exception HttpProbePolicyViolation[source]

Bases: ProbeFailure

Raised when a URL, a redirect target or a resolved address is refused by HttpProbePolicy.

A ProbeFailure: the check ran and the result is a failure with 'HttpProbePolicyViolation: ...' in ProbeResult.error. In the probe constructor the same violation is reported as a ValueError.

normalize_host(host)[source]

Return host in the form used for comparisons: lowercase, without brackets or a trailing dot.

Internationalized names are converted to their IDNA (xn--) form when possible.

Parameters:: host (str) – A host name or IP literal, possibly in [...].
Return type:: str
Returns:: The normalized host.

async resolve_host(host, port=None)[source]

Return every address host currently resolves to, in resolver order and without duplicates.

Uses the event loop’s getaddrinfo; a lookup failure raises socket.gaierror.

Parameters:

host (str) – The host name to resolve.
port (int | None) – The port, passed to the resolver as a hint.

Return type:

tuple[IPv4Address | IPv6Address, ...]

Returns:

The resolved addresses.

class HttpProbePolicy(allowed_schemes=('http', 'https'), allowed_hosts=None, block_private_networks=False, blocked_networks=(), allowed_networks=(), follow_redirects=False, max_redirects=5)[source]

Bases: object

What an HTTP probe may request. Pass it to a probe as policy=.

The URL given to the probe is checked against the policy when the probe is created (a violation is a ValueError), and every URL the probe is about to request, redirect targets included, is checked again before the request (a violation is a HttpProbePolicyViolation in ProbeResult.error). With the defaults any http or https URL passes and redirects are not followed.

Network rules (block_private_networks, blocked_networks, allowed_networks) apply to IP literals in the URL and, when at least one blocking rule is set, to the addresses a host name resolves to. The name is resolved by the probe before the request, and the request is refused if any resolved address is blocked. The client library resolves the name again on its own, so a resolver that answers differently the second time (DNS rebinding) can still reach a blocked address; the policy is not a substitute for network-level controls.

Parameters:

allowed_schemes (Collection[str]) – URL schemes a probe may request; ('http', 'https') by default.
allowed_hosts (Collection[str] | None) – Host names a probe may request, exactly (api.example.com) or as a wildcard for any subdomain (*.example.com matches a.example.com and a.b.example.com but not example.com). None, the default, allows any host.
block_private_networks (bool) – Refuse addresses that are not globally routable: loopback, RFC 1918 and fc00::/7 private ranges, link-local (including the 169.254.169.254 cloud metadata address), carrier-grade NAT, multicast, unspecified and reserved ranges. False by default.
blocked_networks (Collection[str | IPv4Network | IPv6Network]) – Networks to refuse, as '10.0.0.0/8' strings or ipaddress network objects.
allowed_networks (Collection[str | IPv4Network | IPv6Network]) – Networks that are always allowed, taking precedence over both block_private_networks and blocked_networks.
follow_redirects (bool) – Follow 301, 302, 303, 307 and 308 responses. Every redirect target is checked against the policy first; request headers are not sent to a different origin. False by default: a redirect is then an ordinary status code compared with expected_status.
max_redirects (int) – How many redirects to follow at most; 5 by default.

allowed_schemes: Collection[str] = ('http', 'https')

allowed_hosts: Collection[str] | None = None

block_private_networks: bool = False

blocked_networks: Collection[str | IPv4Network | IPv6Network] = ()

allowed_networks: Collection[str | IPv4Network | IPv6Network] = ()

follow_redirects: bool = False

max_redirects: int = 5

property needs_resolution: bool: Whether a host name has to be resolved before the request to enforce the network rules.

check_url(url)[source]

Check a URL without touching the network: scheme, host, allowlist and IP literals.

Parameters:: url (str) – The URL about to be requested.
Return type:: SplitResult
Returns:: The parsed URL, so that callers do not parse it twice.
Raises:: HttpProbePolicyViolation – If the URL is malformed or refused.

blocked_reason(address)[source]

Explain why address may not be contacted, or return None if it may.

Parameters:: address (IPv4Address | IPv6Address) – A resolved or literal IP address.
Return type:: str | None
Returns:: A short reason such as 'is not a global address', or None.

check_address(address, *, host=None)[source]

Refuse address if the network rules block it.

Parameters:

address (IPv4Address | IPv6Address) – A resolved or literal IP address.
host (str | None) – The host name that resolved to it, for the error message.

Raises:

HttpProbePolicyViolation – If the address is blocked.

Return type:

None

async check_resolution(parts)[source]

Resolve the host of a URL that passed check_url() and refuse it if any address is blocked.

Does nothing when no network rule needs resolution, or when the host is an IP literal (already checked by check_url()).

Parameters:

parts (SplitResult) – The parsed URL returned by check_url().

Raises:

HttpProbePolicyViolation – If a resolved address is blocked.
socket.gaierror – If the host cannot be resolved.

Return type:

None

async check(url)[source]

Run every check on a URL: check_url(), then check_resolution().

Parameters:: url (str) – The URL about to be requested.
Return type:: SplitResult
Returns:: The parsed URL.
Raises:: HttpProbePolicyViolation – If the URL is refused.

class HttpProbeBase(url, *, method='GET', expected_status=(200,), headers=None, client=None, policy=None, name=None, timeout=None, success_ttl=None, failed_ttl=None, allow_failure=False)[source]

Bases: ClientProbeBase

Shared logic for HTTP probes: send a request and compare the status code.

Redirects are never left to the client library. The probe sends every request with automatic following disabled and, when its HttpProbePolicy allows redirects, follows them itself so that each target is checked against the policy first.

Subclasses adapt a concrete client library by implementing _temporary_client() (a client for when none was passed, honouring the probe timeout) and _request().

Initialize the probe.

Parameters:

url (str) – URL to request. Must be absolute and allowed by policy.
method (str) – HTTP method, GET by default.
expected_status (int | Collection[int]) – Status code or codes treated as healthy, (200,) by default.
headers (Mapping[str, str] | None) – Extra request headers. Not sent to a different origin after a redirect.
client (Any | Callable[[], Any] | None) – An existing client/session (or a callable returning one). If omitted, a temporary one is created per check with the probe timeout as its request timeout.
policy (HttpProbePolicy | None) – What the probe may request, see HttpProbePolicy. The default allows any http/https URL and does not follow redirects.
name (str | None) – The name of the probe. Defaults to the class name.
timeout (int | None) – The timeout for the probe.
success_ttl (int | timedelta | None) – Cache duration for successful results.
failed_ttl (int | timedelta | None) – Cache duration for failed results.
allow_failure (bool) – If True, a failure of this probe does not affect the overall result.

Raises:

ValueError – If url is refused by policy.

class HttpHttpxProbe(url, *, method='GET', expected_status=(200,), headers=None, client=None, policy=None, name=None, timeout=None, success_ttl=None, failed_ttl=None, allow_failure=False)[source]

Bases: HttpProbeBase

HTTP check with httpx.

Sends a request with an existing httpx.AsyncClient (or a callable returning one) or a temporary one, and fails unless the status code is in expected_status.

Initialize the probe.

Parameters:

url (str) – URL to request. Must be absolute and allowed by policy.
method (str) – HTTP method, GET by default.
expected_status (int | Collection[int]) – Status code or codes treated as healthy, (200,) by default.
headers (Mapping[str, str] | None) – Extra request headers. Not sent to a different origin after a redirect.
client (Any | Callable[[], Any] | None) – An existing client/session (or a callable returning one). If omitted, a temporary one is created per check with the probe timeout as its request timeout.
policy (HttpProbePolicy | None) – What the probe may request, see HttpProbePolicy. The default allows any http/https URL and does not follow redirects.
name (str | None) – The name of the probe. Defaults to the class name.
timeout (int | None) – The timeout for the probe.
success_ttl (int | timedelta | None) – Cache duration for successful results.
failed_ttl (int | timedelta | None) – Cache duration for failed results.
allow_failure (bool) – If True, a failure of this probe does not affect the overall result.

Raises:

ValueError – If url is refused by policy.

class HttpHttpx2Probe(url, *, method='GET', expected_status=(200,), headers=None, client=None, policy=None, name=None, timeout=None, success_ttl=None, failed_ttl=None, allow_failure=False)[source]

Bases: HttpProbeBase

HTTP check with httpx2.

Same behaviour as HttpHttpxProbe, built on httpx2.AsyncClient.

Initialize the probe.

Parameters:

url (str) – URL to request. Must be absolute and allowed by policy.
method (str) – HTTP method, GET by default.
expected_status (int | Collection[int]) – Status code or codes treated as healthy, (200,) by default.
headers (Mapping[str, str] | None) – Extra request headers. Not sent to a different origin after a redirect.
client (Any | Callable[[], Any] | None) – An existing client/session (or a callable returning one). If omitted, a temporary one is created per check with the probe timeout as its request timeout.
policy (HttpProbePolicy | None) – What the probe may request, see HttpProbePolicy. The default allows any http/https URL and does not follow redirects.
name (str | None) – The name of the probe. Defaults to the class name.
timeout (int | None) – The timeout for the probe.
success_ttl (int | timedelta | None) – Cache duration for successful results.
failed_ttl (int | timedelta | None) – Cache duration for failed results.
allow_failure (bool) – If True, a failure of this probe does not affect the overall result.

Raises:

ValueError – If url is refused by policy.

class HttpAiohttpProbe(url, *, method='GET', expected_status=(200,), headers=None, client=None, policy=None, name=None, timeout=None, success_ttl=None, failed_ttl=None, allow_failure=False)[source]

Bases: HttpProbeBase

HTTP check with aiohttp.

Sends a request with an existing aiohttp.ClientSession (or a callable returning one) or a temporary one, and fails unless the status code is in expected_status.

Initialize the probe.

Parameters:

url (str) – URL to request. Must be absolute and allowed by policy.
method (str) – HTTP method, GET by default.
expected_status (int | Collection[int]) – Status code or codes treated as healthy, (200,) by default.
headers (Mapping[str, str] | None) – Extra request headers. Not sent to a different origin after a redirect.
client (Any | Callable[[], Any] | None) – An existing client/session (or a callable returning one). If omitted, a temporary one is created per check with the probe timeout as its request timeout.
policy (HttpProbePolicy | None) – What the probe may request, see HttpProbePolicy. The default allows any http/https URL and does not follow redirects.
name (str | None) – The name of the probe. Defaults to the class name.
timeout (int | None) – The timeout for the probe.
success_ttl (int | timedelta | None) – Cache duration for successful results.
failed_ttl (int | timedelta | None) – Cache duration for failed results.
allow_failure (bool) – If True, a failure of this probe does not affect the overall result.

Raises:

ValueError – If url is refused by policy.

class KafkaAiokafkaProbe(client=None, *, bootstrap_servers=None, name=None, timeout=None, success_ttl=None, failed_ttl=None, allow_failure=False)[source]

Bases: ClientProbeBase

Check Kafka availability with aiokafka.

Fetches cluster metadata through an existing, already started AIOKafkaProducer, AIOKafkaConsumer or AIOKafkaAdminClient (or a callable returning one), or through a short-lived admin client connected to bootstrap_servers.

Initialize the probe.

Parameters:

client (Any | Callable[[], Any] | None) – Started AIOKafkaProducer/AIOKafkaConsumer/AIOKafkaAdminClient or a callable returning one.
bootstrap_servers (str | list[str] | None) – host:port or a list of them. Mutually exclusive with client.
name (str | None) – The name of the probe. Defaults to the class name.
timeout (int | None) – The timeout for the probe.
success_ttl (int | timedelta | None) – Cache duration for successful results.
failed_ttl (int | timedelta | None) – Cache duration for failed results.
allow_failure (bool) – If True, a failure of this probe does not affect the overall result.

class RabbitmqAiopikaProbe(client=None, *, url=None, name=None, timeout=None, success_ttl=None, failed_ttl=None, allow_failure=False)[source]

Bases: ClientProbeBase

Check RabbitMQ availability with aio-pika.

Opens and closes a channel on an existing connection (or a callable returning one), or on a short-lived connection established from url. A closed existing connection is a failure.

Initialize the probe.

Parameters:

client (Any | Callable[[], Any] | None) – aio_pika connection (AbstractConnection) or a callable returning one.
url (str | None) – AMQP URL, e.g. amqp://guest:guest@localhost/. Mutually exclusive with client.
name (str | None) – The name of the probe. Defaults to the class name.
timeout (int | None) – The timeout for the probe.
success_ttl (int | timedelta | None) – Cache duration for successful results.
failed_ttl (int | timedelta | None) – Cache duration for failed results.
allow_failure (bool) – If True, a failure of this probe does not affect the overall result.

class MongoProbeBase(client=None, *, url=None, name=None, timeout=None, success_ttl=None, failed_ttl=None, allow_failure=False)[source]

Bases: ClientProbeBase

Shared logic for MongoDB probes: run the ping admin command.

Subclasses adapt a concrete driver by implementing _new_client() and _close_client().

Initialize the probe.

Parameters:

client (Any | Callable[[], Any] | None) – A MongoDB client or a zero-argument callable returning one.
url (str | None) – MongoDB URI, e.g. mongodb://localhost:27017. Mutually exclusive with client.
name (str | None) – The name of the probe. Defaults to the class name.
timeout (int | None) – The timeout for the probe. For the url mode it is also passed to the driver as serverSelectionTimeoutMS so that the failure is reported by the driver, not as a bare timeout.
success_ttl (int | timedelta | None) – Cache duration for successful results.
failed_ttl (int | timedelta | None) – Cache duration for failed results.
allow_failure (bool) – If True, a failure of this probe does not affect the overall result.

class MongoPymongoProbe(client=None, *, url=None, name=None, timeout=None, success_ttl=None, failed_ttl=None, allow_failure=False)[source]

Bases: MongoProbeBase

Check MongoDB availability with the native async client of PyMongo.

Uses pymongo.AsyncMongoClient of PyMongo (PyMongo >= 4.9). Runs the ping admin command on an existing client (or a callable returning one), or on a short-lived client created from url.

Initialize the probe.

Parameters:

client (Any | Callable[[], Any] | None) – A MongoDB client or a zero-argument callable returning one.
url (str | None) – MongoDB URI, e.g. mongodb://localhost:27017. Mutually exclusive with client.
name (str | None) – The name of the probe. Defaults to the class name.
timeout (int | None) – The timeout for the probe. For the url mode it is also passed to the driver as serverSelectionTimeoutMS so that the failure is reported by the driver, not as a bare timeout.
success_ttl (int | timedelta | None) – Cache duration for successful results.
failed_ttl (int | timedelta | None) – Cache duration for failed results.
allow_failure (bool) – If True, a failure of this probe does not affect the overall result.

class MongoMotorProbe(client=None, *, url=None, name=None, timeout=None, success_ttl=None, failed_ttl=None, allow_failure=False)[source]

Bases: MongoProbeBase

Check MongoDB availability with Motor (AsyncIOMotorClient).

Runs the ping admin command on an existing client (or a callable returning one), or on a short-lived client created from url.

Initialize the probe.

Parameters:

client (Any | Callable[[], Any] | None) – A MongoDB client or a zero-argument callable returning one.
url (str | None) – MongoDB URI, e.g. mongodb://localhost:27017. Mutually exclusive with client.
name (str | None) – The name of the probe. Defaults to the class name.
timeout (int | None) – The timeout for the probe. For the url mode it is also passed to the driver as serverSelectionTimeoutMS so that the failure is reported by the driver, not as a bare timeout.
success_ttl (int | timedelta | None) – Cache duration for successful results.
failed_ttl (int | timedelta | None) – Cache duration for failed results.
allow_failure (bool) – If True, a failure of this probe does not affect the overall result.

Serving the result

Importable from the probirka package, e.g. from probirka import make_fastapi_endpoint.

Generic ASGI app

Scope

ASGI connection scope.

alias of MutableMapping[str, Any]

Message

A single ASGI protocol message.

alias of MutableMapping[str, Any]

Receive

Awaitable returning the next message from the server.

alias of Callable[[], Awaitable[MutableMapping[str, Any]]]

Send

Coroutine function sending a message to the server.

alias of Callable[[MutableMapping[str, Any]], Awaitable[None]]

ASGIApp

An ASGI 3 application.

alias of Callable[[MutableMapping[str, Any], Callable[[], Awaitable[MutableMapping[str, Any]]], Callable[[MutableMapping[str, Any]], Awaitable[None]]], Awaitable[None]]

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

Create a framework-independent ASGI application for a given Probirka instance.

The application is a plain ASGI 3 callable with no third-party dependencies. Register it as a route (Route('/health', app, methods=['GET', 'HEAD']) in Starlette), mount it (app.mount('/health', health_app) in FastAPI, asgi('/health', is_mount=True) in Litestar), or serve it on its own port with uvicorn module:health_app.

Two behaviours differ from the framework-specific adapters, because a mounted application gets no help from the host router:

The request path is ignored. Whatever reaches the application is answered, so an app mounted at /health also answers /health/anything.
Only GET and HEAD are served; every other method gets 405 with an allow header. HEAD returns the headers a GET would have returned, with an empty body.

Parameters:

probirka (Probirka) – The Probirka instance to run.
timeout (int | None) – The timeout for the Probirka run.
with_groups (str | Sequence[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 ASGI application.

Return type:

Callable[[MutableMapping[str, Any], Callable[[], Awaitable[MutableMapping[str, Any]]], Callable[[MutableMapping[str, Any]], Awaitable[None]]], Awaitable[None]]

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 (int | None) – The timeout for the Probirka run.
with_groups (str | Sequence[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 (int | None) – The timeout for the Probirka run.
with_groups (str | Sequence[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[[Request], Coroutine[Any, Any, Response]]

Django

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

Create a Django async view for a given Probirka instance.

The view accepts GET (and HEAD) requests only; other methods get 405. The JSON body is ProbirkaResult.to_dict(), the same format as the other integrations.

Parameters:

probirka (Probirka) – The Probirka instance to run.
timeout (int | None) – The timeout for the Probirka run.
with_groups (str | Sequence[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 Django async view.

Return type:

Callable[[HttpRequest], Coroutine[Any, Any, HttpResponse]]