rmanach/whereis-client
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Wiki Activity

add gps positions get endpoint

Browse Source
This commit is contained in:
rmanach 2024-12-18 11:30:31 +01:00
parent a99621c42b
commit 0e482fcbc4
5 changed files with 86 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
main.py
View File

@ -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"))

4
pyproject.toml
View File

@ -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"]

2
src/__init__.py
View File

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

92
src/client.py
View File

@ -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

4
src/exceptions.py
View File

@ -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