From 0e482fcbc45b4d9764d97e41471a52400e7f485f Mon Sep 17 00:00:00 2001 From: rmanach Date: Wed, 18 Dec 2024 11:30:31 +0100 Subject: [PATCH] add gps positions get endpoint --- main.py | 9 +++-- pyproject.toml | 4 ++- src/__init__.py | 2 +- src/client.py | 92 +++++++++++++++++++++++++++++++++++++---------- src/exceptions.py | 4 +-- 5 files changed, 86 insertions(+), 25 deletions(-) diff --git a/main.py b/main.py index d59fb40..5da3376 100644 --- a/main.py +++ b/main.py @@ -28,7 +28,7 @@ if __name__ == "__main__": cli = Client.from_env() and use all the available methods to interact with the WhereIs API. - + Sessions: - create_session - get_sessions @@ -43,6 +43,8 @@ if __name__ == "__main__": - get_wstokens - create_wstoken - delete_wstoken + GPS positions: + - get_gps_positions """ logging.info(f"WhereIs client v{VERSION}") @@ -111,8 +113,9 @@ if __name__ == "__main__": # get session events cli.watch_session_events(session.get("id")) - # doing your stuff... - time.sleep(2) + # get gps positions from "2024-12-25T23:00:00" to infinity... + gps_positions = cli.get_gps_positions(date_start="2024-12-25T23:00:00") + print(json.dumps(gps_positions, indent=2)) # close the session session = cli.close_session(session.get("id")) diff --git a/pyproject.toml b/pyproject.toml index 8b96326..fe053d3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -8,6 +8,8 @@ dynamic = ["version"] description = "WhereIs API client library" dependencies = [ "requests==2.32.3", + "python-dotenv==1.0.1", + "sseclient-py==1.8.0", ] [tool.hatch.version] @@ -20,7 +22,7 @@ packages = ["src"] only-include = ["src"] [tool.hatch.build.targets.wheel.sources] -"src" = "whereis-client" +"src" = "whereis_client" [tool.ruff] select = ["E", "F", "I"] diff --git a/src/__init__.py b/src/__init__.py index d714875..6160875 100644 --- a/src/__init__.py +++ b/src/__init__.py @@ -2,4 +2,4 @@ from .client import Client, OrderField __all__ = ["Client", "OrderField"] -VERSION = "0.1.0" +VERSION = "0.1.0a0" diff --git a/src/client.py b/src/client.py index 5434fed..d333cfe 100644 --- a/src/client.py +++ b/src/client.py @@ -1,6 +1,7 @@ import logging import os from dataclasses import dataclass, field +from datetime import datetime as dt from enum import Enum from threading import Event, Thread from typing import Any, Callable @@ -29,12 +30,17 @@ def refresh(): def decorator(func): def wrapper(*args, **kwargs): if len(args) > 0 and isinstance(args[0], Client): - try: - return func(*args, **kwargs) - except UnauthorizedException: - logging.warning("refresh access token...") - args[0]._refresh() - return func(*args, **kwargs) + for i in range(2): + try: + return func(*args, **kwargs) + except UnauthorizedException as e: + if i == 1: # second attempt + logging.error( + "second call attempt failed after refreshing token" + ) + raise e + logging.warning("refresh access token...") + args[0]._refresh() return wrapper @@ -211,11 +217,10 @@ class Client: """Get paginate sessions.""" res = self.session.get(url) if res.status_code == 401: - raise UnauthorizedException() + raise UnauthorizedException(url, res) if res.status_code >= 400: - if res.status_code >= 400: - raise WhereIsException(url, res) + raise WhereIsException(url, res) return res.json() @@ -265,7 +270,7 @@ class Client: res = self.session.get(session_url) if res.status_code == 401: - raise UnauthorizedException() + raise UnauthorizedException(session_url, res) if res.status_code >= 400: raise WhereIsException(session_url, res) @@ -293,7 +298,7 @@ class Client: ) if res.status_code == 401: - raise UnauthorizedException() + raise UnauthorizedException(sessions_url, res) if res.status_code >= 400: raise WhereIsException(sessions_url, res) @@ -322,7 +327,7 @@ class Client: res = self.session.patch(session_url, json=data) if res.status_code == 401: - raise UnauthorizedException() + raise UnauthorizedException(session_url, res) if res.status_code >= 400: raise WhereIsException(session_url, res) @@ -341,7 +346,7 @@ class Client: res = self.session.delete(session_url) if res.status_code == 401: - raise UnauthorizedException() + raise UnauthorizedException(session_url, res) if res.status_code >= 400: raise WhereIsException(session_url, res) @@ -363,7 +368,7 @@ class Client: res = self.session.post(session_url, json={"users": users}) if res.status_code == 401: - raise UnauthorizedException() + raise UnauthorizedException(session_url, res) if res.status_code >= 400: raise WhereIsException(session_url, res) @@ -378,7 +383,7 @@ class Client: res = self.session.post(session_url) if res.status_code == 401: - raise UnauthorizedException() + raise UnauthorizedException(session_url, res) if res.status_code >= 400: raise WhereIsException(session_url, res) @@ -446,7 +451,7 @@ class Client: res = self.session.get(wstoken_url) if res.status_code == 401: - raise UnauthorizedException() + raise UnauthorizedException(wstoken_url, res) if res.status_code >= 400: raise WhereIsException(wstoken_url, res) @@ -467,7 +472,7 @@ class Client: res = self.session.post(wstoken_url) if res.status_code == 401: - raise UnauthorizedException() + raise UnauthorizedException(wstoken_url, res) if res.status_code >= 400: raise WhereIsException(wstoken_url, res) @@ -482,9 +487,60 @@ class Client: res = self.session.delete(wstoken_url) if res.status_code == 401: - raise UnauthorizedException() + raise UnauthorizedException(wstoken_url, res) if res.status_code >= 400: raise WhereIsException(wstoken_url, res) return None + + @refresh() + def _get_paginate_gps_postions(self, url: str) -> dict[str, Any] | WhereIsException: + res = self.session.get(url) + + if res.status_code == 401: + raise UnauthorizedException(url, res) + + if res.status_code >= 400: + raise WhereIsException(url, res) + + return res.json() + + def get_gps_positions( + self, + date_start: str | None = None, + date_end: str | None = None, + ) -> list[dict[str, Any]] | ValueError | WhereIsException: + """ + Gets GPS positions data. + + You can get GPS positions filtered by date interval using + optionals arguments `date_start` and `date_end`: + - `date_start`: [date_start,] + - `date_end`: [date_end,] + - `date_start` & `date_end`: [date_start,date_end] + + 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/") + lst_gps_positions: list[dict[str, Any]] = [] + + if date_start: + ds = dt.fromisoformat(date_start) + gps_url += f"?date_start={ds.isoformat()}" + if date_end: + de = dt.fromisoformat(date_end) + if date_start: + gps_url += f"&date_end={de.isoformat()}" + else: + gps_url += f"?date_end={de.isoformat()}" + + while gps_url is not None: + logging.info(f"get gps data from: {gps_url}") + + data = self._get_paginate_gps_postions(gps_url) + lst_gps_positions.extend([d for d in data["results"]]) + gps_url = data["next"] + + return lst_gps_positions diff --git a/src/exceptions.py b/src/exceptions.py index 61af4c9..9eb5cc3 100644 --- a/src/exceptions.py +++ b/src/exceptions.py @@ -1,6 +1,6 @@ from requests import Response -__all__ = ["WhereIsException"] +__all__ = ["WhereIsException", "UnauthorizedException"] class WhereIsException(Exception): @@ -19,5 +19,5 @@ class WhereIsException(Exception): return f"error calling: {self.url} - {self.error_code} - {self.content}" -class UnauthorizedException(Exception): +class UnauthorizedException(WhereIsException): pass