domprob πŸ›°οΈοƒ

Inspired by this blog post, domprob is a Python package to implement observability domain probes in your projects.

PyPI - Package Version PyPI - Python Version codecov pre-commit Checked with mypy Code style: black

πŸ“„ Documentation | 🐍 PyPI

Overview

Keep your business logic comprehensible by abstracting the observability code away.

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

Install using uv:

uv add domprob

Using pip:

pip install domprob

Using poetry:

poetry add domprob

Usage

Define an observation

import logging
from typing import Any

from domprob import announcement, BaseObservation


class CheckoutSuccessful(BaseObservation):

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

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

Calling the observation

   from domprob import probe

   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 detail!

Getting Started

Components