### Run Example Application

Source: https://aioprometheus.readthedocs.io/en/latest/user/index.html

This snippet shows how to run an example asyncio application that serves Prometheus metrics. Ensure you have the 'aiohttp' extra installed for this functionality.

```python
loop = asyncio.get_event_loop()

app = ExampleApp()

loop.run_until_complete(app.start())
try:
    loop.run_forever()
except KeyboardInterrupt:
    pass
finally:
    loop.run_until_complete(app.stop())
loop.close()
```

--------------------------------

### Running the aioprometheus Service Example

Source: https://aioprometheus.readthedocs.io/en/latest/user/index.html

Execute the Python script to start the aioprometheus Service and expose metrics.

```bash
(venv) $ python simple-service-example.py
Serving prometheus metrics on: http://127.0.0.1:8000/metrics

```

--------------------------------

### Install development dependencies

Source: https://aioprometheus.readthedocs.io/en/latest/dev/index.html

Install requirements and the package in editable mode.

```bash
(aioprom) $ pip install -r requirements.dev.txt
(aioprom) $ pip install -e .[aiohttp,binary]
```

--------------------------------

### Installing aioprometheus with aiohttp

Source: https://aioprometheus.readthedocs.io/en/latest/user/index.html

Install the aioprometheus library with the necessary extras for aiohttp support.

```bash
$ pip install aioprometheus[aiohttp]
```

--------------------------------

### Install aioprometheus via pip

Source: https://aioprometheus.readthedocs.io/en/latest/user/index.html

Standard installation command for the aioprometheus library.

```bash
$ pip install aioprometheus
```

--------------------------------

### Setup virtual environment

Source: https://aioprometheus.readthedocs.io/en/latest/dev/index.html

Create and activate a virtual environment to isolate project dependencies.

```bash
$ python3 -m venv venv --prompt aioprom
$ source venv/bin/activate
(aioprom) $
(aioprom) $ pip install pip --upgrade
```

--------------------------------

### Install aioprometheus with optional extras

Source: https://aioprometheus.readthedocs.io/en/latest/user/index.html

Commands to install aioprometheus with specific framework or protocol support.

```bash
$ pip install aioprometheus[starlette]
```

```bash
$ pip install aioprometheus[quart]
```

```bash
$ pip install aioprometheus[aiohttp]
```

```bash
$ pip install aioprometheus[binary]
```

```bash
$ pip install aioprometheus[aiohttp,binary,starlette,quart]
```

--------------------------------

### Basic aioprometheus Service Example

Source: https://aioprometheus.readthedocs.io/en/latest/user/index.html

Demonstrates embedding the aioprometheus Service to expose metrics on an HTTP endpoint. Requires aiohttp extras. The service starts, a counter metric is defined and updated periodically, and the metrics URL is printed.

```python
#!/usr/bin/env python
"""
This example demonstrates how the ``aioprometheus.Service`` can be used to
expose metrics on a HTTP endpoint.

.. code-block:: console

    (env) $ python simple-service-example.py
    Serving prometheus metrics on: http://127.0.0.1:8000/metrics

You can open the URL in a browser or use the ``curl`` command line tool to
fetch metrics manually to verify they can be retrieved by Prometheus server.

"""

import asyncio
import socket

from aioprometheus import Counter
from aioprometheus.service import Service


async def main():

    service = Service()
    events_counter = Counter(
        "events", "Number of events.", const_labels={"host": socket.gethostname()}
    )

    await service.start(addr="127.0.0.1", port=8000)
    print(f"Serving prometheus metrics on: {service.metrics_url}")

    # Now start another coroutine to periodically update a metric to
    # simulate the application making some progress.
    async def updater(c: Counter):
        while True:
            c.inc({"kind": "timer_expiry"})
            await asyncio.sleep(1.0)

    await updater(events_counter)

    # Finally stop server
    await service.stop()


if __name__ == "__main__":

    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        pass

```

--------------------------------

### Install aioprometheus with Starlette extras

Source: https://aioprometheus.readthedocs.io/en/latest/user/index.html

Installs the necessary dependencies for using Starlette-based metrics rendering.

```bash
$ pip install aioprometheus[starlette]
```

--------------------------------

### Prometheus Configuration for Scraping Examples

Source: https://aioprometheus.readthedocs.io/en/latest/user/index.html

This is a minimal Prometheus configuration file to scrape metrics from example scripts. It sets a scrape interval and specifies the target address.

```yaml
global:
  scrape_interval:     15s # By default, scrape targets every 15 seconds.
  evaluation_interval: 15s # By default, scrape targets every 15 seconds.

scrape_configs:
  - job_name:       'test-app'

    # Override the global default and scrape targets from this job every
    # 5 seconds.
    scrape_interval: 5s
    scrape_timeout: 10s

    target_groups:
      - targets: ['localhost:8000']
        labels:
          group: 'dev'
```

--------------------------------

### Test Distribution

Source: https://aioprometheus.readthedocs.io/en/latest/dev/index.html

Verifies the distribution by installing it into a fresh virtual environment and running the test suite.

```bash
(aioprom) $ make test-dist
```

--------------------------------

### GET /

Source: https://aioprometheus.readthedocs.io/en/latest/api/aioprometheus.service.html

Handles requests to the root route, serving a trivial page with a link to the metrics.

```APIDOC
## GET /

### Description
Handle a request to the / route. Serves a trivial page with a link to the metrics. Use this if ever you need to point a health check at your the service.

### Method
GET

### Endpoint
/
```

--------------------------------

### Define and observe a Summary metric

Source: https://aioprometheus.readthedocs.io/en/latest/user/index.html

Example of recording observations in a Summary metric.

```python
from aioprometheus import Summary

http_access =  Summary("http_access_time", "HTTP access time")
http_access.observe({'time': '/static'}, 3.142)
```

```python
from aioprometheus import Summary

http_access =  Summary(
    "http_access_time",
    "HTTP access time",
    invariants=[(0.50, 0.05), (0.99, 0.001)])
```

--------------------------------

### Create and update a Counter metric

Source: https://aioprometheus.readthedocs.io/en/latest/user/index.html

Basic example of importing, instantiating, and incrementing a Counter metric.

```python
from aioprometheus import Counter

events_counter = Counter(
    "events_counter",
    "Total number of events.",
)

events_counter.inc({"kind": "event A"})
```

--------------------------------

### Define a Counter metric

Source: https://aioprometheus.readthedocs.io/en/latest/user/index.html

Example of defining a Counter metric for tracking cumulative events.

```python
from aioprometheus import Counter

uploads_metric = Counter("file_uploads_total", "File total uploads.")
uploads_metric.inc({'type': "png"})
```

--------------------------------

### GET /robots.txt

Source: https://aioprometheus.readthedocs.io/en/latest/api/aioprometheus.service.html

Handles requests to the robots.txt file to discourage indexing.

```APIDOC
## GET /robots.txt

### Description
Handle a request to /robots.txt. If a robot ever stumbles on this server, discourage it from indexing.

### Method
GET

### Endpoint
/robots.txt
```

--------------------------------

### Expose Metrics with ASGI Middleware in FastAPI

Source: https://aioprometheus.readthedocs.io/en/latest/user/index.html

Integrate aioprometheus metrics into a FastAPI application using the ASGI middleware. This example requires fastapi, uvicorn, and aioprometheus.

```python
from fastapi import FastAPI, Request, Response

from aioprometheus import Counter, MetricsMiddleware
from aioprometheus.asgi.starlette import metrics

app = FastAPI()

# Any custom application metrics are automatically included in the exposed
# metrics. It is a good idea to attach the metrics to 'app.state' so they
# can easily be accessed in the route handler - as metrics are often
```

--------------------------------

### GET /metrics

Source: https://aioprometheus.readthedocs.io/en/latest/api/aioprometheus.service.html

Handles requests to the metrics route, returning Prometheus metrics in the most efficient format.

```APIDOC
## GET /metrics

### Description
Handles a request to the metrics route. The request is inspected and the most efficient response data format is chosen.

### Method
GET

### Endpoint
/metrics
```

--------------------------------

### Push Metrics to Prometheus Push Gateway

Source: https://aioprometheus.readthedocs.io/en/latest/user/index.html

This Python snippet demonstrates how to use the Pusher object to send metrics to a Prometheus Push Gateway. It requires the 'aiohttp' extra to be installed. The Pusher can be configured with a job name and grouping keys.

```python
from aioprometheus import REGISTRY, Counter
from aioprometheus.pusher import Pusher

PUSH_GATEWAY_ADDR = "http://127.0.0.1:61423"
pusher = Pusher("my-job", PUSH_GATEWAY_ADDR, grouping_key={"instance": "127.0.0.1:1234"})
c = Counter("total_requests", "Total requests.", {})

c.inc({'url': "/p/user"})

# Push to the push gateway
resp = await pusher.replace(REGISTRY)
```

--------------------------------

### GET /metrics (Starlette)

Source: https://aioprometheus.readthedocs.io/en/latest/api/aioprometheus.asgi.starlette.html

The metrics function renders Prometheus metrics into the format specified by the 'accept' header. It automatically detects the registry from the Starlette app state or falls back to the default registry.

```APIDOC
## GET /metrics

### Description
Render metrics into the format specified by the 'accept' header. This function attempts to retrieve the metrics Registry from the Starlette request app state, falling back to the default registry if not found.

### Method
GET

### Endpoint
/metrics

### Response
#### Success Response (200)
- **body** (string) - The rendered metrics in the format requested by the 'accept' header.
```

--------------------------------

### GET /metrics - Quart Metrics Endpoint

Source: https://aioprometheus.readthedocs.io/en/latest/api/aioprometheus.asgi.quart.html

This endpoint renders Prometheus metrics for your Quart application. It attempts to use a custom registry if configured, otherwise falls back to the default registry.

```APIDOC
## GET /metrics

### Description
Renders metrics into the format specified by the ‘accept’ header. This function first attempts to retrieve the metrics Registry from `current_app` in case the app is using a custom registry instead of the default registry. If this fails then the default registry is used.

### Method
GET

### Endpoint
/metrics

### Parameters
#### Query Parameters
- **accept** (string) - Optional - The content type to render metrics in (e.g., "text/plain; version=0.0.4"). Defaults to Prometheus text format.

### Response
#### Success Response (200)
- **metrics** (string) - The Prometheus metrics in the requested format.
```

--------------------------------

### Count Exceptions with @count_exceptions Decorator

Source: https://aioprometheus.readthedocs.io/en/latest/user/index.html

Utilize the @count_exceptions decorator to count exceptions occurring within a function. This example requires asyncio and aioprometheus.service.

```python
# Create a metric to track requests currently in progress.
REQUEST_EXCEPTIONS = Counter(
    "request_handler_exceptions", "Number of exceptions in requests"
)
REQUESTS = Counter("request_total", "Total number of requests")

# Decorate function with metric.
@count_exceptions(REQUEST_EXCEPTIONS, {"route": "/"})
async def handle_request(duration):
    """A dummy function that occasionally raises an exception"""
    REQUESTS.inc({"route": "/"})
    if duration < 0.3:
        raise Exception("Ooops")
    await asyncio.sleep(duration)


async def handle_requests():
    # Start up the server to expose the metrics.
    await svr.start(port=8000)
    # Generate some requests.
    while True:
        try:
            await handle_request(random.random())
        except Exception:
            pass  # keep handling


if __name__ == "__main__":

    loop = asyncio.get_event_loop()

    svr = Service()

    try:
        loop.run_until_complete(handle_requests())
    except KeyboardInterrupt:
        pass
    finally:
        loop.run_until_complete(svr.stop())
    loop.stop()
    loop.close()
```

--------------------------------

### Manage documentation

Source: https://aioprometheus.readthedocs.io/en/latest/dev/index.html

Build and serve the project documentation locally.

```bash
(aioprom) $ make docs
```

```bash
(aioprom) $ make serve-docs
```

--------------------------------

### Run linting

Source: https://aioprometheus.readthedocs.io/en/latest/dev/index.html

Perform static analysis using Pylint.

```bash
(aioprom) $ make check-lint
```

--------------------------------

### Run unit tests

Source: https://aioprometheus.readthedocs.io/en/latest/dev/index.html

Execute the test suite using Makefile or the unittest module.

```bash
(aioprom) $ make test
```

```bash
(aioprom) $ python -m unittest discover -s tests -v
```

```bash
(aioprom) $ cd aioprometheus/tests
(aioprom) $ python -m unittest test_negotiate
```

--------------------------------

### Apply code style

Source: https://aioprometheus.readthedocs.io/en/latest/dev/index.html

Run the Black formatter and isort to ensure consistent code style.

```bash
(aioprom) $ make style
```

--------------------------------

### Run Prometheus with Configuration File

Source: https://aioprometheus.readthedocs.io/en/latest/user/index.html

Command to run the Prometheus server with a specified configuration file.

```bash
$ ./prometheus -config.file my-prom-config.yaml
```

--------------------------------

### Build Distribution

Source: https://aioprometheus.readthedocs.io/en/latest/dev/index.html

Creates a pure Python wheel for the package.

```bash
(aioprom) $ make dist
```

--------------------------------

### Clone the repository

Source: https://aioprometheus.readthedocs.io/en/latest/dev/index.html

Download the source code from GitHub to your local machine.

```bash
$ git clone git@github.com:claws/aioprometheus.git
$ cd aioprometheus
```

--------------------------------

### Upload to PyPI

Source: https://aioprometheus.readthedocs.io/en/latest/dev/index.html

Publishes the built distribution to the Python Package Index.

```bash
(aioprom) $ make upload-dist
```

--------------------------------

### Tag and Push Release

Source: https://aioprometheus.readthedocs.io/en/latest/dev/index.html

Creates a version tag in the local repository and pushes it to the remote origin.

```bash
$ git tag YY.MM.MICRO -m "A meaningful release tag comment"
$ git tag  # check release tag is in list
$ git push --tags origin master
```

--------------------------------

### Configure FastAPI with ASGI Metrics Middleware

Source: https://aioprometheus.readthedocs.io/en/latest/user/index.html

Attaches the MetricsMiddleware to a FastAPI application and defines a /metrics route for automatic metric collection.

```python
app.state.users_events_counter = Counter("events", "Number of events.")

app.add_middleware(MetricsMiddleware)
app.add_route("/metrics", metrics)


@app.get("/")
async def root(request: Request):  # pylint: disable=unused-argument
    return Response("FastAPI Middleware Example")


@app.get("/users/{user_id}")
async def get_user(
    request: Request,
    user_id: str,
):
    request.app.state.users_events_counter.inc({"path": request.scope["path"]})
    return Response(f"{user_id}")


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app)
```

--------------------------------

### Fetching Metrics from Service

Source: https://aioprometheus.readthedocs.io/en/latest/user/index.html

Use curl to fetch the metrics exposed by the running aioprometheus Service in plain text format.

```bash
$ curl http://127.0.0.1:8000/metrics
# HELP events Number of events.

```

--------------------------------

### Define and Observe Histograms

Source: https://aioprometheus.readthedocs.io/en/latest/user/index.html

Create a Histogram metric and record observations using specific label values.

```python
from aioprometheus import Histogram

http_access =  Histogram("http_access_time", "HTTP access time")
http_access.observe({'time': '/static'}, 3.142)
```

```python
from aioprometheus import Histogram

http_access =  Histogram(
    "http_access_time",
    "HTTP access time",
    buckets=[0.1, 0.5, 1.0, 5.0])
```

--------------------------------

### Expose Prometheus Metrics with aioprometheus Service

Source: https://aioprometheus.readthedocs.io/en/latest/user/index.html

This script implements an asyncio application that exposes application metrics using the aioprometheus Service. It requires the 'psutil' package for collecting system metrics. The Service object provides a new HTTP endpoint for Prometheus metrics.

```python
#!/usr/bin/env python
"""
This example implements an application that exposes application metrics
obtained from the psutil package.

This example requires the ``psutil`` package which can be installed
using ``pip install psutil``.
"""

import asyncio
import logging
import random
import socket

import psutil

from aioprometheus import Counter, Gauge, Histogram, Summary
from aioprometheus.service import Service


class ExampleApp:
    """
    This example application attempts to demonstrates how ``aioprometheus``
    can be integrated within a Python application built upon asyncio.

    This application attempts to simulate a long running distributed system
    process. It is intentionally not a web service application.

    In this case the aioprometheus Service object is used to provide a
    new HTTP endpoint that can be used to expose Prometheus metrics on.
    """

    def __init__(
        self,
        metrics_host="127.0.0.1",
        metrics_port: int = 8000,
    ):

        self.metrics_host = metrics_host
        self.metrics_port = metrics_port
        self.timer = None  # type: asyncio.Handle

        ######################################################################
        # Create application metrics and metrics service

        # Create a metrics server. The server will create a metrics collector
        # registry if one is not specifically created and passed in.
        self.msvr = Service()

        # Define some constant labels that need to be added to all metrics
        const_labels = {
            "host": socket.gethostname(),
            "app": f"{self.__class__.__name__}",
        }

        # Create metrics collectors. No registry is passed when creating the
        # metrics so they get registered with the default registry.

        # Create a counter metric to track requests.
        self.requests_metric = Counter(
            "requests", "Number of requests.", const_labels=const_labels
        )

        # Create a gauge metrics to track memory usage.
        self.ram_metric = Gauge(
            "memory_usage_bytes", "Memory usage in bytes.", const_labels=const_labels
        )

        # Create a gauge metrics to track CPU.
        self.cpu_metric = Gauge(
            "cpu_usage_percent", "CPU usage percent.", const_labels=const_labels
        )

        self.payload_metric = Summary(
            "request_payload_size_bytes",
            "Request payload size in bytes.",
            const_labels=const_labels,
            invariants=[(0.50, 0.05), (0.99, 0.001)],
        )

        self.latency_metric = Histogram(
            "request_latency_seconds",
            "Request latency in seconds",
            const_labels=const_labels,
            buckets=[0.1, 0.5, 1.0, 5.0],
        )

    async def start(self):
        """Start the application"""
        await self.msvr.start(addr=self.metrics_host, port=self.metrics_port)
        logger.debug(f"Serving prometheus metrics on: {self.msvr.metrics_url}")

        # Schedule a timer to update metrics. In a realistic application
        # the metrics would be updated as needed. In this example, a simple
        # timer is used to emulate things happening, which conveniently
        # allows all metrics to be updated at once.
        self.timer = asyncio.get_event_loop().call_later(1.0, self.on_timer_expiry)

    async def stop(self):
        """Stop the application"""
        await self.msvr.stop()
        if self.timer:
            self.timer.cancel()
        self.timer = None

    def on_timer_expiry(self):
        """Update application to simulate work"""

        # Update memory metrics
        self.ram_metric.set({"type": "virtual"}, psutil.virtual_memory().used)
        self.ram_metric.set({"type": "swap"}, psutil.swap_memory().used)

        # Update cpu metrics
        for c, p in enumerate(psutil.cpu_percent(interval=1, percpu=True)):
            self.cpu_metric.set({"core": c}, p)

        # Incrementing a requests counter to emulate webserver app
        self.requests_metric.inc({"path": "/"})

        # Monitor request payload data to emulate webserver app
        self.payload_metric.add({"path": "/data"}, random.random() * 2 ** 10)

        # Monitor request latency to emulate webserver app
        self.latency_metric.add({"path": "/data"}, random.random() * 5)

        # re-schedule another metrics update
        self.timer = asyncio.get_event_loop().call_later(1.0, self.on_timer_expiry)


if __name__ == "__main__":

    logging.basicConfig(level=logging.DEBUG)
    # Silence asyncio and aiohttp loggers
    logging.getLogger("asyncio").setLevel(logging.ERROR)
    logging.getLogger("aiohttp").setLevel(logging.ERROR)
    logger = logging.getLogger(__name__)

```

--------------------------------

### MetricsMiddleware Class

Source: https://aioprometheus.readthedocs.io/en/latest/api/aioprometheus.asgi.middleware.html

The MetricsMiddleware class is designed to wrap ASGI applications to automatically collect and expose Prometheus metrics.

```APIDOC
## MetricsMiddleware

### Description
Implements a Prometheus metrics collection middleware for ASGI applications, tracking requests, responses, and exceptions.

### Parameters
- **app** (Callable) - Required - The ASGI callable representing the next application or middleware in the chain.
- **registry** (Registry) - Optional - The collector registry to use for rendering metrics. Defaults to the default registry.
- **exclude_paths** (Sequence[str]) - Optional - A list of URL paths that should not trigger metric updates. Defaults to ('/metrics', '/metrics/', '/docs', '/openapi.json', '/docs/oauth2-redirect', '/redoc', '/favicon.ico').
- **use_template_urls** (bool) - Optional - Whether to use route template URLs (e.g., '/users/{user_id}') instead of literal paths. Supported in Starlette/FastAPI. Defaults to True.
- **group_status_codes** (bool) - Optional - Whether to group status codes by kind (e.g., 2xx). Defaults to False.
```

--------------------------------

### Check test coverage

Source: https://aioprometheus.readthedocs.io/en/latest/dev/index.html

Generate a report on code coverage.

```bash
(aioprom) $ make coverage
```

--------------------------------

### Check type annotations

Source: https://aioprometheus.readthedocs.io/en/latest/dev/index.html

Verify type safety using mypy.

```bash
(aioprom) $ make check-types
```

--------------------------------

### Bucket Generation Utilities

Source: https://aioprometheus.readthedocs.io/en/latest/api/aioprometheus.histogram.html

Utility functions to generate lists of bucket boundaries for histograms.

```APIDOC
## exponentialBuckets(start, factor, count)

### Description
Returns buckets that are spaced exponentially.

### Parameters
- **start** (Union[float, int]) - Required - The upper bound of the lowest bucket.
- **factor** (Union[float, int]) - Required - The factor by which each following bucket's upper bound is multiplied.
- **count** (int) - Required - The number of buckets to return.

## linearBuckets(start, width, count)

### Description
Returns buckets that are spaced linearly.

### Parameters
- **start** (Union[float, int]) - Required - The upper bound of the lowest bucket.
- **width** (Union[int, float]) - Required - The width of each bucket.
- **count** (int) - Required - The number of buckets to return.
```

--------------------------------

### Metric Name and Labels Notation

Source: https://aioprometheus.readthedocs.io/en/latest/api/aioprometheus.collectors.html

Illustrates the notation for identifying time series using metric names and labels. Labels enable Prometheus's dimensional data model.

```text
<metric name>{<label name>=<label value>, ...}

```

```text
api_http_requests_total{method="POST", handler="/messages"}

```

--------------------------------

### MetricDict Class

Source: https://aioprometheus.readthedocs.io/en/latest/api/aioprometheus.metricdict.html

Documentation for the MetricDict class which inherits from collections.abc.MutableMapping.

```APIDOC
## Class: aioprometheus.metricdict.MetricDict

### Description
MetricDict stores metric data based on labels. It generates custom hash keys based on the provided labels.

### Bases
- collections.abc.MutableMapping

### Attributes
- **EMPTY_KEY** (str) - Constant value '__EMPTY__' used for empty label sets.
```

--------------------------------

### aioprometheus.renderer.render

Source: https://aioprometheus.readthedocs.io/en/latest/api/aioprometheus.renderer.html

Renders metrics from a registry into a specific format based on provided Accept headers. Defaults to Text format if no headers are given.

```APIDOC
## aioprometheus.renderer.render

### Description
Render the metrics in this registry to a specific format. The format chosen is determined by scanning through the ACCEPTS headers and selecting the most efficient format. If no accepts headers information is provided then Text format is used as the default.

### Method
(Implied Function Call, not a standard HTTP method)

### Endpoint
(N/A - This is a Python function)

### Parameters
#### Path Parameters
(N/A)

#### Query Parameters
(N/A)

#### Request Body
(N/A)

### Parameters
- **_registry** (aioprometheus.collectors.Registry_) - Required - A collector registry that contains the metrics to be rendered into a specific format.
- **_accepts_headers** (Sequence[str]) - Required - a list of ACCEPT headers fields extracted from a request.

### Returns
- **Tuple[bytes, dict]** - a 2-tuple where the first item is a bytes object that represents the formatted metrics and the second item is a dict of header fields that can be added to a HTTP response.

### Request Example
(N/A - This is a Python function)

### Response
#### Success Response (200)
(N/A - This is a Python function, not an HTTP response)

#### Response Example
(N/A - This is a Python function)
```

--------------------------------

### Define and manipulate a Gauge metric

Source: https://aioprometheus.readthedocs.io/en/latest/user/index.html

Demonstrates setting, incrementing, decrementing, adding, and subtracting values from a Gauge.

```python
from aioprometheus import Gauge

ram_metric = Gauge("memory_usage_bytes", "Memory usage in bytes.")
ram_metric.set({'type': "virtual"}, 100)
ram_metric.inc({'type': "virtual"})
ram_metric.dec({'type': "virtual"})
ram_metric.add({'type': "virtual"}, 5)
ram_metric.sub({'type': "virtual"}, -5)
```

--------------------------------

### aioprometheus.negotiator.parse_accepts

Source: https://aioprometheus.readthedocs.io/en/latest/api/aioprometheus.negotiator.html

Parses Accept headers into a set of items.

```APIDOC
## parse_accepts(accept_headers)

### Description
Return a sequence of accepts items in the request headers.

### Parameters
#### Arguments
- **accept_headers** (Sequence[str]) - Required - The accept headers to parse.

### Returns
- **Set[str]** - A set of accepted items.
```

--------------------------------

### GitHub Release URL Format

Source: https://aioprometheus.readthedocs.io/en/latest/dev/index.html

The URL structure for accessing the release tarball generated by GitHub.

```text
https://github.com/{username}/{repo}/tarball/{tag}.tar.gz
```

--------------------------------

### Registry Management

Source: https://aioprometheus.readthedocs.io/en/latest/api/aioprometheus.collectors.html

Functionality for accessing and managing the default Prometheus registry.

```APIDOC
## aioprometheus.collectors.get_registry() 

### Description
Return the default Registry.

### Method
Not applicable (static method or function)

### Endpoint
Not applicable
```

--------------------------------

### Pusher.add

Source: https://aioprometheus.readthedocs.io/en/latest/api/aioprometheus.pusher.html

Pushes metrics to the gateway, replacing only those with the same name as the newly pushed metrics.

```APIDOC
## POST /metrics (add)

### Description
Works like replace, but only metrics with the same name as the newly pushed metrics are replaced.

### Method
POST

### Parameters
#### Request Body
- **registry** (aioprometheus.collectors.Registry) - Optional - The registry containing the metrics to push.
```

--------------------------------

### Apply Constant Labels

Source: https://aioprometheus.readthedocs.io/en/latest/user/index.html

Use constant labels to automatically include specific labels in all metrics gathered by a collector.

```python
host = socket.gethostname()
ram_metric = Gauge("memory_usage_bytes", "Memory usage in bytes.")
ram_metric.set({'type': "virtual", 'host': host}, 100)
ram_metric.set({'type': "swap", 'host': host}, 100)
```

```python
ram_metric = Gauge(
    "memory_usage_bytes", "Memory usage in bytes.",
    const_labels={'host': socket.gethostname()})
ram_metric.set({'type': "virtual"}, 100)
ram_metric.set({'type': "swap"}, 100)
```

--------------------------------

### Implement Custom Metrics Handler in FastAPI

Source: https://aioprometheus.readthedocs.io/en/latest/user/index.html

Uses the aioprometheus render function to manually expose metrics from the default registry via a custom route.

```python
#!/usr/bin/env python
"""
This example adds Prometheus metrics to a FastAPI application. In this
example a counter metric is instantiated and gets updated whenever the "/"
route is accessed.

A '/metrics' route is implemented using the render function and added to
the application using the standard web framework method. The metrics route
renders Prometheus metrics from the default collector registry into the
appropriate format.

Run:

  (venv) $ pip install fastapi uvicorn
  (venv) $ python fastapi-example.py

"""

from typing import List

from fastapi import FastAPI, Header, Request, Response

from aioprometheus import REGISTRY, Counter, render

app = FastAPI()
app.state.events_counter = Counter("events", "Number of events.")


@app.get("/")
async def hello(request: Request):
    request.app.state.events_counter.inc({"path": "/"})
    return "FastAPI Hello"


@app.get("/metrics")
async def handle_metrics(
    request: Request,  # pylint: disable=unused-argument
    accept: List[str] = Header(None),
) -> Response:
    content, http_headers = render(REGISTRY, accept)
    return Response(content=content, media_type=http_headers["Content-Type"])


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app)
```

--------------------------------

### get_full_or_template_path

Source: https://aioprometheus.readthedocs.io/en/latest/api/aioprometheus.asgi.middleware.html

Utility method to resolve the route template path for metrics reporting.

```APIDOC
## get_full_or_template_path

### Description
Retrieves the route template URL for a given scope, allowing for more insightful metrics by identifying the route handler rather than specific path instances.

### Parameters
- **scope** (Dict) - Required - The ASGI scope dictionary.
```

--------------------------------

### timer Decorator

Source: https://aioprometheus.readthedocs.io/en/latest/api/aioprometheus.decorators.html

Wraps a callable to measure its execution time and update a metric with the duration.

```APIDOC
## aioprometheus.decorators.timer

### Description
This decorator wraps a callable with code to calculate how long the callable takes to execute and updates the metric with the duration.

### Parameters
#### Path Parameters
- **_metric** (aioprometheus.collectors.Summary_) - Required - A metric to update with the calculated function duration. Must be a Summary metric object.
- **_labels** (Optional[Dict[str, str]]) - Optional - A dict of extra labels to associate with the metric.

### Returns
a callable wrapping the decorated function. The callable will be awaitable if the wrapped function was a coroutine function.
```

--------------------------------

### Collector Methods

Source: https://aioprometheus.readthedocs.io/en/latest/api/aioprometheus.collectors.html

Methods for interacting with collector metrics, such as adding observations and retrieving aggregated data.

```APIDOC
## add(_labels : Dict[str, str]_, _value : Union[int, float, aioprometheus.histogram.Histogram, quantile.Estimator]_) 

### Description
Add a single observation to the summary.

### Method
Not applicable (method of a class instance)

### Endpoint
Not applicable

## observe(_labels : Dict[str, str]_, _value : Union[int, float, aioprometheus.histogram.Histogram, quantile.Estimator]_) 

### Description
Add a single observation to the summary.

### Method
Not applicable (method of a class instance)

### Endpoint
Not applicable

## get(_labels : Dict[str, str]_) 

### Description
Get a dict of values, containing the sum, count and quantiles, matching an arbitrary group of labels.

### Method
Not applicable (method of a class instance)

### Endpoint
Not applicable

### Raises
KeyError if an item with matching labels is not present.
```

--------------------------------

### Manage Metric Labels

Source: https://aioprometheus.readthedocs.io/en/latest/user/index.html

Use labels to group related time series data within a single collector.

```python
ram_metric = Gauge("memory_usage_bytes", "Memory usage in bytes.")
ram_metric.set({'type': "virtual"}, 100)
```

```python
ram_metric.set({'type': "swap"}, 100.1)
```

--------------------------------

### Pusher.replace

Source: https://aioprometheus.readthedocs.io/en/latest/api/aioprometheus.pusher.html

Pushes new values for a group of metrics to the push gateway, replacing all existing metrics with the same grouping key.

```APIDOC
## PUT /metrics (replace)

### Description
Pushes new values for a group of metrics to the push gateway. All existing metrics with the same grouping key specified in the URL will be replaced with the new metrics value.

### Method
PUT

### Parameters
#### Request Body
- **registry** (aioprometheus.collectors.Registry) - Optional - The registry containing the metrics to push.
```

--------------------------------

### Registry Management

Source: https://aioprometheus.readthedocs.io/en/latest/api/aioprometheus.collectors.html

Methods for managing collectors within the Registry.

```APIDOC
## Registry Management

### `clear() -> None`

Clear all registered collectors. This function is mainly of use in tests to reset the default registry which may be used in multiple tests.

### `deregister(_name: str) -> None`

Deregister a collector. This will stop the collector metrics from being emitted.

Parameters:
    name (str): The name of the collector to deregister.

Raises:
    KeyError: if collector is not already registered.

### `get(_name: str) -> aioprometheus.collectors.Collector`

Get a collector by name.

Parameters:
    name (str): The name of the collector to fetch.

Raises:
    KeyError: if collector is not found.

### `get_all() -> List[aioprometheus.collectors.Collector]`

Return a list of all collectors.

### `register(_collector: aioprometheus.collectors.Collector) -> None`

Register a collector into the container. The registry provides a container that can be used to access all metrics when exposing them into a specific format.

Parameters:
    collector (aioprometheus.collectors.Collector): A collector to register in the registry.

Raises:
    TypeError: if collector is not an instance of `Collector`.
    ValueError: if collector is already registered.
```

--------------------------------

### Accessing Metrics Endpoint

Source: https://aioprometheus.readthedocs.io/en/latest/user/index.html

Use curl to access the default metrics endpoint provided by the aioprometheus Service.

```bash
$ curl http://127.0.0.1:8000/
```

--------------------------------

### Counter Set Method

Source: https://aioprometheus.readthedocs.io/en/latest/api/aioprometheus.collectors.html

Sets the Counter to an arbitrary value. Use with caution as it bypasses the cumulative nature of counters.

```python
set(_labels : Dict[str, str]_, _value : Union[int, float, aioprometheus.histogram.Histogram, quantile.Estimator]_) → None¶ 
    
    Set the counter to an arbitrary value. 

```

--------------------------------

### TextFormatter Class

Source: https://aioprometheus.readthedocs.io/en/latest/api/aioprometheus.formats.text.html

Details about the TextFormatter class and its methods.

```APIDOC
## Class aioprometheus.formats.text.TextFormatter

### Description
This formatter encodes into the Text format. The histogram and summary types are difficult to represent in the text format. The following conventions apply:

* The sample sum for a summary or histogram named x is given as a separate sample named x_sum.
* The sample count for a summary or histogram named x is given as a separate sample named x_count.
* Each quantile of a summary named x is given as a separate sample line with the same name x and a label {quantile=”y”}.
* Each bucket count of a histogram named x is given as a separate sample line with the name x_bucket and a label {le=”y”} (where y is the upper bound of the bucket).
* A histogram must have a bucket with {le=”+Inf”}. Its value must be identical to the value of x_count.
* The buckets of a histogram and the quantiles of a summary must appear in increasing numerical order of their label values (for the le or the quantile label, respectively).

### Parameters
#### Constructor Parameters
- **_timestamp** (bool) - Optional - Whether to include timestamps.

### Methods
#### get_headers()

##### Description
Returns a dict of HTTP headers for this response format.

##### Returns
- Dict[str, str] - A dictionary of HTTP headers.

#### marshall(_registry_)

##### Description
Marshalls a registry (containing collectors) into a bytes object.

##### Parameters
- **_registry** (aioprometheus.collectors.Registry) - The registry to marshall.

##### Returns
- bytes - The marshalled registry as bytes.

#### marshall_collector(_collector_)

##### Description
Marshalls a collector into a string containing one or more lines.

##### Parameters
- **_collector** (aioprometheus.collectors.Collector) - The collector to marshall.

##### Returns
- str - A string representation of the collector.

#### marshall_lines(_collector_)

##### Description
Marshalls a collector into a sequence of strings representing the metrics in the collector.

##### Parameters
- **_collector** (aioprometheus.collectors.Collector) - The collector to marshall.

##### Returns
- List[str] - A list of strings representing the metrics.
```

--------------------------------

### Clear Default Registry for Testing

Source: https://aioprometheus.readthedocs.io/en/latest/user/index.html

This snippet shows how to clear the default Prometheus registry, which is essential for resetting the state between unit tests to avoid errors related to existing metrics.

```python
REGISTRY.clear()
```

--------------------------------

### aioprometheus.negotiator.negotiate

Source: https://aioprometheus.readthedocs.io/en/latest/api/aioprometheus.negotiator.html

Negotiates a response format based on provided Accept headers.

```APIDOC
## negotiate(accepts_headers)

### Description
Negotiate a response format by scanning through a list of ACCEPTS headers and selecting the most efficient format. The formatter returned by this function is used to render a response.

### Parameters
#### Arguments
- **accepts_headers** (Sequence[str]) - Required - A list of ACCEPT headers fields extracted from a request.

### Returns
- **Type[aioprometheus.formats.base.IFormatter]** - A formatter class to form up the response into the appropriate representation.
```

--------------------------------

### Track Requests in Progress with Gauge

Source: https://aioprometheus.readthedocs.io/en/latest/user/index.html

Use the Gauge metric and @inprogress decorator to track the number of concurrent requests. Requires asyncio and aioprometheus.service.

```python
# Create a metric to track requests currently in progress.
REQUESTS_IN_PROGRESS = Gauge("request_in_progress", "Number of requests in progress")
REQUESTS = Counter("request_total", "Total number of requests")

# Decorate function with metric.
@inprogress(REQUESTS_IN_PROGRESS, {"route": "/"})
async def handle_request(duration):
    """A dummy function that takes some time"""
    REQUESTS.inc({"route": "/"})
    await asyncio.sleep(duration)


async def handle_requests():
    # Start up the server to expose the metrics.
    await svr.start(port=8000)
    # Generate some requests.
    while True:
        await handle_request(random.random())


if __name__ == "__main__":

    loop = asyncio.get_event_loop()

    svr = Service()

    try:
        loop.run_until_complete(handle_requests())
    except KeyboardInterrupt:
        pass
    finally:
        loop.run_until_complete(svr.stop())
    loop.stop()
    loop.close()
```

--------------------------------

### IFormatter Abstract Base Class

Source: https://aioprometheus.readthedocs.io/en/latest/api/aioprometheus.formats.base.html

The IFormatter class serves as an abstract base class for all metrics formatters within the aioprometheus library. It defines the interface that concrete formatter implementations must adhere to.

```APIDOC
## Abstract Base Class: IFormatter

### Description
An abstract base class for metrics formatters. It defines the interface for classes that will format Prometheus metrics into various output formats.

### Methods

#### `_get_headers()`

- **Description**: Returns a dictionary of HTTP headers required for the response when using this formatter.
- **Abstract**: Yes

#### `_marshall(registry)`

- **Description**: Marshalls a given registry (which contains multiple collectors) into a specific output format as bytes.
- **Parameters**:
  - `registry` (Registry): The registry object containing the metrics to be marshalled.
- **Returns**:
  - `bytes`: The marshalled metrics in the specified format.
- **Abstract**: Yes
```

--------------------------------

### Gauge Collector Methods

Source: https://aioprometheus.readthedocs.io/en/latest/api/aioprometheus.collectors.html

Methods available for the Gauge collector to increment, set, or subtract values.

```APIDOC
## Gauge Collector Methods

### `inc(_labels: Dict[str, str]) -> None`

Increments the gauge by 1.

### `set(_labels: Dict[str, str], _value: Union[int, float, aioprometheus.histogram.Histogram, quantile.Estimator]) -> None`

Set the gauge to an arbitrary value.

### `sub(_labels: Dict[str, str], _value: Union[int, float, aioprometheus.histogram.Histogram, quantile.Estimator]) -> None`

Subtract the given value from the Gauge. The value can be negative, resulting in an increase of the gauge.
```

--------------------------------

### Instrument Code with @timer Decorator

Source: https://aioprometheus.readthedocs.io/en/latest/user/index.html

Use the @timer decorator to automatically measure the execution time of functions.

```python
#!/usr/bin/env python
"""
Usage:

.. code-block:: python

    $ python decorator_timer.py

The example script can be tested using ``curl``.

.. code-block:: console

    $ curl :8000/metrics
    # HELP request_processing_seconds Time spent processing request
    # TYPE request_processing_seconds summary
    request_processing_seconds_count 77
    request_processing_seconds_sum 38.19072341918945
    request_processing_seconds{quantile="0.5"} 0.27150511741638184
    request_processing_seconds{quantile="0.9"} 0.5016570091247559
    request_processing_seconds{quantile="0.99"} 0.6077709197998047

"""

import asyncio
import random

from aioprometheus import Counter, Summary, timer
from aioprometheus.service import Service

# Create a metric to track time spent and requests made.
REQUEST_TIME = Summary("request_processing_seconds", "Time spent processing request")
REQUESTS = Counter("request_total", "Total number of requests")

# Decorate function with metric.
@timer(REQUEST_TIME)
async def handle_request(duration):
    """A dummy function that takes some time"""
    REQUESTS.inc({"route": "/"})
    await asyncio.sleep(duration)


async def handle_requests():
    # Start up the server to expose the metrics.
    await svr.start(port=8000)
    # Generate some requests.
    while True:
        await handle_request(random.random())


if __name__ == "__main__":

    loop = asyncio.get_event_loop()

    svr = Service()

    try:
        loop.run_until_complete(handle_requests())
    except KeyboardInterrupt:
        pass
    finally:
        loop.run_until_complete(svr.stop())
    loop.stop()
    loop.close()
```

--------------------------------

### count_exceptions Decorator

Source: https://aioprometheus.readthedocs.io/en/latest/api/aioprometheus.decorators.html

Wraps a callable to count the number of exceptions it raises.

```APIDOC
## aioprometheus.decorators.count_exceptions

### Description
This decorator wraps a callable with code to count how many times the callable generates an exception.

### Parameters
#### Path Parameters
- **_metric** (aioprometheus.collectors.Counter_) - Required - A metric to increment when an exception is caught. Must be a Counter metric object.
- **_labels** (Optional[Dict[str, str]]) - Optional - A dict of extra labels to associate with the metric.

### Returns
a callable wrapping the decorated function. The callable will be awaitable if the wrapped function was a coroutine function.
```

--------------------------------

### inprogress Decorator

Source: https://aioprometheus.readthedocs.io/en/latest/api/aioprometheus.decorators.html

Wraps a callable to track its in-progress status, incrementing a metric before execution and decrementing it upon completion.

```APIDOC
## aioprometheus.decorators.inprogress

### Description
This decorator wraps a callables with code to track whether it is currently in progress. The metric is incremented before calling the callable and is decremented when the callable is complete.

### Parameters
#### Path Parameters
- **_metric** (aioprometheus.collectors.Gauge_) - Required - A metric to increment and decrement. Must be a Gauge metric object.
- **_labels** (Optional[Dict[str, str]]) - Optional - A dict of extra labels to associate with the metric.

### Returns
a callable wrapping the decorated function. The callable will be awaitable if the wrapped function was a coroutine function.
```

--------------------------------

### Counter Add Method

Source: https://aioprometheus.readthedocs.io/en/latest/api/aioprometheus.collectors.html

Adds a value to a Counter. The value must be non-negative as Counters can only increase.

```python
add(_labels : Dict[str, str]_, _value : Union[int, float, aioprometheus.histogram.Histogram, quantile.Estimator]_) → None¶ 
    
    Add the given value to the counter. 

    Raises 
        
    ValueError if the value is negative. Counters can only increase. 

```

--------------------------------

### Counter Collector

Source: https://aioprometheus.readthedocs.io/en/latest/api/aioprometheus.collectors.html

A Counter is a cumulative metric that only increases. It's used for counting events like requests processed or tasks completed.

```APIDOC
## Counter Collector

### Description
A counter is a cumulative metric that represents a single numerical value that only ever goes up. Typically used to count requests served, tasks completed, errors occurred, etc.

### Methods
- **add(_labels: Dict[str, str]_, _value: Union[int, float, aioprometheus.histogram.Histogram, quantile.Estimator]_)**: Add the given value to the counter. Raises ValueError if the value is negative.
- **get(_labels: Dict[str, str]_)**: Get the Counter value matching an arbitrary group of labels. Raises KeyError if an item with matching labels is not present.
- **inc(_labels: Dict[str, str]_)**: Increments the counter by 1.
- **set(_labels: Dict[str, str]_, _value: Union[int, float, aioprometheus.histogram.Histogram, quantile.Estimator]_)**: Set the counter to an arbitrary value.

### Attributes
- **kind**: Type identifier for the counter (0).
```

--------------------------------

### Summary Collector

Source: https://aioprometheus.readthedocs.io/en/latest/api/aioprometheus.collectors.html

Details on the Summary collector, its purpose, and its components.

```APIDOC
## Summary Collector

### Description
A Summary metric captures individual observations from an event or sample stream and summarizes them in a manner similar to traditional summary statistics:
  1. sum of observations,
  2. observation count,
  3. rank estimations.

Example use cases for Summaries: - Response latency - Request size
```

--------------------------------

### Track In-Progress Requests with @inprogress Decorator

Source: https://aioprometheus.readthedocs.io/en/latest/user/index.html

Use the @inprogress decorator to track the number of concurrent requests currently being processed.

```python
#!/usr/bin/env python
"""

.. code-block:: python

    $ python decorator_inprogress.py

The example script can be tested using ``curl``.

.. code-block:: console

    $ curl :8000/metrics
    # HELP request_in_progress Number of requests in progress
    # TYPE request_in_progress gauge
    request_in_progress{route="/"} 1

"""

import asyncio
import random

from aioprometheus import Counter, Gauge, inprogress
from aioprometheus.service import Service
```

--------------------------------

### Metrics Types Enum

Source: https://aioprometheus.readthedocs.io/en/latest/api/aioprometheus.collectors.html

Enumeration for different types of metrics.

```APIDOC
## Metrics Types Enum

### `MetricsTypes(_value_)

An enumeration.

- `counter` = 0
- `gauge` = 1
- `histogram` = 4
- `summary` = 2
- `untyped` = 3
```

--------------------------------

### Counter Increment Method

Source: https://aioprometheus.readthedocs.io/en/latest/api/aioprometheus.collectors.html

Increments the Counter by 1. Use this for simple counting events.

```python
inc(_labels : Dict[str, str]_) → None¶ 
    
    Increments the counter by 1. 

```

--------------------------------

### Gauge Add Method

Source: https://aioprometheus.readthedocs.io/en/latest/api/aioprometheus.collectors.html

Adds a value to the Gauge. The value can be negative, causing the gauge to decrease.

```python
add(_labels : Dict[str, str]_, _value : Union[int, float, aioprometheus.histogram.Histogram, quantile.Estimator]_) → None¶ 
    
    Add the given value to the Gauge.
    The value can be negative, resulting in a decrease of the gauge. 

```