diff --git a/python/meson.build b/python/meson.build index e0c9bc1..768c7c4 100644 --- a/python/meson.build +++ b/python/meson.build @@ -1,6 +1,6 @@ project( run_command( - 'tomlq', '-r', '.project.name', 'pyproject.toml', + '.venv/bin/tomlq', '-r', '.project.name', 'pyproject.toml', check: true ).stdout().strip('\n'), # 'online.fxreader.uv', diff --git a/python/online/fxreader/pr34/commands_typed/metrics.py b/python/online/fxreader/pr34/commands_typed/metrics.py index 770ccb5..0409346 100644 --- a/python/online/fxreader/pr34/commands_typed/metrics.py +++ b/python/online/fxreader/pr34/commands_typed/metrics.py @@ -6,99 +6,88 @@ import datetime import django.http from typing import ( - Literal, Any, Optional, Annotated, cast, - TypeVar, Protocol, Generic, Callable, + Literal, + Any, + Optional, + Annotated, + cast, + TypeVar, + Protocol, + Generic, + Callable, ) logger = logging.getLogger(__name__) + class Metric(pydantic.BaseModel): - name: str - type: Literal['gauge', 'counter'] - help: Optional[str] = None + name: str + type: Literal['gauge', 'counter'] + help: Optional[str] = None - class Sample(pydantic.BaseModel): - value: str - parameters: dict[str, str] - timestamp: Optional[datetime.datetime] = None + class Sample(pydantic.BaseModel): + value: str + parameters: dict[str, str] + timestamp: Optional[datetime.datetime] = None - samples: list[Sample] = pydantic.Field( - default_factory=lambda: [], - ) + samples: list[Sample] = pydantic.Field( + default_factory=lambda: [], + ) - @classmethod - def sample_serialize( - cls, - o: 'Metric', - s: 'Metric.Sample', - ) -> str: - samples: list[Metric.Sample] = [s,] + @classmethod + def sample_serialize( + cls, + o: 'Metric', + s: 'Metric.Sample', + ) -> str: + samples: list[Metric.Sample] = [ + s, + ] - if o.type == 'gauge': - samples.append( - Metric.Sample( - parameters=s.parameters, - value='NaN', - timestamp=( - s.timestamp + datetime.timedelta(seconds=15) - if s.timestamp - else None - ) - ) - ) + if o.type == 'gauge': + samples.append( + Metric.Sample(parameters=s.parameters, value='NaN', timestamp=(s.timestamp + datetime.timedelta(seconds=15) if s.timestamp else None)) + ) + + return ''.join( + [ + '{metric}{{{parameters}}} {value} {timestamp}\n'.format( + metric=o.name, + parameters=','.join( + [ + '%s=%s' + % ( + k, + json.dumps(v), + ) + for k, v in s2.parameters.items() + ] + ), + value=s2.value, + timestamp=('%.f' % (s2.timestamp.timestamp() * 1000,) if s2.timestamp else ''), + ) + for s2 in samples + ] + ) - return ''.join([ - '{metric}{{{parameters}}} {value} {timestamp}\n'.format( - metric=o.name, - parameters=','.join([ - '%s=%s' % ( - k, - json.dumps(v), - ) - for k, v in s2.parameters.items() - ]), - value=s2.value, - timestamp=( - '%.f' % (s2.timestamp.timestamp() * 1000,) - if s2.timestamp - else '' - ), - ) - for s2 in samples - ]) def serialize( - metrics: list[Metric], + metrics: list[Metric], ): - return django.http.HttpResponse( - ''.join([ - '{help}{type}{samples}'.format( - #help='# HELP %s some metric' % o.name, - #type='# TYPE %s counter' % o.name, - help=( - '# HELP {0} {1}\n'.format( - o.name, - o.help - ) - if o.help - else '' - ), - type=( - '# TYPE {0} {1}\n'.format( - o.name, - o.type - ) - if o.type - else '' - ), - samples=''.join([ - Metric.sample_serialize(o, s) - for s in o.samples - ]), - ) - for o in metrics - if len(o.samples) > 0 - ]), - content_type='text/plain; version=0.0.4; charset=utf-8', - ) + return django.http.HttpResponse( + ''.join( + [ + '{help}{type}{samples}'.format( + # help='# HELP %s some metric' % o.name, + # type='# TYPE %s counter' % o.name, + help=('# HELP {0} {1}\n'.format(o.name, o.help) if o.help else ''), + type=('# TYPE {0} {1}\n'.format(o.name, o.type) if o.type else ''), + samples=''.join([Metric.sample_serialize(o, s) for s in o.samples]), + ) + for o in metrics + if len(o.samples) > 0 + ] + ), + content_type='text/plain; version=0.0.4; charset=utf-8', + )