This commit is contained in:
rmanach 2024-12-18 14:58:08 +01:00
parent 3fab2cb649
commit 46cbbd749c

View File

@ -24,7 +24,7 @@ __all__ = ["Client", "OrderField"]
def refresh(): def refresh():
""" """
Catch 401 status code (UnauthorizedException) Catch 401 status code (UnauthorizedException)
and refresh the access token and retry. refresh the access token and retry.
""" """
def decorator(func): def decorator(func):
@ -183,7 +183,7 @@ class Client:
Initialize the client from env variables (.env & global) and Initialize the client from env variables (.env & global) and
log in into the application. log in into the application.
If the login fails, and exception is raised. If the login fails, an exception is raised.
""" """
env_data = { env_data = {
"WHEREIS_API_EMAIL": os.getenv("WHEREIS_API_EMAIL", ""), "WHEREIS_API_EMAIL": os.getenv("WHEREIS_API_EMAIL", ""),
@ -337,7 +337,8 @@ class Client:
@refresh() @refresh()
def delete_session(self, id_: UUID) -> None | WhereIsException: def delete_session(self, id_: UUID) -> None | WhereIsException:
""" """
Close and delete the session. Users can't be added anymore. Close and delete the session.
NOTE: The GPS positions associated to the session are not deleted ! NOTE: The GPS positions associated to the session are not deleted !
""" """
session_url = urljoin(self.base_url, f"/sessions/{id_}/") session_url = urljoin(self.base_url, f"/sessions/{id_}/")
@ -377,6 +378,7 @@ class Client:
@refresh() @refresh()
def close_session(self, id_: UUID) -> dict[str, Any] | WhereIsException: def close_session(self, id_: UUID) -> dict[str, Any] | WhereIsException:
"""Close the DEFINITIVELY the session. Users can't be added anymore."""
session_url = urljoin(self.base_url, f"/sessions/{id_}/close/") session_url = urljoin(self.base_url, f"/sessions/{id_}/close/")
logging.info(f"close session: {session_url}") logging.info(f"close session: {session_url}")
@ -495,7 +497,9 @@ class Client:
return None return None
@refresh() @refresh()
def _get_paginate_gps_postions(self, url: str) -> dict[str, Any] | WhereIsException: def _get_paginate_gps_positions(
self, url: str
) -> dict[str, Any] | WhereIsException:
res = self.session.get(url) res = self.session.get(url)
if res.status_code == 401: if res.status_code == 401:
@ -539,7 +543,7 @@ class Client:
while gps_url is not None: while gps_url is not None:
logging.info(f"get gps data from: {gps_url}") logging.info(f"get gps data from: {gps_url}")
data = self._get_paginate_gps_postions(gps_url) data = self._get_paginate_gps_positions(gps_url)
lst_gps_positions.extend([d for d in data["results"]]) lst_gps_positions.extend([d for d in data["results"]])
gps_url = data["next"] gps_url = data["next"]