include login assertion in requests to eliminate unnecessary calls to the API

This commit was merged in pull request #27.
This commit is contained in:
2025-04-30 08:42:44 +02:00
parent 453490c0f5
commit 248b811786
4 changed files with 49 additions and 127 deletions

View File

@@ -12,6 +12,8 @@ from delta_barth.errors import STATUS_HANDLER
from delta_barth.types import DelBarApiError, ExportResponse, ResponseType, Status
if TYPE_CHECKING:
from requests import Response
from delta_barth.session import Session
@@ -57,10 +59,10 @@ def get_sales_prognosis_data(
start_date: Datetime | None = None,
) -> tuple[SalesPrognosisResponse, Status]:
# TODO check elimination of assertion for login, #25
_, status = session.assert_login()
if status != STATUS_HANDLER.SUCCESS:
response = SalesPrognosisResponse(daten=tuple())
return response, status
# _, status = session.assert_login()
# if status != STATUS_HANDLER.SUCCESS:
# response = SalesPrognosisResponse(daten=tuple())
# return response, status
ROUTE: Final[str] = "verkauf/umsatzprognosedaten"
URL: Final = combine_route(session.base_url, ROUTE)
@@ -70,6 +72,7 @@ def get_sales_prognosis_data(
BuchungsDatum=start_date,
)
empty_response = SalesPrognosisResponse(daten=tuple())
resp: Response | None = None
try:
for attempt in range(1, (MAX_LOGIN_RETRIES + 1)):
resp = requests.get(
@@ -82,7 +85,6 @@ def get_sales_prognosis_data(
_, status = session.relogin()
if status != STATUS_HANDLER.SUCCESS and attempt == MAX_LOGIN_RETRIES:
return empty_response, status
except requests.exceptions.Timeout:
return empty_response, STATUS_HANDLER.pipe_states.CONNECTION_TIMEOUT
except requests.exceptions.RequestException:
@@ -90,6 +92,7 @@ def get_sales_prognosis_data(
response: SalesPrognosisResponse
status: Status
assert resp is not None, "tried to use not defined response"
if resp.status_code == 200:
response = SalesPrognosisResponse(**resp.json())
status = STATUS_HANDLER.SUCCESS

View File

@@ -300,45 +300,3 @@ class Session:
self._remove_session_token()
return self.login()
def assert_login(
self,
) -> tuple[LoginResponse, Status]:
# check if login token is still valid
# re-login if necessary
if self.session_token is None:
return self.login()
# use known endpoint which requires a valid token in its header
# evaluate the response to decide if:
# current token is still valid, token is not valid, other errors occurred
ROUTE: Final[str] = "verkauf/umsatzprognosedaten"
URL: Final = combine_route(self.base_url, ROUTE)
params: dict[str, int] = {"FirmaId": 999999}
empty_response = LoginResponse(token="")
try:
resp = requests.get(
URL,
params=params,
headers=self.headers, # type: ignore
timeout=API_CON_TIMEOUT,
)
except requests.exceptions.Timeout: # pragma: no cover
return empty_response, STATUS_HANDLER.pipe_states.CONNECTION_TIMEOUT
except requests.exceptions.RequestException: # pragma: no cover
return empty_response, STATUS_HANDLER.pipe_states.CONNECTION_ERROR
response: LoginResponse
status: Status
if resp.status_code == 200:
response = LoginResponse(token=self.session_token)
status = STATUS_HANDLER.SUCCESS
elif resp.status_code == 401:
self._remove_session_token()
response, status = self.login()
else:
response = empty_response
err = DelBarApiError(status_code=resp.status_code, **resp.json())
status = STATUS_HANDLER.api_error(err)
return response, status