src/delta_barth/api/requests.py

@@ -7,7 +7,7 @@ import requests
 from dopt_basics.io import combine_route
 from pydantic import BaseModel, PositiveInt, SkipValidation
 
-from delta_barth.constants import API_CON_TIMEOUT
+from delta_barth.constants import API_CON_TIMEOUT, MAX_LOGIN_RETRIES
 from delta_barth.errors import STATUS_HANDLER
 from delta_barth.types import DelBarApiError, ExportResponse, ResponseType, Status
 
@@ -71,12 +71,18 @@ def get_sales_prognosis_data(
     )
     empty_response = SalesPrognosisResponse(daten=tuple())
     try:
-        resp = requests.get(
+        for attempt in range(1, (MAX_LOGIN_RETRIES + 1)):
-            URL,
+            resp = requests.get(
-            params=sales_prog_req.model_dump(mode="json", exclude_none=True),
+                URL,
-            headers=session.headers,  # type: ignore[argumentType]
+                params=sales_prog_req.model_dump(mode="json", exclude_none=True),
-            timeout=API_CON_TIMEOUT,
+                headers=session.headers,  # type: ignore[argumentType]
-        )
+                timeout=API_CON_TIMEOUT,
+            )
+            if resp.status_code == 401:
+                _, status = session.relogin()
+                if status != STATUS_HANDLER.SUCCESS and attempt == MAX_LOGIN_RETRIES:
+                    return empty_response, status
+
     except requests.exceptions.Timeout:
         return empty_response, STATUS_HANDLER.pipe_states.CONNECTION_TIMEOUT
     except requests.exceptions.RequestException:

src/delta_barth/constants.py

@@ -51,6 +51,8 @@ class KnownDelBarApiErrorCodes(enum.Enum):
 
 # ** API
 API_CON_TIMEOUT: Final[float] = 10.0  # secs to response
+MAX_LOGIN_RETRIES: Final[int] = 1
+
 # ** API response parsing
 # ** column mapping [API-Response --> Target-Features]
 COL_MAP_SALES_PROGNOSIS: Final[DualDict[str, str]] = DualDict(

src/delta_barth/session.py

@@ -292,6 +292,15 @@ class Session:
 
         return None, status
 
+    def relogin(
+        self,
+    ) -> tuple[LoginResponse, Status]:
+        if self.session_token is None:
+            return self.login()
+
+        self._remove_session_token()
+        return self.login()
+
     def assert_login(
         self,
     ) -> tuple[LoginResponse, Status]: