You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
117 lines
3.9 KiB
Python
117 lines
3.9 KiB
Python
"""
|
|
Backend API Client
|
|
Handles communication with Django backend
|
|
"""
|
|
import requests
|
|
import logging
|
|
from typing import Optional, Dict, Any
|
|
from requests.adapters import HTTPAdapter
|
|
from urllib3.util.retry import Retry
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class BackendClient:
|
|
def __init__(self, base_url: str, timeout: int = 5, max_retries: int = 3):
|
|
self.base_url = base_url
|
|
self.timeout = timeout
|
|
self.max_retries = max_retries
|
|
self.session = self._create_session()
|
|
|
|
def _create_session(self) -> requests.Session:
|
|
"""Create a requests session with retry logic"""
|
|
session = requests.Session()
|
|
|
|
retry_strategy = Retry(
|
|
total=self.max_retries,
|
|
backoff_factor=1,
|
|
status_forcelist=[429, 500, 502, 503, 504],
|
|
allowed_methods=["HEAD", "GET", "OPTIONS", "POST"]
|
|
)
|
|
|
|
adapter = HTTPAdapter(max_retries=retry_strategy)
|
|
session.mount("http://", adapter)
|
|
session.mount("https://", adapter)
|
|
|
|
return session
|
|
|
|
def post_reading(self, port: str, data: str) -> bool:
|
|
"""Post a serial port reading to the backend"""
|
|
try:
|
|
url = f"{self.base_url}/api/readings/"
|
|
payload = {
|
|
'port': port,
|
|
'data': data
|
|
}
|
|
|
|
response = self.session.post(
|
|
url,
|
|
json=payload,
|
|
timeout=self.timeout
|
|
)
|
|
|
|
if response.status_code in [200, 201]:
|
|
logger.info(f"Successfully posted reading to backend")
|
|
return True
|
|
else:
|
|
logger.warning(f"Backend returned status {response.status_code}")
|
|
return False
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
logger.error(f"Error posting reading to backend: {e}")
|
|
return False
|
|
|
|
def get_latest_reading(self, port: Optional[str] = None) -> Optional[Dict[str, Any]]:
|
|
"""Get the latest reading from backend"""
|
|
try:
|
|
url = f"{self.base_url}/api/readings/latest/"
|
|
params = {'port': port} if port else {}
|
|
|
|
response = self.session.get(
|
|
url,
|
|
params=params,
|
|
timeout=self.timeout
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
else:
|
|
logger.warning(f"Backend returned status {response.status_code}")
|
|
return None
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
logger.error(f"Error getting latest reading: {e}")
|
|
return None
|
|
|
|
def get_readings(self, port: Optional[str] = None, limit: int = 10) -> Optional[list]:
|
|
"""Get readings from backend"""
|
|
try:
|
|
url = f"{self.base_url}/api/readings/"
|
|
params = {'limit': limit}
|
|
if port:
|
|
params['port'] = port
|
|
|
|
response = self.session.get(
|
|
url,
|
|
params=params,
|
|
timeout=self.timeout
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
else:
|
|
logger.warning(f"Backend returned status {response.status_code}")
|
|
return None
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
logger.error(f"Error getting readings: {e}")
|
|
return None
|
|
|
|
def health_check(self) -> bool:
|
|
"""Check if backend is available"""
|
|
try:
|
|
url = f"{self.base_url}/api/health/"
|
|
response = self.session.get(url, timeout=self.timeout)
|
|
return response.status_code == 200
|
|
except Exception as e:
|
|
logger.error(f"Backend health check failed: {e}")
|
|
return False |