delta-barth/delta-barth-py
1
0
Fork 0
Code Issues 3 Pull Requests Actions Packages Projects 3 Releases 3 Wiki Activity

apply new state handling to API functions

Browse Source
This commit is contained in:
Florian Förster 2025-03-05 15:50:42 +01:00
parent 7b82d051e2
commit b4b4181bd0
3 changed files with 72 additions and 74 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

70
src/delta_barth/api/common.py
View File

@ -2,32 +2,26 @@ from __future__ import annotations
import re
from datetime import datetime as Datetime
from typing import Final, Never
from typing import TYPE_CHECKING, Final, Never
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, KnownDelBarApiErrorCodes
from delta_barth.errors import (
ApiConnectionError,
UnknownApiErrorCode,
UnspecifiedRequestType,
)
from delta_barth.types import DelBarApiError, HttpRequestTypes
if TYPE_CHECKING:
from delta_barth.types import Status
LOGIN_ERROR_CODES_KNOWN: Final[frozenset[int]] = frozenset((400, 401, 409, 500))
def _raise_for_unknown_error(
resp: Response,
) -> Never:
raise UnknownApiErrorCode(
f"Unknown response code for request. Status Code: {resp.status_code}, "
f"Content: {resp.text}"
)
def _assert_login_status() -> None:
if not HTTP_CURRENT_CONNECTION.logged_in:
raise ApiConnectionError("Curent session is not logged in")
@ -73,7 +67,6 @@ class LoginRequest(BaseModel):
class LoginResponse(BaseModel):
token: str
error: DelBarApiError | None = None
def login(
@ -82,7 +75,7 @@ def login(
password: str,
database: str,
mandant: str,
) -> LoginResponse:
) -> tuple[LoginResponse, Status]:
ROUTE: Final[str] = "user/login"
URL: Final = combine_route(base_url, ROUTE)
@ -99,26 +92,23 @@ def login(
)
response: LoginResponse
status: Status
if resp.status_code == 200:
response = LoginResponse(**resp.json())
status = STATE_HANDLER.pipe_states.SUCCESS
HTTP_CURRENT_CONNECTION.add_session_token(response.token)
elif resp.status_code in KnownDelBarApiErrorCodes.COMMON.value:
else:
response = LoginResponse(token="")
err = DelBarApiError(status_code=resp.status_code, **resp.json())
response = LoginResponse(token="", error=err)
else: # pragma: no cover
_raise_for_unknown_error(resp)
status = STATE_HANDLER.api_error(err)
return response
return response, status
# ** logout
class LogoutResponse(BaseModel):
error: DelBarApiError | None = None
def logout(
base_url: str,
) -> LogoutResponse:
) -> tuple[None, Status]:
ROUTE: Final[str] = "user/logout"
URL: Final = combine_route(base_url, ROUTE)
@ -127,17 +117,16 @@ def logout(
headers=HTTP_CURRENT_CONNECTION.headers, # type: ignore
)
response: LogoutResponse
response = None
status: Status
if resp.status_code == 200:
response = LogoutResponse()
status = STATE_HANDLER.SUCCESS
HTTP_CURRENT_CONNECTION.remove_session_token()
elif resp.status_code in KnownDelBarApiErrorCodes.COMMON.value:
else:
err = DelBarApiError(status_code=resp.status_code, **resp.json())
response = LogoutResponse(error=err)
else: # pragma: no cover
_raise_for_unknown_error(resp)
status = STATE_HANDLER.api_error(err)
return response
return response, status
# ** sales data
@ -156,14 +145,14 @@ class SalesPrognosisResponseEntry(BaseModel):
class SalesPrognosisResponse(BaseModel):
daten: tuple[SalesPrognosisResponseEntry, ...]
error: DelBarApiError | None = None
# error: DelBarApiError | None = None
def get_sales_prognosis_data(
base_url: str,
company_id: int | None = None,
start_date: Datetime | None = None,
) -> SalesPrognosisResponse:
) -> tuple[SalesPrognosisResponse, Status]:
_assert_login_status()
ROUTE: Final[str] = "verkauf/umsatzprognosedaten"
URL: Final = combine_route(base_url, ROUTE)
@ -179,12 +168,19 @@ def get_sales_prognosis_data(
)
response: SalesPrognosisResponse
status: Status
if resp.status_code == 200:
response = SalesPrognosisResponse(**resp.json())
elif resp.status_code in KnownDelBarApiErrorCodes.COMMON.value: # pragma: no cover
status = STATE_HANDLER.SUCCESS
else:
response = SalesPrognosisResponse(daten=tuple())
err = DelBarApiError(status_code=resp.status_code, **resp.json())
response = SalesPrognosisResponse(daten=tuple(), error=err)
else: # pragma: no cover
_raise_for_unknown_error(resp)
status = STATE_HANDLER.api_error(err)
return response
# elif resp.status_code in KnownDelBarApiErrorCodes.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, status

4
src/delta_barth/errors.py
View File

@ -13,10 +13,6 @@ class UnspecifiedRequestType(Exception):
"""exception raised if for a given API endpoint a not defined operation is requested"""
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"""

72
tests/api/test_common.py
View File

@ -3,10 +3,9 @@ from datetime import datetime as Datetime
import pytest
from delta_barth.api import common
from delta_barth.constants import HTTP_CURRENT_CONNECTION
from delta_barth.constants import DEFAULT_API_ERR_CODE, HTTP_CURRENT_CONNECTION
from delta_barth.errors import (
ApiConnectionError,
UnknownApiErrorCode,
UnspecifiedRequestType,
)
from delta_barth.types import HttpRequestTypes
@ -68,46 +67,43 @@ def test_ping(api_base_url):
resp = common.ping(api_base_url, HttpRequestTypes.POST)
@pytest.mark.api_con_required
def test_raise_unknown_error(api_base_url):
resp = common.ping(api_base_url, HttpRequestTypes.GET)
with pytest.raises(UnknownApiErrorCode):
common._raise_for_unknown_error(resp)
@pytest.mark.api_con_required
def test_login_logout(credentials, api_base_url):
assert HTTP_CURRENT_CONNECTION.session_token is None
resp = common.login(
resp, state = 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
assert state.code == 0
assert HTTP_CURRENT_CONNECTION.session_token is not None
resp = common.logout(
resp, state = common.logout(
base_url=api_base_url,
)
assert resp.error is None
assert resp is None
assert state.code == 0
assert HTTP_CURRENT_CONNECTION.session_token is None
assert "DelecoToken" not in HTTP_CURRENT_CONNECTION.headers
resp = common.login(
resp, state = common.login(
base_url=api_base_url,
user_name=credentials["user"],
password="WRONG_PASSWORD",
database=credentials["db"],
mandant=credentials["mandant"],
)
assert resp.error is not None
assert resp.error.status_code == 409
assert resp.error.message == "Nutzer oder Passwort falsch."
assert resp is not None
assert state.code == DEFAULT_API_ERR_CODE
assert state.api_server_error is not None
assert state.api_server_error.status_code == 409
assert state.api_server_error.message == "Nutzer oder Passwort falsch."
@pytest.mark.api_con_required
def test_get_sales_prognosis_data(credentials, api_base_url):
resp = common.login(
resp, state = common.login(
base_url=api_base_url,
user_name=credentials["user"],
password=credentials["pwd"],
@ -115,35 +111,45 @@ def test_get_sales_prognosis_data(credentials, api_base_url):
mandant=credentials["mandant"],
)
# test without company ID
assert resp.error is None
assert state.code == 0
date = Datetime(2022, 6, 1)
resp = common.get_sales_prognosis_data(api_base_url, None, date)
assert resp.error is None
resp, state = common.get_sales_prognosis_data(api_base_url, None, date)
assert state.code == 0
assert len(resp.daten) > 0
date = Datetime(2030, 1, 1)
resp = common.get_sales_prognosis_data(api_base_url, None, date)
assert resp.error is None
resp, state = common.get_sales_prognosis_data(api_base_url, None, date)
assert state.code == 0
assert len(resp.daten) == 0
# test with company ID
assert resp.error is None
assert state.code == 0
date = Datetime(2022, 6, 1)
company_id = 1024
resp = common.get_sales_prognosis_data(api_base_url, company_id, date)
assert resp.error is None
resp, state = common.get_sales_prognosis_data(api_base_url, company_id, date)
assert state.code == 0
assert len(resp.daten) > 0
date = Datetime(2030, 1, 1)
resp = common.get_sales_prognosis_data(api_base_url, company_id, date)
assert resp.error is None
resp, state = common.get_sales_prognosis_data(api_base_url, company_id, date)
assert state.code == 0
assert len(resp.daten) == 0
# test with non-existent company ID
assert resp.error is None
assert state.code == 0
date = Datetime(2022, 6, 1)
company_id = 1000024
resp = common.get_sales_prognosis_data(api_base_url, company_id, date)
assert resp.error is None
resp, state = common.get_sales_prognosis_data(api_base_url, company_id, date)
# TODO check if this behaviour is still considered "successful"
assert state.code == 0
assert len(resp.daten) == 0
# test without date
company_id = 1024
resp, state = common.get_sales_prognosis_data(api_base_url, company_id, None)
assert state.code == 0
assert len(resp.daten) > 0
# test without filters
resp, state = common.get_sales_prognosis_data(api_base_url, None, None)
assert state.code == 0
assert len(resp.daten) > 0
# close connection
resp = common.logout(
resp, state = common.logout(
base_url=api_base_url,
)
assert resp.error is None
assert state.code == 0