X7ROOT File Manager
Current Path:
/opt/hc_python/lib/python3.12/site-packages/prometheus_client
opt
/
hc_python
/
lib
/
python3.12
/
site-packages
/
prometheus_client
/
📁
..
📄
__init__.py
(1.77 KB)
📁
__pycache__
📄
asgi.py
(1.57 KB)
📁
bridge
📄
context_managers.py
(2.29 KB)
📄
core.py
(860 B)
📄
decorator.py
(15.43 KB)
📄
exposition.py
(25.56 KB)
📄
gc_collector.py
(1.48 KB)
📄
metrics.py
(27.45 KB)
📄
metrics_core.py
(15.18 KB)
📄
mmap_dict.py
(5.27 KB)
📄
multiprocess.py
(7.36 KB)
📁
openmetrics
📄
parser.py
(7.26 KB)
📄
platform_collector.py
(1.83 KB)
📄
process_collector.py
(3.77 KB)
📄
py.typed
(0 B)
📄
registry.py
(6.05 KB)
📄
samples.py
(1.65 KB)
📁
twisted
📄
utils.py
(594 B)
📄
values.py
(4.88 KB)
Editing: asgi.py
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
Upload File
Create Folder