Merge branch 'release/v0.1.2'

v0.1.1

src/__init__.py

@@ -2,4 +2,4 @@ from .client import Client, OrderField
 
 __all__ = ["Client", "OrderField"]
 
-VERSION = "0.1.1"
+VERSION = "0.1.2"

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)
@@ -177,6 +180,24 @@ class Client:
 
         return None
 
+    @staticmethod
+    def _init_client(
+        base_url: str, email: str, password: str, verify: bool = True
+    ) -> "Client":
+        cli = Client(base_url, email, password)
+        cli.session = Session()
+        cli.session.headers.update({"content-type": "application/json"})
+
+        if not verify:
+            urllib3.disable_warnings()
+
+        cli.session.verify = verify
+
+        cli._login()
+
+        logging.info(f"client successfully initialized for user: {cli.email}")
+        return cli
+
     @classmethod
     def from_env(cls) -> "Client":
         """
@@ -193,24 +214,17 @@ class Client:
         dotenv_data = dotenv_values()
         env_data.update(dotenv_data)  # type: ignore
 
-        cli = Client(
+        return Client._init_client(
             env_data.get("WHEREIS_API_BASE_URL", ""),
             env_data.get("WHEREIS_API_EMAIL", ""),
             env_data.get("WHEREIS_API_PASSWORD", ""),
         )
-        cli.session = Session()
-        cli.session.headers.update({"content-type": "application/json"})
 
-        is_cert_verify = env_data.get("WHEREIS_CERT_VERIFY", "") != "false"
+    @classmethod
-        if not is_cert_verify:
+    def from_creds(
-            urllib3.disable_warnings()
+        cls, base_url: str, email: str, password: str, verify: bool = False
-
+    ) -> "Client":
-        cli.session.verify = is_cert_verify
+        return Client._init_client(base_url, email, password, verify)
-
-        cli._login()
-
-        logging.info(f"client successfully initialized for user: {cli.email}")
-        return cli
 
     @refresh()
     def _get_sessions_page(self, url: str) -> dict[str, Any] | WhereIsException:
@@ -236,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:
@@ -265,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)
@@ -281,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}
@@ -313,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}
@@ -341,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)
@@ -363,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})
@@ -379,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)
@@ -447,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)
@@ -468,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)
@@ -483,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)
@@ -527,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: