remove urljoin (concat instead)

This commit is contained in:
rmanach 2025-05-22 09:29:02 +02:00
parent 8b1a5ac175
commit 683e9dfa0c

View File

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