diff --git a/src/client.py b/src/client.py index b9a716f..cdb5345 100644 --- a/src/client.py +++ b/src/client.py @@ -5,7 +5,6 @@ from datetime import datetime as dt from enum import Enum from threading import Event, Thread from typing import Any, Callable -from urllib.parse import urljoin from uuid import UUID import requests @@ -86,7 +85,7 @@ class SessionWatcher: callback: Callable[[str], None] | None = None, verify: bool = True, ) -> "SessionWatcher": - session_url = urljoin(base_url, f"/sessions/{id_}/events/") + session_url = base_url + f"/sessions/{id_}/events/" headers = {**headers, "Accept": "text/event-stream"} resp = requests.get(session_url, stream=True, headers=headers, verify=verify) @@ -134,7 +133,7 @@ class Client: sessions = cli.get_sessions() """ - base_url: str + _base_url: str email: str password: str @@ -144,6 +143,10 @@ class Client: default_factory=dict, init=False ) + @property + def base_url(self) -> str: + return self._base_url.removesuffix("/") + def _login(self) -> WhereIsException | None: """Get the access token and store it in the `Session` header""" data = { @@ -151,7 +154,7 @@ class Client: "password": self.password, } - login_url = urljoin(self.base_url, "/auth/token/") + login_url = self.base_url + "/auth/token/" logging.info(f"login: {login_url}") res = self.session.post(login_url, json=data) @@ -165,7 +168,7 @@ class Client: def _refresh(self) -> WhereIsException | None: """Refresh the access token and store it in the `Session` header""" - refresh_url = urljoin(self.base_url, "/auth/refresh/") + refresh_url = self.base_url + "/auth/refresh/" logging.info(f"refresh: {refresh_url}") res = self.session.post(refresh_url) @@ -247,7 +250,7 @@ class Client: search: str, search sessions over username and description ordering: OrderField, ordering sessions by dates """ - sessions_url = urljoin(self.base_url, "/sessions/") + sessions_url = self.base_url + "/sessions/" has_query_param = False if search is not None: @@ -276,7 +279,7 @@ class Client: @refresh() def get_session(self, id_: UUID) -> list[dict[str, Any]] | WhereIsException: - session_url = urljoin(self.base_url, f"/sessions/{id_}/") + session_url = self.base_url + f"/sessions/{id_}/" logging.info(f"get session: {session_url}") res = self.session.get(session_url) @@ -292,7 +295,7 @@ class Client: def create_session( self, name: str, description: str | None = None, is_public: bool = False ) -> dict[str, Any] | WhereIsException: - sessions_url = urljoin(self.base_url, "/sessions/") + sessions_url = self.base_url + "/sessions/" logging.info(f"create session: {sessions_url}") data = {"name": name, "description": description, "is_public": is_public} @@ -324,7 +327,7 @@ class Client: description: str | None = None, is_public: bool | None = None, ) -> dict[str, Any] | WhereIsException: - session_url = urljoin(self.base_url, f"/sessions/{id_}/") + session_url = self.base_url + f"/sessions/{id_}/" logging.info(f"update session: {session_url}") data = {"name": name, "description": description, "is_public": is_public} @@ -352,7 +355,7 @@ class Client: NOTE: The GPS positions associated to the session are not deleted ! """ - session_url = urljoin(self.base_url, f"/sessions/{id_}/") + session_url = self.base_url + f"/sessions/{id_}/" logging.info(f"delete session: {session_url}") res = self.session.delete(session_url) @@ -374,7 +377,7 @@ class Client: WARN: An empty users list parameter cleans all users. """ - session_url = urljoin(self.base_url, f"/sessions/{id_}/users/") + session_url = self.base_url + f"/sessions/{id_}/users/" logging.info(f"update session users: {session_url}") res = self.session.post(session_url, json={"users": users}) @@ -390,7 +393,7 @@ class Client: @refresh() 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 = self.base_url + f"/sessions/{id_}/close/" logging.info(f"close session: {session_url}") res = self.session.post(session_url) @@ -458,7 +461,7 @@ class Client: @refresh() def get_wstokens(self) -> list[dict[str, Any]] | WhereIsException: - wstoken_url = urljoin(self.base_url, "/auth/ws-token/") + wstoken_url = self.base_url + "/auth/ws-token/" logging.info(f"get ws token: {wstoken_url}") res = self.session.get(wstoken_url) @@ -479,7 +482,7 @@ class Client: NOTE: only one, and only one ws token per user is allowed. If it expired, delete it and create a new one. """ - wstoken_url = urljoin(self.base_url, "/auth/ws-token/") + wstoken_url = self.base_url + "/auth/ws-token/" logging.info(f"create ws token: {wstoken_url}") res = self.session.post(wstoken_url) @@ -494,7 +497,7 @@ class Client: @refresh() def delete_wstoken(self, id_: UUID) -> None | WhereIsException: - wstoken_url = urljoin(self.base_url, f"/auth/ws-token/{id_}/") + wstoken_url = self.base_url + f"/auth/ws-token/{id_}/" logging.info(f"delete ws token: {wstoken_url}") res = self.session.delete(wstoken_url) @@ -538,7 +541,7 @@ class Client: The dates formats must be in any valid ISO 8601 format otherwise it will raise a ValueError. """ - gps_url = urljoin(self.base_url, "/gps/positions/") + gps_url = self.base_url + "/gps/positions/" lst_gps_positions: list[dict[str, Any]] = [] if date_start: