Skip to content
Snippets Groups Projects
Commit f6fd65fc authored by Erik Reid's avatar Erik Reid
Browse files

typehinting fixes, there's apparnetly no stub for influxdb v1 - ignoring

parent ff66f379
No related branches found
No related tags found
No related merge requests found
import os
import contextlib
from influxdb import InfluxDBClient
from threading import Event
import time
from typing import Generator, Any
import logging
......@@ -59,7 +59,7 @@ CACHED_SCID_RATES_SCHEMA = {
}
@contextlib.contextmanager
def influx_client(influx_params: config.InfluxConnectionParams):
def influx_client(influx_params: config.InfluxConnectionParams) -> Generator[InfluxDBClient]:
"""
context manager for creating an influx db connection
......@@ -87,7 +87,9 @@ def influx_client(influx_params: config.InfluxConnectionParams):
client.close()
def _load_scid_rates_rows(influx_params: config.InfluxConnectionParams, window: str = '24h'):
def _load_scid_rates_rows(
influx_params: config.InfluxConnectionParams,
window: str = '24h') -> Generator[dict[str, str|float|None], None, None]:
"""
Return the count of all fields, grouped by
hostname & interface_name.
......@@ -127,14 +129,16 @@ def _load_scid_rates_rows(influx_params: config.InfluxConnectionParams, window:
yield rate
def load_scid_rates(influx_params: config.InfluxConnectionParams):
def load_scid_rates(influx_params: config.InfluxConnectionParams) -> list[dict[str, Any]]:
rates = []
for r in _load_scid_rates_rows(influx_params):
def _bitrate_or_none(field_name: str) -> float | None:
if field_name in r:
return r[field_name] * 8
return None
_rate = r.get(field_name, None)
assert isinstance(_rate, float | None) # mypy noise
if _rate:
return _rate * 8
return _rate # could be 0 or None
values = {
'latest': {
......
......@@ -22,9 +22,9 @@ def init(cache_dir: str) -> None:
logger.debug(f"set cache directory: {_cache_dir}")
def set(filename: str, data: dict[str, Any]) -> None:
def set(filename: str, data: Any) -> None:
"""
data must be a JSON-serializable dict.
data must be JSON-serializable.
"""
assert _cache_dir is not None, "cache_dir hasn't been initialized"
with open(os.path.join(_cache_dir, filename), 'w') as f:
......
......@@ -11,6 +11,10 @@ strict = true
warn_unused_ignores = true
warn_return_any = true
[[tool.mypy.overrides]]
module = ["influxdb"]
ignore_missing_imports = true
[tool.coverage.run]
source = ["mapping_provider"]
omit = [
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment