From 58f0e3dbfa88243eb097ed3a9d9e4be5f61b4131 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Florian=20F=C3=B6rster?= Date: Fri, 7 Mar 2025 07:05:54 +0100 Subject: [PATCH] refactoring --- src/delta_barth/_management.py | 3 ++ src/delta_barth/api/common.py | 59 ++++++++++++++++++++++++++++------ src/delta_barth/constants.py | 6 +--- src/delta_barth/types.py | 42 ------------------------ tests/api/test_common.py | 11 ++++--- 5 files changed, 60 insertions(+), 61 deletions(-) diff --git a/src/delta_barth/_management.py b/src/delta_barth/_management.py index 1efc23c..ae2bbc4 100644 --- a/src/delta_barth/_management.py +++ b/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) diff --git a/src/delta_barth/api/common.py b/src/delta_barth/api/common.py index 484d8f4..ccc3130 100644 --- a/src/delta_barth/api/common.py +++ b/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.constants import HTTP_CURRENT_CONNECTION +from delta_barth._management import CURRENT_SESSION, STATE_HANDLER 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 diff --git a/src/delta_barth/constants.py b/src/delta_barth/constants.py index 91c8bec..029a2e6 100644 --- a/src/delta_barth/constants.py +++ b/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)) diff --git a/src/delta_barth/types.py b/src/delta_barth/types.py index 0c09eaf..f815d2d 100644 --- a/src/delta_barth/types.py +++ b/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: diff --git a/tests/api/test_common.py b/tests/api/test_common.py index 63d8671..f653984 100644 --- a/tests/api/test_common.py +++ b/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 "DelecoToken" not in HTTP_CURRENT_CONNECTION.headers + assert CURRENT_SESSION.session_token is None + assert "DelecoToken" not in CURRENT_SESSION.headers resp, state = common.login( base_url=api_base_url,