Basic Tutorial

Defining an ‘instrument’

Any object can be used as an ‘instrument’. Including objects from other libraries and custom objects.

 1import requests
 2
 3
 4class MetricsAdapter:
 5
 6    def __init__(self, uri: str) -> None:
 7        self.uri = uri
 8
 9    def increment(self, metric_name: str, increment_amount: int = 1) -> None:
10        requests.post(self.uri, {
11            "metric_name": metric_name,
12            "action": {
13                "type": "increment",
14                "amount": increment_amount
15            }
16        })

Defining an observation

The best way to define an observation is by creating a class that inherits from BaseObservation. Strictly, any class that implements the ObservationProtocol is supported, however this isn’t recommended.

announcement() decorators can also be stacked. See the component docs for more details.

Note

When the required parameter in an announcement() decorator is set to True, an exception will be raised upon no implementation of the same instrument type existing in the probe used to make observations. required parameter is set to False by default.

 1import logging
 2
 3from domprob import sensor, BaseObservation
 4
 5
 6class CheckoutSuccessful(BaseObservation):
 7
 8    def __init__(self, **order_details: Any) -> None:
 9        self.order_details = order_details
10
11    @staticmethod
12    @sensor(instrum=logging.Logger, required=True)
13    def log_checkout_successful(log: logging.Logger) -> None:
14        log.info("Checkout successful!", **self.order_details)
15
16    @staticmethod
17    @sensor(instrum=MetricsAdapter)
18    def increment_successful_checkouts(metric_app: MetricsAdapter) -> None:
19        metric_app.increment('successful-checkouts', 1)

Calling an observation

With the default probe:

The default probe has a single, default logger instance available. This allows you to start using the library easily; configuring a custom probe is recommended before use in production systems.

 1from domprob import probe
 2
 3class OrderService:
 4
 5    def checkout(self):
 6        try:
 7            self.checkout_service.checkout_order(self.order)
 8        except CheckoutError as e:
 9            raise
10        probe.observe(CheckoutSuccessfulObservation())

With a custom probe:

Configuring a custom probe with user-defined instruments is easy with get_probe(). For more control over customisation, any user-defined class that implements DispatcherProtocol can be directly passed into the Probe.

 1from domprob import get_probe
 2
 3class OrderService:
 4
 5    def __init__(self):
 6        self.probe = get_probe(logging.getLogger(), MetricsAdapter("<api_endpoint>"))
 7
 8    def checkout(self):
 9        try:
10            self.checkout_service.checkout_order(self.order)
11        except CheckoutError as e:
12            raise
13        self.probe.observe(CheckoutSuccessful(**self.order_aggregate))