(request: web.Request)
| 14 | """Create a aiohttp handler which serves the metrics from a registry.""" |
| 15 | |
| 16 | async def prometheus_handler(request: web.Request) -> web.Response: |
| 17 | # Prepare parameters |
| 18 | params = {key: request.query.getall(key) for key in request.query.keys()} |
| 19 | accept_header = ",".join(request.headers.getall(hdrs.ACCEPT, [])) |
| 20 | accept_encoding_header = "" |
| 21 | # Bake output |
| 22 | status, headers, output = _bake_output( |
| 23 | registry, |
| 24 | accept_header, |
| 25 | accept_encoding_header, |
| 26 | params, |
| 27 | # use AIOHTTP's compression |
| 28 | disable_compression=True, |
| 29 | ) |
| 30 | response = web.Response( |
| 31 | status=int(status.split(" ")[0]), |
| 32 | headers=headers, |
| 33 | body=output, |
| 34 | ) |
| 35 | if not disable_compression: |
| 36 | response.enable_compression() |
| 37 | return response |
| 38 | |
| 39 | return prometheus_handler |
nothing calls this directly
no test coverage detected