PATH:
opt
/
hc_python
/
lib
/
python3.12
/
site-packages
/
prometheus_client
from typing import Callable from urllib.parse import parse_qs from .exposition import _bake_output from .registry import CollectorRegistry, REGISTRY def make_asgi_app(registry: CollectorRegistry = REGISTRY, disable_compression: bool = False) -> Callable: """Create a ASGI app which serves the metrics from a registry.""" async def prometheus_app(scope, receive, send): assert scope.get("type") == "http" # Prepare parameters params = parse_qs(scope.get('query_string', b'')) accept_header = ",".join([ value.decode("utf8") for (name, value) in scope.get('headers') if name.decode("utf8").lower() == 'accept' ]) accept_encoding_header = ",".join([ value.decode("utf8") for (name, value) in scope.get('headers') if name.decode("utf8").lower() == 'accept-encoding' ]) # Bake output status, headers, output = _bake_output(registry, accept_header, accept_encoding_header, params, disable_compression) formatted_headers = [] for header in headers: formatted_headers.append(tuple(x.encode('utf8') for x in header)) # Return output payload = await receive() if payload.get("type") == "http.request": await send( { "type": "http.response.start", "status": int(status.split(' ')[0]), "headers": formatted_headers, } ) await send({"type": "http.response.body", "body": output}) return prometheus_app
[+]
__pycache__
[-] metrics_core.py
[edit]
[-] metrics.py
[edit]
[-] registry.py
[edit]
[-] platform_collector.py
[edit]
[-] __init__.py
[edit]
[+]
bridge
[-] mmap_dict.py
[edit]
[+]
twisted
[-] parser.py
[edit]
[-] multiprocess.py
[edit]
[-] gc_collector.py
[edit]
[-] values.py
[edit]
[+]
..
[-] context_managers.py
[edit]
[-] process_collector.py
[edit]
[-] py.typed
[edit]
[-] samples.py
[edit]
[-] decorator.py
[edit]
[-] exposition.py
[edit]
[+]
openmetrics
[-] core.py
[edit]
[-] utils.py
[edit]
[-] asgi.py
[edit]