Cache the Dell token to disk so it survives restarts

The in-process token cache was lost on every app restart, forcing a new token
request each time and tripping Dell's token-endpoint rate limit. Add an L2 file
cache in the instance dir (keyed by a hash of the client id, so the id never
lands on disk), mirroring warranty_sync.py. Once one token is obtained it is
reused for ~1h across restarts and workers, avoiding the cooldown in normal use.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
cproudlock
2026-07-09 16:24:40 -04:00
parent 9cc7cc7529
commit b90a13c7e5

View File

@@ -8,17 +8,49 @@ Per-vendor config lives in settings (warranty_<name>_*), masked where secret,
so nothing is hardcoded and the integration is toggled per site.
"""
import hashlib
import json
import os
import time
import requests
from flask import current_app
from shopdb.api import db
from shopdb.core.models import Setting
# Process-wide Dell token cache. Dell rate-limits the token endpoint, so a fresh
# request per refresh trips a 401 cooldown. Tokens live ~1h; cache + reuse.
# Two-level Dell token cache. Dell rate-limits the token endpoint, so a fresh
# request per refresh (or per app restart) trips a 401 cooldown. Tokens live ~1h.
# L1 = process memory (fast); L2 = a file in the instance dir so the token
# survives restarts and is shared across workers - mirrors warranty_sync.py.
_dell_token_cache = {}
def _token_file(clientid):
# Per-client file, name keyed by a hash so the id never lands on disk.
tag = hashlib.sha256((clientid or '').encode()).hexdigest()[:16]
return os.path.join(current_app.instance_path, f'.dell_token_{tag}.json')
def _read_file_token(clientid):
try:
with open(_token_file(clientid)) as handle:
cached = json.load(handle)
if cached.get('expires_at', 0) > time.time() + 60:
return cached['token']
except (OSError, ValueError, KeyError):
pass
return None
def _write_file_token(clientid, token, expires_at):
try:
os.makedirs(current_app.instance_path, exist_ok=True)
with open(_token_file(clientid), 'w') as handle:
json.dump({'token': token, 'expires_at': expires_at}, handle)
except OSError:
pass
# Dell TechDirect defaults, overridable via settings if Dell changes endpoints.
# The asset-entitlements service lives under sbil/eapi (device.warranty path 404s
# with "Service Not Found" for this account). Verified against the live API.
@@ -82,10 +114,15 @@ class DellProvider(WarrantyProvider):
}
def _get_token(self, config):
# Reuse a cached token while valid - Dell rate-limits the token endpoint.
cached = _dell_token_cache.get(config['clientid'])
clientid = config['clientid']
# L1: process memory.
cached = _dell_token_cache.get(clientid)
if cached and cached['expires_at'] > time.time() + 60:
return cached['token']
# L2: file cache (survives restarts / shared across workers).
file_token = _read_file_token(clientid)
if file_token:
return file_token
# Dell expects the client id/secret as HTTP Basic auth, grant type in the
# body. (Passing them in the body trips the token endpoint's rate limiter.)
@@ -111,9 +148,9 @@ class DellProvider(WarrantyProvider):
raise WarrantyLookupError(f'Dell auth failed: {exc}')
if not token:
raise WarrantyLookupError('Dell auth returned no access_token')
_dell_token_cache[config['clientid']] = {
'token': token, 'expires_at': time.time() + expires_in - 60,
}
expires_at = time.time() + expires_in - 60
_dell_token_cache[clientid] = {'token': token, 'expires_at': expires_at}
_write_file_token(clientid, token, expires_at)
return token
def _fetch(self, config, token, servicetag):