observallize 0.1.1

Agnostic observability (spans + custom metrics) for RPA executions

Observallize

Instrumentação agnóstica para execuções (principalmente RPAs) baseada em spans (eventos), com:

• medição de tempo e árvore pai/filho (parent_span_id)
• memória RSS (início/fim/delta e pico aproximado por amostragem)
• métricas customizadas durante a execução (obs.add_metric(...))
• captura de exceções por span (exception_type/message/stacktrace)

Este README (raiz) é propositalmente direto e focado em uso. A documentação detalhada por módulo fica em observallize/README.md (link será apontado no GitHub).

Instalação

pip install observallize

Organização de imports (projetos com muitos módulos)

Você pode usar a instância pronta obs, mas o padrão recomendado é criar sua própria instância de Observability por execução/processo, para ter um correlation_id estável do início ao fim do .py (mais simples do que depender só de contexto).

Para respeitar SOLID e evitar criar instâncias em vários lugares, um padrão comum é criar/reexportar uma única instância de Observability a partir de um módulo do seu projeto (ex.: src/infra/observability.py) e importar esse ponto único em todos os Services/Controllers.

Exemplo completo (modelo Controller/Service + executor)

O exemplo abaixo segue o mesmo padrão do main.py.

• O executor define o execution_id uma vez para o processo inteiro
• Cada controller reutiliza esse execution_id para associar tudo ao mesmo “job”
• O executor envolve o handler com use_exporter e persiste no finally
• O handler é observado para registrar “métricas macro” do controller

from pathlib import Path
from time import sleep
from typing import Any

from observallize.retry import retry
from observallize import Observability

obs = Observability(Path("events.json"))


class Repository:
    @staticmethod
    def insert_batch(events: list[dict[str, Any]]):
        print(f"Inserindo batch no DB: {len(events)} eventos")


class Service:
    @staticmethod
    @obs.observe(name="login")
    def login(user: str, password: str):
        sleep(2)
        obs.add_metric("auth_provider", "site_x")
        obs.add_metric("attempt", 1)
        return "A", 30, {"result": "...", "status": True}

    @staticmethod
    @obs.observe(name="baixar_relatorio")
    def baixar_relatorio(dia: int):
        return {"dia": dia, "rows": 123}

    @staticmethod
    @obs.observe(name="processamento_do_pedido")
    def processar_pedido(algo: str, outro: str, alguma_coisa: int):
        obs.add_metric("parametro", outro)
        return True


class DataService:
    @staticmethod
    @obs.observe(name="tratando_dados")
    def treat_records(): ...


class Controller:
    def __init__(self, service: Service, data: DataService, db: Repository):
        self.correlation_id = obs.correlation_id
        self.export = obs.create_exporter(also_queue=False)
        self.service = service
        self.data = data
        self.db = db

    @obs.observe(name="controller.handler", capture_io=False, capture_result=False)
    def handler(self):
        obs.add_metric("controller_name", self.__class__.__name__)
        obs.add_metric("rpa_name", "meu_rpa")
        obs.add_metric("batch_id", "2026-04-11_001")
        obs.add_metric("ambiente", "prod")

        self.service.login("usuario", "senha")
        self.service.baixar_relatorio(15)
        self.service.processar_pedido("algo", "outro", 100)


@retry(3)
def executor(controllers, execution_id: str | None = None):
    obs.set_execution_context(execution_id)
    obs.activate()
    for controller_cls, params in controllers:
        controller = controller_cls(**params)
        with obs.use_exporter(controller.export):
            try:
                controller.handler()
            finally:
                events = obs.span_store.get(controller.correlation_id)
                rows = obs.to_rows(events, include_io=False, include_result=False)
                controller.db.insert_batch(rows)
                obs.span_store.clear(controller.correlation_id)


def run():
    controllers = [
        (Controller, {"service": Service, "data": DataService, "db": Repository}),
    ]
    executor(controllers)

Persistência (modos)

• Store + rows (normalizado): create_exporter(also_queue=False) → events = obs.span_store.get(execution_id) → rows = obs.to_rows(events) → insert_batch(rows)
• Fila + flush (eventos crus): create_exporter(also_queue=True) → obs.flush(insert_batch)

Inspeção (durante desenvolvimento)

events = obs.span_store.get(obs.get_execution_id())
print(obs.to_json(events, include_io=False, include_result=False, datetime_format="%d/%m/%Y %H:%M:%S"))

Boas práticas (para não perder spans)

• Envolva a chamada da função decorada com with obs.use_exporter(...) (ex.: controller.handler()).
• Evite setar exporter “por dentro” do handler decorado.
• Deixe o try/except no executor e faça a persistência no finally.

Erros: o que acontece com try/finally

• Se controller.handler() levantar uma exceção, ela sobe normalmente (não é “engolida”).
• O bloco finally sempre roda antes da exceção continuar subindo, então você persiste os eventos e depois a execução aborta (a menos que você capture e trate).
• Se você estiver usando @retry(...), a exceção dispara nova tentativa até esgotar attempts; na última falha, a exceção é propagada.

Documentação detalhada

• Detalhes por módulo: README