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. .. py:currentmodule:: apifrom.monitoring Overview -------- **Classes** * :py:class:`JSONExporter` * :py:class:`LogExporter` * :py:class:`Metric` * :py:class:`MetricType` * :py:class:`MetricsCollector` * :py:class:`MetricsMiddleware` * :py:class:`PrometheusExporter` Classes ------- .. py:class:: JSONExporter(collector = None):bases: MetricsExporter Exports metrics data in JSON format. Initialize the exporter with a metrics collector. .. method:: 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. .. method:: 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. .. py:class:: 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. .. :: level .. :: logger .. method:: export(include_details = False, **kwargs) Export metrics to logs. :param include_details: Whether to include detailed metrics information. .. method:: 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. .. py:class:: Metric Represents a single metric with its metadata and values. .. :: buckets :annotation: List[float] .. :: description :annotation: str .. :: labels :annotation: Dict[str, str] .. :: name :annotation: str .. :: quantiles :annotation: List[float] .. :: type :annotation: MetricType .. :: value :annotation: Union[float, int, List[float]] .. method:: __post_init__() Initialize the metric based on its type. .. method:: get_histogram_buckets() Get histogram bucket counts. .. method:: get_stats() Get basic statistics for histogram and summary metrics. .. method:: get_summary_quantiles() Get summary quantiles. .. method:: increment(amount = 1.0, labels = None) Increment a counter metric. .. method:: observe(value, labels = None) Add an observation to a histogram or summary metric. .. method:: set(value, labels = None) Set a gauge metric to a specific value. .. py:class:: MetricType:bases: enum.Enum Types of metrics that can be collected. .. :: COUNTER .. :: GAUGE .. :: HISTOGRAM .. :: SUMMARY .. py:class:: MetricsCollector 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. .. :: _initialized .. :: _instance .. :: _metrics .. :: _timers .. method:: create_counter(name, description, labels = None) Create a new counter metric. .. method:: create_gauge(name, description, labels = None) Create a new gauge metric. .. method:: create_histogram(name, description, buckets = None, labels = None) Create a new histogram metric. .. method:: create_metric(metric) Register a new metric with the collector. :param metric: The metric to register :returns: The registered metric .. method:: create_summary(name, description, quantiles = None, labels = None) Create a new summary metric. .. method:: get_all_metrics() Get all metrics. .. method:: get_metric(name) Get a metric by name. .. method:: increment(name, amount = 1.0, labels = None) Increment a counter metric. .. method:: observe(name, value, labels = None) Add an observation to a histogram or summary metric. .. method:: reset() Reset all metrics. .. method:: set(name, value, labels = None) Set a gauge metric to a specific value. .. method:: 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 .. method:: 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 .. method:: track_error(error_type, endpoint = None) Track an API error. .. method:: 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 .. method:: 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 .. py:class:: MetricsMiddleware(collector = None) 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. .. :: _request_timers .. :: collector .. method:: 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. .. method:: 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. .. method:: 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. .. method:: register(app) :abstractmethod: Register the middleware with the application. :param app: The application to register with. .. py:class:: PrometheusExporter(collector = None):bases: MetricsExporter Exports metrics in Prometheus text format. Initialize the exporter with a metrics collector. .. method:: _format_labels(labels) Format labels for Prometheus exposition format. .. method:: export(**kwargs) Export metrics in Prometheus text format. :returns: String in Prometheus exposition format. .. method:: export_to_file(file_path) Export metrics to a file in Prometheus text format. :param file_path: Path to the output file. .. method:: 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.