refactoring

src/delta_barth/_management.py

@@ -1,5 +1,8 @@
 from typing import Final
 
+from delta_barth.api.common import Session
+from delta_barth.constants import HTTP_BASE_CONTENT_HEADERS
 from delta_barth.errors import StateHandler
 
 STATE_HANDLER: Final[StateHandler] = StateHandler()
+CURRENT_SESSION: Final[Session] = Session(HTTP_BASE_CONTENT_HEADERS)

src/delta_barth/api/common.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import re
+import warnings
 from datetime import datetime as Datetime
 from typing import TYPE_CHECKING, Final
 
@@ -8,13 +9,12 @@ import requests
 from pydantic import BaseModel, PositiveInt, SkipValidation
 from requests import Response
 
-from delta_barth._management import STATE_HANDLER
+from delta_barth._management import CURRENT_SESSION, STATE_HANDLER
-from delta_barth.constants import HTTP_CURRENT_CONNECTION
 from delta_barth.errors import (
     ApiConnectionError,
     UnspecifiedRequestType,
 )
-from delta_barth.types import DelBarApiError, HttpRequestTypes
+from delta_barth.types import DelBarApiError, HttpContentHeaders, HttpRequestTypes
 
 if TYPE_CHECKING:
     from delta_barth.types import Status
@@ -22,8 +22,49 @@ if TYPE_CHECKING:
 LOGIN_ERROR_CODES_KNOWN: Final[frozenset[int]] = frozenset((400, 401, 409, 500))
 
 
+class Session:
+    def __init__(
+        self,
+        base_headers: HttpContentHeaders,
+    ) -> None:
+        self._headers = base_headers
+        self._session_token: str | None = None
+        self._logged_in: bool = False
+
+    @property
+    def headers(self) -> HttpContentHeaders:
+        return self._headers
+
+    @property
+    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,
+    ) -> None:
+        if self.session_token is not None:
+            warnings.warn(
+                "Setting new API session token despite it was already set. "
+                "Overwriting existing token."
+            )
+        self._session_token = token
+        self._headers.update(DelecoToken=token)
+        self._logged_in = True
+
+    def remove_session_token(self) -> None:
+        if "DelecoToken" in self.headers:
+            del self._headers["DelecoToken"]
+            self._session_token = None
+            self._logged_in = False
+
+
 def _assert_login_status() -> None:
-    if not HTTP_CURRENT_CONNECTION.logged_in:
+    if not CURRENT_SESSION.logged_in:
         raise ApiConnectionError("Curent session is not logged in")
 
 
@@ -88,7 +129,7 @@ def login(
     resp = requests.put(
         URL,
         login_req.model_dump_json(),
-        headers=HTTP_CURRENT_CONNECTION.headers,  # type: ignore
+        headers=CURRENT_SESSION.headers,  # type: ignore
     )
 
     response: LoginResponse
@@ -96,7 +137,7 @@ def login(
     if resp.status_code == 200:
         response = LoginResponse(**resp.json())
         status = STATE_HANDLER.pipe_states.SUCCESS
-        HTTP_CURRENT_CONNECTION.add_session_token(response.token)
+        CURRENT_SESSION.add_session_token(response.token)
     else:
         response = LoginResponse(token="")
         err = DelBarApiError(status_code=resp.status_code, **resp.json())
@@ -114,14 +155,14 @@ def logout(
 
     resp = requests.put(
         URL,
-        headers=HTTP_CURRENT_CONNECTION.headers,  # type: ignore
+        headers=CURRENT_SESSION.headers,  # type: ignore
     )
 
     response = None
     status: Status
     if resp.status_code == 200:
         status = STATE_HANDLER.SUCCESS
-        HTTP_CURRENT_CONNECTION.remove_session_token()
+        CURRENT_SESSION.remove_session_token()
     else:
         err = DelBarApiError(status_code=resp.status_code, **resp.json())
         status = STATE_HANDLER.api_error(err)
@@ -164,7 +205,7 @@ def get_sales_prognosis_data(
     resp = requests.get(
         URL,
         params=sales_prog_req.model_dump(mode="json", exclude_none=True),
-        headers=HTTP_CURRENT_CONNECTION.headers,  # type: ignore[argumentType]
+        headers=CURRENT_SESSION.headers,  # type: ignore[argumentType]
     )
 
     response: SalesPrognosisResponse

src/delta_barth/constants.py

@@ -1,7 +1,7 @@
 import enum
 from typing import Final
 
-from delta_barth.types import CurrentConnection, HttpContentHeaders
+from delta_barth.types import HttpContentHeaders
 
 # ** error handling
 DEFAULT_INTERNAL_ERR_CODE: Final[int] = 100
@@ -12,10 +12,6 @@ HTTP_BASE_CONTENT_HEADERS: Final[HttpContentHeaders] = {
     "Accept": "application/json",
 }
 
-HTTP_CURRENT_CONNECTION: Final[CurrentConnection] = CurrentConnection(
-    HTTP_BASE_CONTENT_HEADERS
-)
-
 
 class KnownDelBarApiErrorCodes(enum.Enum):
     COMMON = frozenset((400, 401, 409, 500))

src/delta_barth/types.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import enum
-import warnings
 from dataclasses import dataclass, field
 from typing import NotRequired, TypeAlias, TypedDict
 
@@ -61,47 +60,6 @@ HttpContentHeaders = TypedDict(
 )
 
 
-class CurrentConnection:
-    def __init__(
-        self,
-        base_headers: HttpContentHeaders,
-    ) -> None:
-        self._headers = base_headers
-        self._session_token: str | None = None
-        self._logged_in: bool = False
-
-    @property
-    def headers(self) -> HttpContentHeaders:
-        return self._headers
-
-    @property
-    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,
-    ) -> None:
-        if self.session_token is not None:
-            warnings.warn(
-                "Setting new API session token despite it was already set. "
-                "Overwriting existing token."
-            )
-        self._session_token = token
-        self._headers.update(DelecoToken=token)
-        self._logged_in = True
-
-    def remove_session_token(self) -> None:
-        if "DelecoToken" in self.headers:
-            del self._headers["DelecoToken"]
-            self._session_token = None
-            self._logged_in = False
-
-
 # ** forecasts
 @dataclass(slots=True)
 class CustomerDataSalesForecast:

tests/api/test_common.py

@@ -2,8 +2,9 @@ from datetime import datetime as Datetime
 
 import pytest
 
+from delta_barth._management import CURRENT_SESSION
 from delta_barth.api import common
-from delta_barth.constants import DEFAULT_API_ERR_CODE, HTTP_CURRENT_CONNECTION
+from delta_barth.constants import DEFAULT_API_ERR_CODE
 from delta_barth.errors import (
     ApiConnectionError,
     UnspecifiedRequestType,
@@ -69,7 +70,7 @@ def test_ping(api_base_url):
 
 @pytest.mark.api_con_required
 def test_login_logout(credentials, api_base_url):
-    assert HTTP_CURRENT_CONNECTION.session_token is None
+    assert CURRENT_SESSION.session_token is None
     resp, state = common.login(
         base_url=api_base_url,
         user_name=credentials["user"],
@@ -78,14 +79,14 @@ def test_login_logout(credentials, api_base_url):
         mandant=credentials["mandant"],
     )
     assert state.code == 0
-    assert HTTP_CURRENT_CONNECTION.session_token is not None
+    assert CURRENT_SESSION.session_token is not None
     resp, state = common.logout(
         base_url=api_base_url,
     )
     assert resp is None
     assert state.code == 0
-    assert HTTP_CURRENT_CONNECTION.session_token is None
+    assert CURRENT_SESSION.session_token is None
-    assert "DelecoToken" not in HTTP_CURRENT_CONNECTION.headers
+    assert "DelecoToken" not in CURRENT_SESSION.headers
 
     resp, state = common.login(
         base_url=api_base_url,