Fix HE without flow (add FetchError class)

CHANGELOG.md

@@ -1,3 +1,7 @@
+## 1.0.10 (2026-02-17)
+### Fixed
+- Fix HE without flow (add FetchError class)
+
 ## 1.0.9 (2025-12-18)
 ### Fixed
 - Fix SH pegel page scraping

LICENSE

@@ -1,6 +1,6 @@
 MIT License
 
-Copyright (c) 2025 stephan192
+Copyright (c) 2026 stephan192
 
 Permission is hereby granted, free of charge, to any person obtaining a copy
 of this software and associated documentation files (the "Software"), to deal

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "lhpapi"
-version = "1.0.9"
+version = "1.0.10"
 authors = [
     { name="stephan192", email="stephan192@outlook.com" },
 ]

src/lhpapi/api_utils.py

@@ -4,11 +4,13 @@
 
 from dataclasses import dataclass
 from datetime import UTC, datetime
+from json import JSONDecodeError
 from typing import Any
 from zoneinfo import ZoneInfo
 
 from bs4 import BeautifulSoup
 from requests import get
+from requests.exceptions import RequestException
 
 
 class LHPError(Exception):
@@ -22,6 +24,10 @@ def __init__(self, exception: Exception | str, location: str) -> None:
             super().__init__(f"{location}: {exception.__class__.__name__}: {exception}")
 
 
+class FetchError(Exception):
+    """ "Exception occurred while fetching data via requests."""
+
+
 @dataclass
 class StaticData:
     """Class containing the static data."""
@@ -46,22 +52,28 @@ class DynamicData:
 
 def fetch_json(url: str, timeout: float = 10.0) -> Any:
     """Fetch data via json."""
-    response = get(url=url, timeout=timeout)
-    response.raise_for_status()
-    json_data = response.json()
-    return json_data
+    try:
+        response = get(url=url, timeout=timeout)
+        response.raise_for_status()
+        json_data = response.json()
+        return json_data
+    except (RequestException, JSONDecodeError) as err:
+        raise FetchError(err) from err
 
 
 def fetch_text(url: str, timeout: float = 10.0, forced_encoding: str = None) -> str:
     """Fetch data via text."""
-    response = get(url=url, timeout=timeout)
-    if forced_encoding is not None:
-        response.encoding = forced_encoding
-    else:
-        # Override encoding by real educated guess (required for e.g. BW)
-        response.encoding = response.apparent_encoding
-    response.raise_for_status()
-    return response.text
+    try:
+        response = get(url=url, timeout=timeout)
+        if forced_encoding is not None:
+            response.encoding = forced_encoding
+        else:
+            # Override encoding by real educated guess (required for e.g. BW)
+            response.encoding = response.apparent_encoding
+        response.raise_for_status()
+        return response.text
+    except RequestException as err:
+        raise FetchError(err) from err
 
 
 def fetch_soup(

src/lhpapi/he_api.py

@@ -4,6 +4,7 @@
 
 from .api_utils import (
     DynamicData,
+    FetchError,
     LHPError,
     StaticData,
     calc_stage,
@@ -51,22 +52,25 @@ def get_basic_station_data(ident: str) -> tuple[str, str, str, str]:
 def get_stage_levels(internal_url: str) -> list[float]:
     """Get stage levels."""
     stage_levels = [None] * 4
-    alarmlevels = fetch_json(internal_url + "/W/alarmlevel.json")
-    for station_data in alarmlevels:
-        if (
-            "ts_name" in station_data
-            and "data" in station_data
-            and isinstance(station_data["data"], list)
-            and len(station_data["data"]) > 0
-        ):
-            # Check if ts_name is one of the desired values
-            if station_data["ts_name"] == "Meldestufe1":
-                stage_levels[0] = convert_to_float(station_data["data"][-1][1])
-            elif station_data["ts_name"] == "Meldestufe2":
-                stage_levels[1] = convert_to_float(station_data["data"][-1][1])
-            # No equivalent to stage_levels[2] available
-            elif station_data["ts_name"] == "Meldestufe3":
-                stage_levels[3] = convert_to_float(station_data["data"][-1][1])
+    try:
+        alarmlevels = fetch_json(internal_url + "/W/alarmlevel.json")
+        for station_data in alarmlevels:
+            if (
+                "ts_name" in station_data
+                and "data" in station_data
+                and isinstance(station_data["data"], list)
+                and len(station_data["data"]) > 0
+            ):
+                # Check if ts_name is one of the desired values
+                if station_data["ts_name"] == "Meldestufe1":
+                    stage_levels[0] = convert_to_float(station_data["data"][-1][1])
+                elif station_data["ts_name"] == "Meldestufe2":
+                    stage_levels[1] = convert_to_float(station_data["data"][-1][1])
+                # No equivalent to stage_levels[2] available
+                elif station_data["ts_name"] == "Meldestufe3":
+                    stage_levels[3] = convert_to_float(station_data["data"][-1][1])
+    except FetchError:
+        pass
     return stage_levels
 
 
@@ -107,7 +111,7 @@ def update_HE(static_data: StaticData) -> DynamicData:  # pylint: disable=invali
                     level = convert_to_float(dataset["data"][-1][1])
                     stage = calc_stage(level, static_data.stage_levels)
                     break
-        except (IndexError, KeyError):
+        except (IndexError, KeyError, FetchError):
             level = None
             stage = None
 
@@ -122,7 +126,7 @@ def update_HE(static_data: StaticData) -> DynamicData:  # pylint: disable=invali
                     last_update_str_q = dataset["data"][-1][0]
                     flow = convert_to_float(dataset["data"][-1][1])
                     break
-        except (IndexError, KeyError):
+        except (IndexError, KeyError, FetchError):
             flow = None
 
         last_update = None

src/lhpapi/nw_api.py

@@ -4,6 +4,7 @@
 
 from .api_utils import (
     DynamicData,
+    FetchError,
     LHPError,
     StaticData,
     calc_stage,
@@ -45,36 +46,42 @@ def get_stage_levels(internal_url: str) -> list[float]:
     """Get stage levels."""
     stage_levels = [None] * 4
     nw_stages = fetch_json(internal_url + "/S/alarmlevel.json")
-    for station_data in nw_stages:
-        # Unfortunately the source data seems quite incomplete.
-        # So we check if the required keys are present in the station_data dictionary:
-        if (
-            "ts_name" in station_data
-            and "data" in station_data
-            and isinstance(station_data["data"], list)
-            and len(station_data["data"]) > 0
-        ):
-            # Check if ts_name is one of the desired values
-            if station_data["ts_name"] == "W.Informationswert_1":
-                stage_levels[0] = convert_to_float(station_data["data"][-1][1])
-            elif station_data["ts_name"] == "W.Informationswert_2":
-                stage_levels[1] = convert_to_float(station_data["data"][-1][1])
-            elif station_data["ts_name"] == "W.Informationswert_3":
-                stage_levels[2] = convert_to_float(station_data["data"][-1][1])
+    try:
+        for station_data in nw_stages:
+            # Unfortunately the source data seems quite incomplete.
+            # So we check if the required keys are present in the station_data dictionary:
+            if (
+                "ts_name" in station_data
+                and "data" in station_data
+                and isinstance(station_data["data"], list)
+                and len(station_data["data"]) > 0
+            ):
+                # Check if ts_name is one of the desired values
+                if station_data["ts_name"] == "W.Informationswert_1":
+                    stage_levels[0] = convert_to_float(station_data["data"][-1][1])
+                elif station_data["ts_name"] == "W.Informationswert_2":
+                    stage_levels[1] = convert_to_float(station_data["data"][-1][1])
+                elif station_data["ts_name"] == "W.Informationswert_3":
+                    stage_levels[2] = convert_to_float(station_data["data"][-1][1])
+    except FetchError:
+        pass
     return stage_levels
 
 
 def get_hint(internal_url: str) -> str:
     """Get hint."""
     hint = None
-    data = fetch_json(internal_url + "/S/week.json")
-    if len(data[0]["AdminStatus"].strip()) > 0:
-        hint = data[0]["AdminStatus"].strip()
-    if len(data[0]["AdminBemerkung"].strip()) > 0:
-        if len(hint) > 0:
-            hint += " / " + data[0]["AdminBemerkung"].strip()
-        else:
-            hint = data[0]["AdminBemerkung"].strip()
+    try:
+        data = fetch_json(internal_url + "/S/week.json")
+        if len(data[0]["AdminStatus"].strip()) > 0:
+            hint = data[0]["AdminStatus"].strip()
+        if len(data[0]["AdminBemerkung"].strip()) > 0:
+            if len(hint) > 0:
+                hint += " / " + data[0]["AdminBemerkung"].strip()
+            else:
+                hint = data[0]["AdminBemerkung"].strip()
+    except FetchError:
+        pass
     return hint
 
 

src/lhpapi/st_api.py

@@ -2,10 +2,9 @@
 
 from __future__ import annotations
 
-import requests
-
 from .api_utils import (
     DynamicData,
+    FetchError,
     LHPError,
     StaticData,
     calc_stage,
@@ -70,8 +69,7 @@ def get_stage_levels(internal_url: str) -> list[float]:
                     stage_levels[2] = convert_to_float(station_data["data"][-1][1])
                 elif station_data["ts_name"] == "Alarmstufe 4":
                     stage_levels[3] = convert_to_float(station_data["data"][-1][1])
-    # eg, 502180/W/alarmlevel.json does not exist (404)
-    except requests.exceptions.HTTPError:
+    except FetchError:
         pass
     return stage_levels
 
@@ -108,8 +106,7 @@ def update_ST(static_data: StaticData) -> DynamicData:  # pylint: disable=invali
             last_update_str_w = data[0]["data"][-1][0]
             level = convert_to_float(data[0]["data"][-1][1])
             stage = calc_stage(level, static_data.stage_levels)
-        # requests.exceptions.HTTPError for handling 404 etc
-        except (IndexError, KeyError, requests.exceptions.HTTPError):
+        except (IndexError, KeyError, FetchError):
             level = None
             stage = None
 
@@ -120,8 +117,7 @@ def update_ST(static_data: StaticData) -> DynamicData:  # pylint: disable=invali
             # Parse data
             last_update_str_q = data[0]["data"][-1][0]
             flow = convert_to_float(data[0]["data"][-1][1])
-        # requests.exceptions.HTTPError for handling 404 etc
-        except (IndexError, KeyError, requests.exceptions.HTTPError):
+        except (IndexError, KeyError, FetchError):
             flow = None
 
         last_update = None