retrieval of sales prognosis data

src/delta_barth/api/common.py

@@ -5,7 +5,7 @@ from datetime import datetime as Datetime
 from typing import Final, Never
 
 import requests
-from pydantic import BaseModel, PositiveInt
+from pydantic import BaseModel, PositiveFloat, PositiveInt
 from requests import Response
 
 from delta_barth.constants import HTTP_CURRENT_CONNECTION, KnownApiErrorCodes
@@ -21,9 +21,13 @@ LOGIN_ERROR_CODES_KNOWN: Final[frozenset[int]] = frozenset((400, 401, 409, 500))
 
 class DelBarApiError(BaseModel):
     status_code: int
-    message: str
+    message: str = ""
-    code: str | None
+    code: str | None = None
-    hints: str | None
+    hints: str | None = None
+    type: str | None = None
+    title: str | None = None
+    errors: dict | None = None
+    traceId: str | None = None
 
 
 def _raise_for_unknown_error(
@@ -35,7 +39,7 @@ def _raise_for_unknown_error(
     )
 
 
-def _assert_login() -> None:
+def _assert_login_status() -> None:
     if not HTTP_CURRENT_CONNECTION.logged_in:
         raise ApiConnectionError("Curent session is not logged in")
 
@@ -102,13 +106,11 @@ def login(
     resp = requests.put(
         URL,
         login_req.model_dump_json(),
-        headers=HTTP_CURRENT_CONNECTION.as_dict(),  # type: ignore
+        headers=HTTP_CURRENT_CONNECTION.headers,  # type: ignore
     )
 
     response: LoginResponse
-
     if resp.status_code == 200:
-        # success
         response = LoginResponse(**resp.json())
         HTTP_CURRENT_CONNECTION.add_session_token(response.token)
     elif resp.status_code in KnownApiErrorCodes.COMMON.value:
@@ -133,13 +135,11 @@ def logout(
 
     resp = requests.put(
         URL,
-        headers=HTTP_CURRENT_CONNECTION.as_dict(),  # type: ignore
+        headers=HTTP_CURRENT_CONNECTION.headers,  # type: ignore
     )
 
     response: LogoutResponse
-
     if resp.status_code == 200:
-        # success
         response = LogoutResponse()
         HTTP_CURRENT_CONNECTION.remove_session_token()
     elif resp.status_code in KnownApiErrorCodes.COMMON.value:
@@ -157,8 +157,42 @@ class SalesPrognosisRequest(BaseModel):
 
 
 class SalesPrognosisResponseEntry(BaseModel):
-    artikelID: PositiveInt
+    artikelId: PositiveInt
     firmaId: PositiveInt
-    betrag: PositiveInt
+    betrag: float  # negative values are filtered out later
-    menge: PositiveInt
+    menge: float  # reasons for negative values unknown
     buchungsDatum: Datetime
+
+
+class SalesPrognosisResponse(BaseModel):
+    daten: tuple[SalesPrognosisResponseEntry, ...]
+    error: DelBarApiError | None = None
+
+
+def get_sales_prognosis_data(
+    base_url: str,
+    start_date: Datetime,
+) -> SalesPrognosisResponse:
+    _assert_login_status()
+    ROUTE: Final[str] = "verkauf/umsatzprognosedaten"
+    URL: Final = combine_route(base_url, ROUTE)
+
+    sales_prog_req = SalesPrognosisRequest(
+        berechnungszeitpunkt=start_date,
+    )
+    resp = requests.get(
+        URL,
+        params=sales_prog_req.model_dump(mode="json"),
+        headers=HTTP_CURRENT_CONNECTION.headers,  # type: ignore[argumentType]
+    )
+
+    response: SalesPrognosisResponse
+    if resp.status_code == 200:
+        response = SalesPrognosisResponse(**resp.json())
+    elif resp.status_code in KnownApiErrorCodes.COMMON.value:  # pragma: no cover
+        err = DelBarApiError(status_code=resp.status_code, **resp.json())
+        response = SalesPrognosisResponse(daten=tuple(), error=err)
+    else:  # pragma: no cover
+        _raise_for_unknown_error(resp)
+
+    return response

tests/api/test_common.py

@@ -1,3 +1,5 @@
+from datetime import datetime as Datetime
+
 import pytest
 
 from delta_barth.api import common
@@ -50,7 +52,7 @@ def test_combine_route(base, route, expect):
 
 def test_assert_login():
     with pytest.raises(ApiConnectionError):
-        common._assert_login()
+        common._assert_login_status()
 
 
 @pytest.mark.api_con_required
@@ -88,8 +90,9 @@ def test_login_logout(credentials, api_base_url):
     resp = common.logout(
         base_url=api_base_url,
     )
+    assert resp.error is None
     assert HTTP_CURRENT_CONNECTION.session_token is None
-    assert "DelecoToken" not in HTTP_CURRENT_CONNECTION
+    assert "DelecoToken" not in HTTP_CURRENT_CONNECTION.headers
     resp = common.login(
         base_url=api_base_url,
         user_name=credentials["user"],
@@ -100,3 +103,28 @@ def test_login_logout(credentials, api_base_url):
     assert resp.error is not None
     assert resp.error.status_code == 409
     assert resp.error.message == "Nutzer oder Passwort falsch."
+
+
+@pytest.mark.api_con_required
+def test_get_sales_prognosis_data(credentials, api_base_url):
+    resp = common.login(
+        base_url=api_base_url,
+        user_name=credentials["user"],
+        password=credentials["pwd"],
+        database=credentials["db"],
+        mandant=credentials["mandant"],
+    )
+    assert resp.error is None
+    date = Datetime(2022, 6, 1)
+    resp = common.get_sales_prognosis_data(api_base_url, date)
+    assert resp.error is None
+    assert len(resp.daten) > 0
+    date = Datetime(2030, 1, 1)
+    resp = common.get_sales_prognosis_data(api_base_url, date)
+    assert resp.error is None
+    assert len(resp.daten) == 0
+    # close connection
+    resp = common.logout(
+        base_url=api_base_url,
+    )
+    assert resp.error is None