82 lines
3.4 KiB
Python
82 lines
3.4 KiB
Python
import asyncio
|
|
import logging
|
|
from database import SessionLocal
|
|
from models import VitalReading, TransmissionLog
|
|
from schemas import NormalizedVitals
|
|
from api_client import forward_vitals_to_api
|
|
from config import settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
async def retry_failed_transmissions():
|
|
"""
|
|
Background task that periodically checks for untransmitted vital readings
|
|
and attempts to forward them to the API.
|
|
"""
|
|
logger.info("Starting background task for retrying failed transmissions...")
|
|
|
|
while True:
|
|
try:
|
|
# 1. Fetch untransmitted readings in a quick DB session and close it immediately
|
|
db = SessionLocal()
|
|
readings_to_retry = []
|
|
try:
|
|
failed_readings = db.query(VitalReading).filter(VitalReading.transmitted == False).order_by(VitalReading.id.desc()).limit(20).all()
|
|
for r in failed_readings:
|
|
readings_to_retry.append({
|
|
"id": r.id,
|
|
"vitals": NormalizedVitals(
|
|
device_id=r.device_id,
|
|
patient_id=r.patient_id,
|
|
timestamp=r.timestamp,
|
|
heart_rate=r.heart_rate,
|
|
spo2=r.spo2,
|
|
systolic_bp=r.systolic_bp,
|
|
diastolic_bp=r.diastolic_bp,
|
|
map_bp=r.map_bp,
|
|
respiratory_rate=r.respiratory_rate,
|
|
temperature=r.temperature
|
|
)
|
|
})
|
|
finally:
|
|
db.close()
|
|
|
|
# 2. Forward outside of any DB transaction
|
|
successful_ids = []
|
|
failed_ids = []
|
|
consecutive_failures = 0
|
|
|
|
for item in readings_to_retry:
|
|
success = await forward_vitals_to_api(item["vitals"])
|
|
if success:
|
|
successful_ids.append(item["id"])
|
|
consecutive_failures = 0
|
|
else:
|
|
failed_ids.append(item["id"])
|
|
consecutive_failures += 1
|
|
# If external API is unresponsive, stop this batch immediately so we don't spam
|
|
if consecutive_failures >= 3:
|
|
logger.warning("External API appears unreachable during retry; pausing retry queue.")
|
|
break
|
|
|
|
# 3. Update status in a quick DB session
|
|
if successful_ids or failed_ids:
|
|
db = SessionLocal()
|
|
try:
|
|
for rid in successful_ids:
|
|
db.query(VitalReading).filter(VitalReading.id == rid).update({"transmitted": True})
|
|
db.add(TransmissionLog(reading_id=rid, status="success", response_code=200))
|
|
for rid in failed_ids:
|
|
db.add(TransmissionLog(reading_id=rid, status="failed", error_message="Retry failed"))
|
|
db.commit()
|
|
except Exception as e:
|
|
logger.error(f"Failed updating retry transmission status: {e}")
|
|
db.rollback()
|
|
finally:
|
|
db.close()
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error in retry background task: {e}")
|
|
|
|
await asyncio.sleep(settings.retry_interval_seconds)
|