function to check login status

src/delta_barth/api/common.py

@@ -1,14 +1,19 @@
 from __future__ import annotations
 
 import re
+from datetime import datetime as Datetime
 from typing import Final, Never
 
 import requests
-from pydantic import BaseModel
+from pydantic import BaseModel, PositiveInt
 from requests import Response
 
-from delta_barth.constants import HTTP_CONTENT_HEADERS, KnownApiErrorCodes
-from delta_barth.errors import UnknownApiErrorCode, UnspecifiedRequestType
+from delta_barth.constants import HTTP_CURRENT_CONNECTION, KnownApiErrorCodes
+from delta_barth.errors import (
+    ApiConnectionError,
+    UnknownApiErrorCode,
+    UnspecifiedRequestType,
+)
 from delta_barth.types import HttpRequestTypes
 
 LOGIN_ERROR_CODES_KNOWN: Final[frozenset[int]] = frozenset((400, 401, 409, 500))
@@ -30,6 +35,11 @@ def _raise_for_unknown_error(
     )
 
 
+def _assert_login() -> None:
+    if not HTTP_CURRENT_CONNECTION.logged_in:
+        raise ApiConnectionError("Curent session is not logged in")
+
+
 def _strip_url_components(string: str) -> str:
     return re.sub(r"^[ /]+|[ /]+$", "", string)
 
@@ -92,7 +102,7 @@ def login(
     resp = requests.put(
         URL,
         login_req.model_dump_json(),
-        headers=HTTP_CONTENT_HEADERS.as_dict(),  # type: ignore
+        headers=HTTP_CURRENT_CONNECTION.as_dict(),  # type: ignore
     )
 
     response: LoginResponse
@@ -100,7 +110,7 @@ def login(
     if resp.status_code == 200:
         # success
         response = LoginResponse(**resp.json())
-        HTTP_CONTENT_HEADERS.add_session_token(response.token)
+        HTTP_CURRENT_CONNECTION.add_session_token(response.token)
     elif resp.status_code in KnownApiErrorCodes.COMMON.value:
         err = DelBarApiError(status_code=resp.status_code, **resp.json())
         response = LoginResponse(token="", error=err)
@@ -123,7 +133,7 @@ def logout(
 
     resp = requests.put(
         URL,
-        headers=HTTP_CONTENT_HEADERS.as_dict(),  # type: ignore
+        headers=HTTP_CURRENT_CONNECTION.as_dict(),  # type: ignore
     )
 
     response: LogoutResponse
@@ -131,7 +141,7 @@ def logout(
     if resp.status_code == 200:
         # success
         response = LogoutResponse()
-        HTTP_CONTENT_HEADERS.remove_session_token()
+        HTTP_CURRENT_CONNECTION.remove_session_token()
     elif resp.status_code in KnownApiErrorCodes.COMMON.value:
         err = DelBarApiError(status_code=resp.status_code, **resp.json())
         response = LogoutResponse(error=err)
@@ -139,3 +149,16 @@ def logout(
         _raise_for_unknown_error(resp)
 
     return response
+
+
+# ** sales data
+class SalesPrognosisRequest(BaseModel):
+    berechnungszeitpunkt: Datetime
+
+
+class SalesPrognosisResponseEntry(BaseModel):
+    artikelID: PositiveInt
+    firmaId: PositiveInt
+    betrag: PositiveInt
+    menge: PositiveInt
+    buchungsDatum: Datetime

src/delta_barth/constants.py

@@ -1,14 +1,14 @@
 import enum
 from typing import Final
 
-from delta_barth.types import CurrentContentHeaders, HttpContentHeaders
+from delta_barth.types import CurrentConnection, HttpContentHeaders
 
 HTTP_BASE_CONTENT_HEADERS: Final[HttpContentHeaders] = {
     "Content-type": "application/json",
     "Accept": "application/json",
 }
 
-HTTP_CONTENT_HEADERS: Final[CurrentContentHeaders] = CurrentContentHeaders(
+HTTP_CURRENT_CONNECTION: Final[CurrentConnection] = CurrentConnection(
     HTTP_BASE_CONTENT_HEADERS
 )
 

src/delta_barth/errors.py

@@ -4,3 +4,7 @@ class UnspecifiedRequestType(Exception):
 
 class UnknownApiErrorCode(Exception):
     """exception raised if for a given request a unknown error response code is transmitted"""
+
+
+class ApiConnectionError(Exception):
+    """exception raised if an established connection is needed, but the current session is not connected"""

src/delta_barth/types.py

@@ -24,13 +24,14 @@ HttpContentHeaders = TypedDict(
 )
 
 
-class CurrentContentHeaders:
+class CurrentConnection:
     def __init__(
         self,
         base_headers: HttpContentHeaders,
     ) -> None:
         self._headers = base_headers
         self._session_token: str | None = None
+        self._logged_in: bool = False
 
     def __getitem__(self, key: str) -> str:
         return self.headers[key]
@@ -49,6 +50,10 @@ class CurrentContentHeaders:
     def session_token(self) -> str | None:
         return self._session_token
 
+    @property
+    def logged_in(self) -> bool:
+        return self._logged_in
+
     def add_session_token(
         self,
         token: str,
@@ -60,11 +65,13 @@ class CurrentContentHeaders:
             )
         self._session_token = token
         self._headers.update(DelecoToken=token)
+        self._logged_in = True
 
     def remove_session_token(self) -> None:
         if "DelecoToken" in self:
             del self._headers["DelecoToken"]
             self._session_token = None
+            self._logged_in = False
 
 
 # ** forecasts

tests/api/test_common.py

@@ -1,8 +1,12 @@
 import pytest
 
 from delta_barth.api import common
-from delta_barth.constants import HTTP_CONTENT_HEADERS
-from delta_barth.errors import UnknownApiErrorCode, UnspecifiedRequestType
+from delta_barth.constants import HTTP_CURRENT_CONNECTION
+from delta_barth.errors import (
+    ApiConnectionError,
+    UnknownApiErrorCode,
+    UnspecifiedRequestType,
+)
 from delta_barth.types import HttpRequestTypes
 
 "http://test.com/ "
@@ -44,6 +48,11 @@ def test_combine_route(base, route, expect):
     assert res == expect
 
 
+def test_assert_login():
+    with pytest.raises(ApiConnectionError):
+        common._assert_login()
+
+
 @pytest.mark.api_con_required
 def test_ping(api_base_url):
     resp = common.ping(api_base_url, HttpRequestTypes.GET)
@@ -66,7 +75,7 @@ def test_raise_unknown_error(api_base_url):
 
 @pytest.mark.api_con_required
 def test_login_logout(credentials, api_base_url):
-    assert HTTP_CONTENT_HEADERS.session_token is None
+    assert HTTP_CURRENT_CONNECTION.session_token is None
     resp = common.login(
         base_url=api_base_url,
         user_name=credentials["user"],
@@ -75,12 +84,12 @@ def test_login_logout(credentials, api_base_url):
         mandant=credentials["mandant"],
     )
     assert resp.error is None
-    assert HTTP_CONTENT_HEADERS.session_token is not None
+    assert HTTP_CURRENT_CONNECTION.session_token is not None
     resp = common.logout(
         base_url=api_base_url,
     )
-    assert HTTP_CONTENT_HEADERS.session_token is None
-    assert "DelecoToken" not in HTTP_CONTENT_HEADERS
+    assert HTTP_CURRENT_CONNECTION.session_token is None
+    assert "DelecoToken" not in HTTP_CURRENT_CONNECTION
     resp = common.login(
         base_url=api_base_url,
         user_name=credentials["user"],