From 0042d2367ad07a62664d92a82ae0cfda58a1c333 Mon Sep 17 00:00:00 2001 From: Scott Wallace Date: Sat, 11 Dec 2021 16:37:49 +0000 Subject: [PATCH] Strict typing --- main.py | 36 ++++++++++++++++++++++++++++-------- 1 file changed, 28 insertions(+), 8 deletions(-) diff --git a/main.py b/main.py index bfd42a4..6e2f247 100644 --- a/main.py +++ b/main.py @@ -3,8 +3,8 @@ Present observational weather data to Prometheus """ import json -import time import os +from typing import Dict, List, TypedDict import requests from flask import Flask, Response @@ -12,7 +12,21 @@ from flask import Flask, Response app = Flask(__name__) -def fetch_weather_data(location: int, apikey: str) -> dict: +class PromData(TypedDict): + """ + Describes the data being returned to Prometheus + + Args: + TypedDict ([type]): data to return to Prometheus + """ + + key: str + labels: Dict[str, str] + type: str + value: float + + +def fetch_weather_data(location: int, apikey: str) -> Dict[str, Dict[str, float]]: """ Fetch current data from the Met Office for the provided postcode """ @@ -24,20 +38,26 @@ def fetch_weather_data(location: int, apikey: str) -> dict: f'aqi=yes' ) - return json.loads(obs_data.content) + return dict(json.loads(obs_data.content)) @app.route('/metrics') -def metrics(): +def metrics() -> Response: """ Output Prometheus-style metrics """ apikey = os.environ.get('WEATHER_APIKEY') location = os.environ.get('WEATHER_LOCATION') - latest_data = fetch_weather_data(location, apikey)['current'] + if not location: + return Response( + f'There was a problem finding the weather for location: "{location}"', + status=500, + ) - ret_data = [ + latest_data = fetch_weather_data(int(location), str(apikey))['current'] + + ret_data: List[PromData] = [ { 'key': 'sensor_weather_outdoor_temperature_celsius', 'labels': { @@ -120,12 +140,12 @@ def metrics(): }, ] - ret_strs = list() + ret_strs = [] for item in ret_data: ret_strs.append(f'# HELP {item["key"]} Weather metric') ret_strs.append(f'# TYPE {item["key"]} {item["type"]}') ret_strs.append( - item["key"] + str(item["key"]) + '{' + " ".join([f'{key}="{val}"' for key, val in item["labels"].items()]) + '} '