Metadata-Version: 2.5
Name: probirka
Version: 0.9.0
Summary: Framework-agnostic library for running health probes in Python applications, with built-in probes and integrations for popular HTTP frameworks.
Project-URL: Repository, https://github.com/appKODE/probirka
Author-email: KODE <slurm@kode.ru>
License-Expression: MIT
License-File: LICENSE
Keywords: health-check,monitoring,probes
Classifier: Development Status :: 5 - Production/Stable
Classifier: Framework :: AsyncIO
Classifier: Framework :: Django
Classifier: Framework :: FastAPI
Classifier: Framework :: aiohttp
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Programming Language :: Python :: 3.15
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: System :: Monitoring
Classifier: Typing :: Typed
Requires-Python: >=3.11
Description-Content-Type: text/markdown

# PROB🧪RKA

[![PyPI](https://img.shields.io/pypi/v/probirka.svg)](https://pypi.python.org/pypi/probirka)
[![Python](https://img.shields.io/pypi/pyversions/probirka.svg)](https://pypi.python.org/pypi/probirka)
[![PyPI](https://img.shields.io/pypi/dm/probirka.svg)](https://pypi.python.org/pypi/probirka)
[![Coverage Status](https://coveralls.io/repos/github/appKODE/probirka/badge.svg?branch=main)](https://coveralls.io/github/appKODE/probirka?branch=main)
[![Ruff](https://img.shields.io/endpoint?url=https://raw.githubusercontent.com/astral-sh/ruff/main/assets/badge/v2.json)](https://github.com/astral-sh/ruff)

Framework-agnostic library for running health probes in Python applications.

A small async engine that runs checks concurrently, handles timeouts, caches results and exposes them over HTTP, with ready-made probes for common infrastructure such as PostgreSQL, Redis, HTTP, Kafka, RabbitMQ and MongoDB.

* 🚫 no runtime dependencies in the core
* ⚡ concurrent async execution
* ⏱ per-probe and overall timeouts
* 💾 separate caching of successful and failed results
* 🧩 optional probe groups — cheap liveness, expensive readiness
* 🟡 `allow_failure` — non-critical probes that may fail without failing the check
* 🔌 ready-made probes for common infrastructure
* 🌐 FastAPI, aiohttp and Django adapters, plus a dependency-free ASGI app for everything else
* 🐍 fully typed

[Documentation](https://appkode.github.io/probirka/) · [Changelog](CHANGELOG.md)

## Installation

```bash
pip install probirka
```

Requires Python 3.11 or newer. The core has no runtime dependencies.

Ready-made probes and framework integrations use client libraries that you install alongside. Install only what you need:

```bash
pip install probirka asyncpg redis fastapi
```

If a library is missing, accessing the corresponding probe or adapter raises an `ImportError` that names the package to install:

```text
ImportError: RedisProbe requires the 'redis' package, install it with: pip install redis
```

See the [documentation](https://appkode.github.io/probirka/probes.html) for the complete list of probes and their dependencies.

## Quick start

The simplest way to add a probe is a function:

```python
import asyncio

from probirka import Probirka


probirka = Probirka(success_ttl=30)


@probirka.add(name='database', timeout=2)
async def check_database():
    return True


@probirka.add(name='external_api', timeout=2)
async def check_external_api():
    return True


async def main():
    result = await probirka.run()

    print(result.ok)
    print(result.to_dict())


asyncio.run(main())
```

Probes are executed concurrently. Synchronous functions are supported too — they run in the event loop's default executor, so they neither block the loop nor escape their timeout.

## How it works

A **probe** is a single check of some application dependency or subsystem. `Probirka` runs probes and aggregates their results into a single health-check result.

```text
                 Probirka
                    │
        ┌───────────┼───────────┐
        ▼           ▼           ▼
    PostgreSQL     Redis       HTTP
      probe        probe       probe
        │           │           │
        └───────────┼───────────┘
                    ▼
             ProbirkaResult
```

Probes are independent and can be implemented as simple functions or as reusable `Probe` classes.

A probe never crashes the health check. Whatever it does is turned into a `ProbeResult`:

| The probe                  | `ok`    | `error`                                        |
| -------------------------- | ------- | ---------------------------------------------- |
| returns `True` or `None`   | `True`  | `None`                                         |
| returns `False`            | `False` | `None`                                         |
| raises an exception        | `False` | `'ExcType: message'`                           |
| exceeds its `timeout`      | `False` | `'TimeoutError: probe timed out after 2s'`     |

A failing check therefore looks like this:

```python
{
    'ok': False,
    'started_at': '2026-09-12T16:00:00.000000+03:00',
    'elapsed': 2.001,
    'info': None,
    'checks': [
        {
            'name': 'database',
            'ok': False,
            'cached': False,
            'started_at': '2026-09-12T16:00:00.000000+03:00',
            'elapsed': 2.0,
            'info': None,
            'error': 'TimeoutError: probe timed out after 2s',
            'allow_failure': False,
        },
        ...,
    ],
    'error': None,
}
```

The top-level `ok` is `True` when every probe that ran passed, except probes marked with `allow_failure` — those may fail without affecting it, see [Allowing failures](#allowing-failures).

## Ready-made probes

| Probe                  | Dependency | What it checks                                             |
| ---------------------- | ---------- | ---------------------------------------------------------- |
| `TcpProbe`             | —          | A TCP connection to `host:port` can be established         |
| `PostgresAsyncpgProbe` | `asyncpg`  | `SELECT 1` on a pool, connection or `dsn`                  |
| `RedisProbe`           | `redis`    | `PING` via `redis.asyncio`                                 |
| `HttpHttpxProbe`       | `httpx`    | Response status code of a request                          |
| `HttpHttpx2Probe`      | `httpx2`   | Same, with `httpx2`                                        |
| `HttpAiohttpProbe`     | `aiohttp`  | Same, with `aiohttp.ClientSession`                         |
| `KafkaAiokafkaProbe`   | `aiokafka` | Cluster metadata via a producer, consumer or admin client  |
| `RabbitmqAiopikaProbe` | `aio-pika` | A channel can be opened and closed                         |
| `MongoPymongoProbe`    | `pymongo`  | `ping` command via `pymongo.AsyncMongoClient`              |
| `MongoMotorProbe`      | `motor`    | `ping` command via `AsyncIOMotorClient`                    |

For example, checking application dependencies:

```python
from probirka import (
    Probirka,
    PostgresAsyncpgProbe,
    RedisProbe,
    HttpHttpxProbe,
)

probirka = Probirka()

probirka.add_probes(
    PostgresAsyncpgProbe(
        lambda: app.state.pool,
        name='postgres',
        timeout=2,
    ),
    RedisProbe(
        url='redis://cache:6379/0',
        name='redis',
        timeout=1,
    ),
    HttpHttpxProbe(
        'https://example.com/health',
        name='external-api',
        timeout=2,
    ),
)
```

Every probe that talks to a service takes either an existing client or a connection string — exactly one of the two:

* **an existing client** (first positional argument): the probe reuses your pool and its configuration and never closes it;
* **a connection string** (`dsn`, `url`, `bootstrap_servers`): the probe opens a short-lived connection for each check and closes it afterwards.

When the client is created after the probes are registered — in a FastAPI lifespan, an aiohttp startup signal — pass a zero-argument function instead of the client, as in the `lambda: app.state.pool` above. It is called on every check. This makes it possible to reuse application connection pools instead of creating additional connections just for health checks.

A probe that reaches the service but does not like the answer raises `ProbeFailure`, which lands in `error` as, for example, `'ProbeFailure: unexpected status 503'`.

The HTTP probes also take a `policy` that limits which URLs they request and which redirects they follow, see [Security policy](#security-policy).

## Custom probes

Ready-made probes are just regular `Probe` implementations. For application-specific checks, use a function:

```python
@probirka.add(name='application')
async def check_application():
    return await something_is_working()
```

Or create a reusable probe:

```python
from probirka import ProbeBase


class MyProbe(ProbeBase):
    async def _check(self) -> bool:
        return await check_something()
```

There is no special API for custom probes. They use the same execution, timeout, caching and result handling as built-in probes.

## Metadata

Both the health check and individual probes can carry arbitrary metadata, which ends up in the `info` field of the result:

```python
probirka.add_info('version', '1.4.0')
probirka.add_info('environment', 'production')


class DatabaseProbe(ProbeBase):
    async def _check(self) -> bool:
        pool = app.state.pool
        self.add_info('pool_size', pool.get_size())
        return True
```

## Secrets

Health endpoints are rarely behind authentication, and the `error` field carries whatever message the client library produced — which for a wrong password or an unreachable host is often the whole connection string. Probirka masks secrets before they reach the response:

* **Known values.** Every ready-made probe registers the password of its connection string (as written, URL-decoded and as the `Basic` credentials HTTP clients derive from it) and the values of its request headers. Wherever they show up in an exception message, `ProbeResult.error` has `'***'` instead: `'InvalidPasswordError: ... postgresql://app:***@db/app'`. A custom probe can do the same with `self._register_secrets(value, ...)`.
* **Shape-based.** `to_dict()` walks `info` and masks values under keys that look sensitive (`password`, `api_key`, `Authorization`, `secret_ttl`, ... — the same heuristic Django uses for its error pages), and passwords in URLs and query parameters like `?token=` in every string, `error` included. Objects that are not JSON types are written as their `str()`, masked the same way.

The user name, host, port and database are kept, as they are what makes the message useful. Masking is on by default, so all HTTP integrations return masked output; `result.to_dict(redact=False)` gives the raw data for your own handler. The helpers are available too: `probirka.mask_url('redis://:s3cret@cache/0')` returns `'redis://:***@cache/0'`, `probirka.redact_value(mapping)` cleans a dictionary.

Masking is a safety net, not a substitute for care: only what a probe knows about or what looks like a secret is caught, so do not put credentials into `info` in the first place.

## Security policy

An HTTP probe requests whatever URL it was configured with. When that URL comes from a config file or an admin UI, or when the target answers with a redirect, the probe can be pointed at an internal service or at a cloud metadata endpoint such as `http://169.254.169.254/` — the classic server-side request forgery. `HttpProbePolicy` limits what the HTTP probes will contact:

```python
from probirka import HttpHttpxProbe, HttpProbePolicy

policy = HttpProbePolicy(
    allowed_hosts=('api.example.com', '*.example.com'),
    block_private_networks=True,
    allowed_networks=('10.20.0.0/16',),  # our own service mesh
    follow_redirects=True,
    max_redirects=3,
)

probirka.add_probes(
    HttpHttpxProbe('https://api.example.com/health', policy=policy, timeout=3),
)
```

The defaults are deliberately soft, because health checks usually target internal services: any `http` or `https` URL is accepted, no network is blocked, and redirects are not followed — a `302` is an ordinary status code compared with `expected_status`. Everything stricter is opt-in:

* `allowed_schemes` — `('http', 'https')` by default; `('https',)` also refuses a redirect that downgrades to plain HTTP.
* `allowed_hosts` — host names the probe may request, exactly or as `*.example.com` for any subdomain (not `example.com` itself). `None`, the default, allows any host.
* `block_private_networks` — refuses every address that is not globally routable: loopback, the RFC 1918 and `fc00::/7` private ranges, link-local including `169.254.169.254`, carrier-grade NAT, multicast, unspecified and reserved ranges.
* `blocked_networks` and `allowed_networks` — your own CIDR lists. `allowed_networks` wins, so "no private networks except our mesh" is `block_private_networks=True, allowed_networks=('10.20.0.0/16',)`.
* `follow_redirects` and `max_redirects` — see below.

The URL is checked when the probe is created; a refused URL is a `ValueError`, so a misconfiguration surfaces at startup rather than in the health output. Before every request the URL is checked again, and when a network rule is set the host name is resolved first: the request is refused if *any* resolved address is blocked, reported as `'HttpProbePolicyViolation: internal.svc resolves to 10.0.0.5, which is not a global address'`. `HttpProbePolicyViolation` is a `ProbeFailure`, so `allow_failure` applies as usual.

Redirects are never left to the client library. Every request is sent with automatic following disabled, and with `follow_redirects=True` the probe follows `301`, `302`, `303`, `307` and `308` itself: each target is checked against the policy before it is requested, `POST` becomes `GET` on `301`, `302` and `303` as browsers do, and the request `headers` are dropped when the target is another origin, so an `Authorization` header meant for your API never reaches a third party. More than `max_redirects` hops is a failure. This is the same for httpx, httpx2 and aiohttp, which means `HttpAiohttpProbe` no longer follows redirects unless asked to.

Limitations: the probe resolves the name and then the client library resolves it again, so a resolver that answers differently the second time (DNS rebinding) can still reach a blocked address; a proxy configured on your own client is not inspected. The policy narrows what a probe does, it does not replace network-level controls.

## Groups

Some checks may be too expensive or too slow to run on every health request. Put them into an optional group:

```python
@probirka.add(
    name='external',
    groups=['external'],
)
async def check_external_service():
    return True
```

Probes without groups are required and run on every call. Grouped probes run only when their group is requested:

```python
await probirka.run(with_groups=['external'])
```

Pass `skip_required=True` to run the requested groups alone:

```python
await probirka.run(with_groups='external', skip_required=True)
```

This separates cheap liveness checks from expensive dependency checks — one `Probirka` instance can back both Kubernetes endpoints:

```python
probirka = Probirka()


@probirka.add(name='self')
async def liveness():
    return True


probirka.add_probes(
    PostgresAsyncpgProbe(lambda: app.state.pool, name='postgres', timeout=2),
    RedisProbe(lambda: app.state.redis, name='redis', timeout=1),
    groups='readiness',
)

app.add_api_route('/livez', make_fastapi_endpoint(probirka))
app.add_api_route('/readyz', make_fastapi_endpoint(probirka, with_groups='readiness'))
```

Groups that were never registered are ignored rather than reported as failures.

## Allowing failures

Not every dependency is critical. A cache or a metrics backend may be down while the service still serves traffic, and a health check that reports `503` in that case only causes restarts. Mark such probes with `allow_failure=True`, named after the same option in GitLab CI:

```python
@probirka.add(name='cache', allow_failure=True)
async def check_cache():
    return await redis.ping()


probirka.add_probes(
    TcpProbe('smtp.internal', 25, name='smtp', allow_failure=True),
)
```

A probe with `allow_failure` runs and is reported like any other: its `ok`, `error` and `elapsed` are in `checks`, and its result carries `allow_failure: true`. It just does not count towards the top-level `ok`, so the HTTP integrations return `success_code` even when it fails:

```python
{
    'ok': True,
    'checks': [
        {'name': 'database', 'ok': True, 'allow_failure': False, ...},
        {'name': 'cache', 'ok': False, 'allow_failure': True, 'error': 'ConnectionError: ...', ...},
    ],
    'error': None,
}
```

The flag can also be set for a whole [group](#groups). It then overrides the probes' own setting, in both directions:

```python
probirka.add_probes(
    HttpHttpxProbe('https://partner.example/health', name='partner'),
    TcpProbe('smtp.internal', 25, name='smtp'),
    groups='external',
    allow_failure=True,
)
```

Without `allow_failure` the group leaves every probe as it is. When a probe is run through several sources at once — the required list and a group, or two groups — it is allowed to fail only if every source allows it, so a strict source always wins. `add_probes(..., allow_failure=...)` without `groups` raises `ValueError`: for required probes set the flag on the probe itself.

The overall `run(timeout=...)` follows the same rule: if only probes with `allow_failure` did not finish in time, `ok` stays `True`, while `error` still reports the timeout.

## Caching

Health checks should not necessarily hit every dependency on every request. Probirka can cache successful and failed probe results independently:

```python
probirka = Probirka(
    success_ttl=30,
    failed_ttl=5,
)
```

These defaults apply to probes registered with the `@probirka.add(...)` decorator; a probe can override them, and `success_ttl=0` or `failed_ttl=0` disables that half of the cache for one probe. Probe objects passed to `add_probes()` — including all ready-made probes — carry their own TTLs instead, so set them on the probe:

```python
probirka.add_probes(
    RedisProbe(url='redis://cache:6379/0', success_ttl=30, failed_ttl=5),
)
```

TTLs are counted with a monotonic clock, so clock adjustments do not extend or shorten them.

Cached results are marked with `cached=True`, so consumers can distinguish a fresh check from a cached result. A probe with no TTL configured reports `cached=None`.

## Timeouts

Every probe can have its own timeout:

```python
@probirka.add(
    name='database',
    timeout=2,
)
async def check_database(): ...
```

There is also an overall timeout for the complete run:

```python
result = await probirka.run(timeout=5)
```

This prevents one slow dependency from keeping the whole health check request open indefinitely. `run()` never raises on the overall timeout: probes that did not finish in time are cancelled and reported as failed, finished ones keep their results, and the run itself gets `error` set to `'TimeoutError: probirka run timed out after 5s'`. `ok` becomes `False` unless every probe that did not finish has [`allow_failure`](#allowing-failures).

## HTTP integrations

Probirka provides thin adapters for popular Python HTTP frameworks, and a generic ASGI application for everything else. All of them run the probes, map `ok` to a status code and return `ProbirkaResult.to_dict()` as JSON, so every endpoint answers in the same format.

| Where it runs                            | Use                                     |
| ---------------------------------------- | --------------------------------------- |
| FastAPI                                  | `make_fastapi_endpoint`                 |
| aiohttp                                  | `make_aiohttp_endpoint`                 |
| Django                                   | `make_django_view`                      |
| Starlette, Litestar, Falcon, BlackSheep  | `make_asgi_app`                         |
| A port of its own                        | `make_asgi_app` served by uvicorn       |
| Sanic, Quart, Flask                      | `probirka.run()` from your own handler  |

Every factory takes the same keyword arguments:

| Argument         | Default | Meaning                                                |
| ---------------- | ------- | ------------------------------------------------------ |
| `timeout`        | `None`  | Overall timeout for the run, in seconds                |
| `with_groups`    | `''`    | Optional groups to include                             |
| `skip_required`  | `False` | Run only the requested groups                          |
| `return_results` | `True`  | Return the result as JSON; `False` sends an empty body |
| `success_code`   | `200`   | Status code when every probe passed                    |
| `error_code`     | `500`   | Status code when at least one probe failed             |

### FastAPI

```python
from fastapi import FastAPI

from probirka import Probirka, make_fastapi_endpoint

app = FastAPI()

probirka = Probirka()

# add probes...

app.add_api_route(
    '/health',
    make_fastapi_endpoint(probirka),
)
```

A complete example with liveness, readiness and startup probes and an authenticated report is in the [documentation](https://appkode.github.io/probirka/integration.html#kubernetes-probes).

### aiohttp

```python
from aiohttp import web

from probirka import Probirka, make_aiohttp_endpoint

app = web.Application()

probirka = Probirka()

# add probes...

app.router.add_get(
    '/health',
    make_aiohttp_endpoint(probirka),
)
```

### Django

```python
# urls.py
from django.urls import path

from probirka import Probirka, make_django_view

probirka = Probirka()

# add probes...

urlpatterns = [
    path('health', make_django_view(probirka)),
]
```

The Django view answers `GET` and `HEAD`; other methods get `405 Method Not Allowed`.

### Any ASGI framework

`make_asgi_app` returns a plain ASGI 3 application. It is the only integration with no third-party dependency at all, so it works on a bare `pip install probirka`:

```python
from starlette.applications import Starlette
from starlette.routing import Route

from probirka import Probirka, make_asgi_app

probirka = Probirka()

# add probes...

app = Starlette(
    routes=[
        Route('/health', make_asgi_app(probirka), methods=['GET', 'HEAD']),
    ],
)
```

Mounting works too — `app.mount('/health', make_asgi_app(probirka))` in FastAPI, `asgi('/health', is_mount=True)(make_asgi_app(probirka))` in Litestar — and so does serving it separately:

```bash
uvicorn health:health_app --port 8081
```

Because a mounted app gets no help from the host router, it handles two things itself: the request path is ignored, and only `GET` and `HEAD` are served, everything else gets `405`.

> [!WARNING]
> Starlette's `Mount('/health', app)` answers on `/health/` and redirects the bare `/health` with a `307`. Kubernetes counts any status below 400 as success, so such a probe would report the service healthy without ever running the checks. Use `Route`, or point the probe at the trailing slash.

## Results

Results are typed immutable data classes. `started_at` is timezone-aware in the local zone of the host, and `elapsed` is measured with a monotonic clock, so it is unaffected by clock adjustments:

```python
result = await probirka.run()

print(result.ok)

for check in result.checks:
    print(
        check.name,
        check.ok,
        check.cached,
        check.elapsed,
        check.error,
        check.allow_failure,
    )
```

`result.to_dict()` converts it to a JSON-compatible dictionary, which is what every HTTP integration returns and what monitoring systems consume. Secrets in `info` and `error` are masked on the way, see [Secrets](#secrets); `to_dict(redact=False)` skips that.

## Why Probirka?

Probirka deliberately keeps the core small. It does not try to manage connections, discover services or prescribe how your application should be structured. Instead, it provides three things:

1. **Probe implementations** — checks for your dependencies.
2. **Execution engine** — concurrency, timeouts, caching and groups.
3. **Integrations** — expose the same result through your HTTP framework, a mountable ASGI app, or a handler of your own.

Your application remains responsible for creating and managing its clients, pools and connections.

## Development

The project uses [uv](https://docs.astral.sh/uv/) and [just](https://github.com/casey/just):

```bash
uv sync --dev
just tests
just lint
just ty
just doc
```

`just lint` and `just fmt` cover `probirka`, `tests` and `docs`. The architecture boundaries, the rule that `__init__.py` files only re-export, and the commit and merge request conventions are in [AGENTS.md](AGENTS.md).

Integration tests run the ready-made probes against real services in Docker. They are skipped unless `PROBIRKA_INTEGRATION=1` is set, so `just tests` needs no Docker:

```bash
just services-up          # docker compose -f tests/integration/compose.yaml up -d --wait
just tests-integration    # PROBIRKA_INTEGRATION=1 pytest -m integration tests/integration/
just services-down
```

Service addresses default to the ports published by the compose file; override them with `PROBIRKA_IT_POSTGRES_DSN`, `PROBIRKA_IT_REDIS_URL`, `PROBIRKA_IT_RABBITMQ_URL`, `PROBIRKA_IT_MONGO_URL`, `PROBIRKA_IT_KAFKA_BOOTSTRAP` and `PROBIRKA_IT_HTTP_URL`. In CI the same containers are started by the `services:` block of the `integration` job.

## License

MIT
