domprob 🛰️

Observability Domain Probes Python Package

📄 Documentation | 🐍 PyPI

 

domprob 🛰️

Implement the domain probe pattern with observability events easily with domprob.

Motivation

Observability often focuses on low-level signals, missing domain-specific insights that explain why a system behaves a certain way. Inspired by this blog post titled “Domain-Oriented Observability”, domprob helps expose meaningful domain events and causal relationships, enabling teams to reason about their system in business terms.

Key Features

Tidy Domain Logic: Keeps domain logic clean by separating observability concerns, ensuring insights don’t clutter core business code.

Turn this (20 lines):

class OrderService:
    def checkout(self):
        self.logger.log(f"Attempting to checkout order {self.order}")
        try:
            self.checkout_service.checkout_order(self.order)
        except CheckoutError as e:
            self.logger.error(f"Checkout for order {self.order} failed: {e}")
            self.metrics.increment("checkout-failed", {
                "failed_orders": 1, "customer": 6234654
            })
            return
        self.metrics.increment("checkout-successful", {
            "successful_orders": 1,
         })
        self.logger.log(f"Order checkout completed successfully", {
            "successful_orders": 1,
            "customer": 6234654,
            "order_number": 2374,
            "sku": "JH-374-VJHV"
        })

→ Into ✨this✨ (9 lines):

class Order:
    def checkout(self):
        probe.observe(AttemptingCheckoutObservation())
        try:
            self.checkout_service.checkout_order(self.order)
        except CheckoutError as e:
            probe.observe(CheckoutFailedObservation())
            return
        probe.observe(CheckoutSuccessfulObservation())

Installation

uv

uv add domprob

poetry

poetry add domprob

pip

pip install domprob

Usage

Define an observation:

import logging
from typing import Any

from domprob import sensor, BaseObservation


class CheckoutSuccessful(BaseObservation):

    def __init__(self, **order_details: Any) -> None:
        self.order_details = order_details

    @sensor(instrum=logging.Logger)
    def log_observation(self, log: logging.Logger) -> None:
        log.info("Checkout successful!", **self.order_details)

Calling the observation:

import logging
from domprob import get_probe

probe = get_probe(logging.getLogger(__name__))


class OrderService:

    def checkout(self):
        try:
            self.checkout_service.checkout_order(self.order)
        except CheckoutError as e:
            raise
        probe.observe(CheckoutSuccessful(**self.order_entity))

Check out the docs for more detailed examples!

Contributing

• Our code of conduct is here.

• Check out all our issues here. We’ve compiled a least of good first issues here.