apifrom.monitoring

Monitoring and metrics collection module for APIFromAnything.

This module provides tools for monitoring API performance, collecting metrics, and integrating with popular monitoring systems like Prometheus, Grafana, and others.

Overview

Classes

Classes

JSONExporter(collector = None):bases: MetricsExporter

Exports metrics data in JSON format.

Initialize the exporter with a metrics collector.

apifrom.monitoring.export(pretty=False)

Export metrics as JSON.

param pretty:

Whether to format the JSON with indentation for readability.

returns:

JSON string representation of the metrics.

apifrom.monitoring.export_to_file(file_path, pretty=True)

Export metrics to a JSON file.

param file_path:

Path to the output file.

param pretty:

Whether to format the JSON with indentation for readability.

LogExporter(collector = None, logger = None, level = logging.INFO):bases: MetricsExporter

Exports metrics to Python’s logging system.

Initialize the log exporter.

param collector:

Metrics collector to export from.

param logger:

Logger to use for logging metrics.

param level:

Logging level to use.

apifrom.monitoring.export(include_details=False, **kwargs)

Export metrics to logs.

param include_details:

Whether to include detailed metrics information.

apifrom.monitoring.export_periodic(interval_seconds=60, include_details=False)

Start a background thread to periodically export metrics to logs.

param interval_seconds:

Interval between exports in seconds.

param include_details:

Whether to include detailed metrics information.

class apifrom.monitoring.Metric[source]

Represents a single metric with its metadata and values.

apifrom.monitoring.__post_init__()

Initialize the metric based on its type.

apifrom.monitoring.get_histogram_buckets()

Get histogram bucket counts.

apifrom.monitoring.get_stats()

Get basic statistics for histogram and summary metrics.

apifrom.monitoring.get_summary_quantiles()

Get summary quantiles.

apifrom.monitoring.increment(amount=1.0, labels=None)

Increment a counter metric.

apifrom.monitoring.observe(value, labels=None)

Add an observation to a histogram or summary metric.

apifrom.monitoring.set(value, labels=None)

Set a gauge metric to a specific value.

MetricType:bases: enum.Enum

Types of metrics that can be collected.

class apifrom.monitoring.MetricsCollector[source]

Collects and manages metrics for the API.

This class provides a centralized way to create, update, and retrieve metrics for monitoring API performance and usage.

Initialize the metrics collector.

apifrom.monitoring.create_counter(name, description, labels=None)

Create a new counter metric.

apifrom.monitoring.create_gauge(name, description, labels=None)

Create a new gauge metric.

apifrom.monitoring.create_histogram(name, description, buckets=None, labels=None)

Create a new histogram metric.

apifrom.monitoring.create_metric(metric)

Register a new metric with the collector.

param metric:

The metric to register

returns:

The registered metric

apifrom.monitoring.create_summary(name, description, quantiles=None, labels=None)

Create a new summary metric.

apifrom.monitoring.get_all_metrics()

Get all metrics.

apifrom.monitoring.get_metric(name)

Get a metric by name.

apifrom.monitoring.increment(name, amount=1.0, labels=None)

Increment a counter metric.

apifrom.monitoring.observe(name, value, labels=None)

Add an observation to a histogram or summary metric.

apifrom.monitoring.reset()

Reset all metrics.

apifrom.monitoring.set(name, value, labels=None)

Set a gauge metric to a specific value.

apifrom.monitoring.start_timer(name)

Start a timer for measuring durations.

param name:

The name of the metric to record the duration

returns:

A unique timer ID

apifrom.monitoring.stop_timer(timer_id, labels=None)

Stop a timer and return the duration in milliseconds.

param timer_id:

The timer ID

param labels:

Optional labels to attach to the observation

returns:

The duration in milliseconds

apifrom.monitoring.track_error(error_type, endpoint=None)

Track an API error.

apifrom.monitoring.track_request(endpoint)

Track the start of an API request.

param endpoint:

The API endpoint being called

returns:

A timer ID for tracking the request duration

apifrom.monitoring.track_request_end(timer_id, endpoint, status_code)

Track the end of an API request.

param timer_id:

The timer ID returned by track_request

param endpoint:

The API endpoint that was called

param status_code:

The HTTP status code of the response

returns:

The request duration in milliseconds

class apifrom.monitoring.MetricsMiddleware(collector=None)[source]
Parameters:

collector (Optional[apifrom.monitoring.metrics.MetricsCollector])

Middleware for collecting metrics during API request processing.

This middleware tracks request counts, durations, and error rates for each endpoint in the API.

Initialize the metrics middleware.

param collector:

Metrics collector to use for tracking metrics.

apifrom.monitoring.on_error(request, error, endpoint)
:async:

Process an error that occurred during request handling.

param request:

The request object.

param error:

The exception that was raised.

param endpoint:

The endpoint being called.

apifrom.monitoring.post_request(request, response)
:async:

Process a response after it is generated by the endpoint.

param request:

The request object.

param response:

The response object.

apifrom.monitoring.pre_request(request, endpoint)
:async:

Process a request before it is handled by the endpoint.

param request:

The request object.

param endpoint:

The endpoint being called.

apifrom.monitoring.register(app)
:abstractmethod:

Register the middleware with the application.

param app:

The application to register with.

PrometheusExporter(collector = None):bases: MetricsExporter

Exports metrics in Prometheus text format.

Initialize the exporter with a metrics collector.

apifrom.monitoring._format_labels(labels)

Format labels for Prometheus exposition format.

apifrom.monitoring.export(**kwargs)

Export metrics in Prometheus text format.

returns:

String in Prometheus exposition format.

apifrom.monitoring.export_to_file(file_path)

Export metrics to a file in Prometheus text format.

param file_path:

Path to the output file.

apifrom.monitoring.serve_metrics(host='localhost', port=8000, endpoint='/metrics')

Start a simple HTTP server to expose metrics for Prometheus scraping.

param host:

Host to bind the server to.

param port:

Port to listen on.

param endpoint:

URL endpoint for metrics.