diff --git a/app/routes.py b/app/routes.py index 264378a..478299c 100644 --- a/app/routes.py +++ b/app/routes.py @@ -2,22 +2,31 @@ import time from flask import Blueprint, current_app, jsonify, render_template, request from app.services import analytics, model -from app.services.cache import all_meta, all_records, connect, count_records, get_meta, query_earthquakes, seed_bootstrap_if_empty, set_meta, upsert_earthquakes +from app.services.cache import all_meta, all_records, connect, count_records, feed_meta_key, query_earthquakes, seed_bootstrap_if_empty, set_meta, upsert_earthquakes from app.services.usgs_client import fetch_feed from app.utils import bool_arg, parse_time_ms, safe_float, safe_int bp=Blueprint('main',__name__) def feed_url(name): feeds=current_app.config['FEEDS']; return current_app.config['USGS_FEED_URL'] if name=='custom' else feeds.get(name, feeds.get(current_app.config['DEFAULT_FEED_WINDOW'], current_app.config['USGS_FEED_URL'])) +def feed_meta_value(meta, key, feed, default=''): + scoped=feed_meta_key(key,feed) + if scoped in meta: return meta[scoped] + return meta.get(key, default) if meta.get('last_refresh_source') == feed else default + def refresh_if_needed(feed='day', force=False): - db=current_app.config['SQLITE_DB_PATH']; ttl=current_app.config['CACHE_TTL_SECONDS']; last=int(get_meta(db,'last_successful_refresh_epoch','0') or 0) + db=current_app.config['SQLITE_DB_PATH']; ttl=current_app.config['CACHE_TTL_SECONDS']; meta=all_meta(db) + last=int(feed_meta_value(meta,'last_successful_refresh_epoch',feed,'0') or 0) if not force and int(time.time())-last <= ttl: return 'cache' try: rows=fetch_feed(feed_url(feed), current_app.config['REQUEST_TIMEOUT_SECONDS']); n=upsert_earthquakes(db,rows,feed,False) return f'live:{n}' except Exception as exc: current_app.logger.warning('USGS refresh failed: %s', exc) - with connect(db) as c: set_meta(c,'last_refresh_error',str(exc)[:300]); set_meta(c,'last_refresh_source','fallback_cache'); c.commit() + with connect(db) as c: + set_meta(c,'last_refresh_error',str(exc)[:300]); set_meta(c,'last_refresh_source','fallback_cache') + set_meta(c,feed_meta_key('last_refresh_error',feed),str(exc)[:300]); set_meta(c,feed_meta_key('last_refresh_source',feed),'fallback_cache') + c.commit() if count_records(db)==0: seed_bootstrap_if_empty(db, current_app.config['ENABLE_BOOTSTRAP_DATA']); return 'bootstrap_fallback' return 'fallback_cache' def filters_from_args(): @@ -42,13 +51,16 @@ def api_earthquakes(): limit=safe_int(request.args.get('limit'),500,1,current_app.config['MAX_API_LIMIT']); sort=request.args.get('sort','time_desc') filt=filters_from_args(); rows=query_earthquakes(current_app.config['SQLITE_DB_PATH'],filt,limit,sort); meta=all_meta(current_app.config['SQLITE_DB_PATH']) warnings=[] - if 'fallback' in refresh or meta.get('last_refresh_source') in {'bootstrap_demo','fallback_cache'}: warnings.append('Using cached or bootstrap data because live USGS refresh was unavailable or skipped.') - return jsonify({'meta':{'source':meta.get('last_refresh_source','cache'),'refresh_result':refresh,'last_refresh_epoch':int(meta.get('last_successful_refresh_epoch','0') or 0),'count':len(rows),'available_feeds':list(current_app.config['FEEDS'].keys())},'filters':filt,'count':len(rows),'source':meta.get('last_refresh_source','cache'),'warnings':warnings,'data':rows}) + source=feed_meta_value(meta,'last_refresh_source',feed,'cache') + last_epoch=int(feed_meta_value(meta,'last_successful_refresh_epoch',feed,'0') or 0) + if 'fallback' in refresh or source in {'bootstrap_demo','fallback_cache'}: warnings.append('Using cached or bootstrap data because live USGS refresh was unavailable or skipped.') + return jsonify({'meta':{'source':source,'refresh_result':refresh,'last_refresh_epoch':last_epoch,'count':len(rows),'available_feeds':list(current_app.config['FEEDS'].keys())},'filters':filt,'count':len(rows),'source':source,'warnings':warnings,'data':rows}) @bp.route('/api/summary') def api_summary(): refresh_if_needed(request.args.get('feed', current_app.config['DEFAULT_FEED_WINDOW']), False) rows=query_earthquakes(current_app.config['SQLITE_DB_PATH'],filters_from_args(),current_app.config['MAX_API_LIMIT'],'time_desc') - return jsonify({'summary':analytics.summarize(rows),'count':len(rows),'source':all_meta(current_app.config['SQLITE_DB_PATH']).get('last_refresh_source','cache')}) + feed=request.args.get('feed', current_app.config['DEFAULT_FEED_WINDOW']); meta=all_meta(current_app.config['SQLITE_DB_PATH']) + return jsonify({'summary':analytics.summarize(rows),'count':len(rows),'source':feed_meta_value(meta,'last_refresh_source',feed,'cache')}) @bp.route('/api/regions') def api_regions(): return jsonify({'regions':analytics.regions(query_earthquakes(current_app.config['SQLITE_DB_PATH'],filters_from_args(),current_app.config['MAX_API_LIMIT'],'time_desc'))}) @bp.route('/api/timeseries') diff --git a/app/services/cache.py b/app/services/cache.py index 40cf491..b2099c2 100644 --- a/app/services/cache.py +++ b/app/services/cache.py @@ -26,6 +26,9 @@ def init_db(db_path: str) -> None: c.execute(f'CREATE INDEX IF NOT EXISTS {idx} ON earthquakes({col})') c.commit() +def feed_meta_key(k: str, feed: str) -> str: + return f'{k}:{feed}' + def set_meta(c, k, v): c.execute('INSERT INTO refresh_metadata(key,value) VALUES(?,?) ON CONFLICT(key) DO UPDATE SET value=excluded.value',(k,str(v))) def get_meta(db_path: str, k: str, default: str='') -> str: with connect(db_path) as c: @@ -35,7 +38,6 @@ def all_meta(db_path: str) -> dict[str,str]: with connect(db_path) as c: return {r['key']:r['value'] for r in c.execute('SELECT key,value FROM refresh_metadata')} def upsert_earthquakes(db_path: str, rows: list[dict[str,Any]], source_feed='unknown', is_bootstrap=False) -> int: - if not rows: return 0 now=int(time.time()) vals=[] for r in rows: @@ -45,7 +47,9 @@ def upsert_earthquakes(db_path: str, rows: list[dict[str,Any]], source_feed='unk updates=','.join([f'{c}=excluded.{c}' for c in COLUMNS[1:]]+['ingested_at=excluded.ingested_at','source_feed=excluded.source_feed','is_bootstrap=excluded.is_bootstrap']) with connect(db_path) as c: c.executemany(f'INSERT INTO earthquakes({",".join(COLUMNS)},ingested_at,source_feed,is_bootstrap) VALUES({ph}) ON CONFLICT(id) DO UPDATE SET {updates}', vals) - set_meta(c,'last_successful_refresh_epoch',now); set_meta(c,'last_refresh_source',source_feed); set_meta(c,'last_record_count',len(rows)); c.commit() + set_meta(c,'last_successful_refresh_epoch',now); set_meta(c,'last_refresh_source',source_feed); set_meta(c,'last_record_count',len(rows)) + set_meta(c,feed_meta_key('last_successful_refresh_epoch',source_feed),now); set_meta(c,feed_meta_key('last_refresh_source',source_feed),source_feed); set_meta(c,feed_meta_key('last_record_count',source_feed),len(rows)); set_meta(c,feed_meta_key('last_refresh_error',source_feed),'') + c.commit() return len(rows) def seed_bootstrap_if_empty(db_path: str, enabled=True) -> None: diff --git a/tests/test_app.py b/tests/test_app.py index ed04f3e..36217e5 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -11,7 +11,7 @@ def client(tmp_path, monkeypatch): app=create_app(); app.config['TESTING']=True from app.services.cache import upsert_earthquakes from app.services.usgs_client import parse_geojson_features - upsert_earthquakes(app.config['SQLITE_DB_PATH'], parse_geojson_features(SAMPLE), 'test') + upsert_earthquakes(app.config['SQLITE_DB_PATH'], parse_geojson_features(SAMPLE), 'day') return app.test_client() def test_geojson_parsing(): @@ -53,3 +53,86 @@ def test_forecast_fallback(client): def test_model_status(client): r=client.get('/api/model/status') assert r.status_code==200 and 'model_exists' in r.json + + +def test_refresh_metadata_is_per_feed(tmp_path, monkeypatch): + monkeypatch.setenv('SQLITE_DB_PATH', str(tmp_path/'feeds.db')) + monkeypatch.setenv('ENABLE_BOOTSTRAP_DATA','false') + from app import create_app + from app import routes + from app.services.cache import all_meta + + app=create_app(); app.config['TESTING']=True; app.config['CACHE_TTL_SECONDS']=3600 + calls=[] + def fake_fetch(url, timeout): + calls.append(url) + idx=len(calls) + return [{'id':f'event-{idx}','time_ms':idx,'updated_ms':idx,'place':'Test','magnitude':1.0,'longitude':1,'latitude':1,'depth':1}] + monkeypatch.setattr(routes, 'fetch_feed', fake_fetch) + + feeds=['day','7day','30day','significant','m1','m2.5','m4.5'] + with app.app_context(): + for feed in feeds: + assert routes.refresh_if_needed(feed) == 'live:1' + assert routes.refresh_if_needed(feed) == 'cache' + + assert len(calls) == len(feeds) + meta=all_meta(app.config['SQLITE_DB_PATH']) + for feed in feeds: + assert int(meta[f'last_successful_refresh_epoch:{feed}']) > 0 + assert meta[f'last_refresh_source:{feed}'] == feed + + +def test_legacy_global_metadata_only_applies_to_matching_feed(tmp_path, monkeypatch): + monkeypatch.setenv('SQLITE_DB_PATH', str(tmp_path/'legacy.db')) + monkeypatch.setenv('ENABLE_BOOTSTRAP_DATA','false') + from app import create_app + from app import routes + from app.services.cache import connect, set_meta + import time + + app=create_app(); app.config['TESTING']=True; app.config['CACHE_TTL_SECONDS']=3600 + with connect(app.config['SQLITE_DB_PATH']) as c: + set_meta(c, 'last_successful_refresh_epoch', int(time.time())) + set_meta(c, 'last_refresh_source', 'day') + c.commit() + + calls=[] + def fake_fetch(url, timeout): + calls.append(url) + return [{'id':'legacy-7day','time_ms':1,'updated_ms':1,'place':'Test','magnitude':1.0,'longitude':1,'latitude':1,'depth':1}] + monkeypatch.setattr(routes, 'fetch_feed', fake_fetch) + + with app.app_context(): + assert routes.refresh_if_needed('day') == 'cache' + assert routes.refresh_if_needed('7day') == 'live:1' + + assert len(calls) == 1 + + +def test_earthquakes_api_refreshes_each_feed_independently(tmp_path, monkeypatch): + monkeypatch.setenv('SQLITE_DB_PATH', str(tmp_path/'api-feeds.db')) + monkeypatch.setenv('ENABLE_BOOTSTRAP_DATA','false') + from app import create_app + from app import routes + + app=create_app(); app.config['TESTING']=True; app.config['CACHE_TTL_SECONDS']=3600 + calls=[] + def fake_fetch(url, timeout): + calls.append(url) + idx=len(calls) + return [{'id':f'api-event-{idx}','time_ms':idx,'updated_ms':idx,'place':'Test','magnitude':1.0,'longitude':1,'latitude':1,'depth':1}] + monkeypatch.setattr(routes, 'fetch_feed', fake_fetch) + + client=app.test_client() + feeds=['day','7day','30day','significant','m1','m2.5','m4.5'] + for feed in feeds: + first=client.get(f'/api/earthquakes?feed={feed}&limit=1') + second=client.get(f'/api/earthquakes?feed={feed}&limit=1') + assert first.status_code == 200 + assert second.status_code == 200 + assert first.json['meta']['refresh_result'] == 'live:1' + assert second.json['meta']['refresh_result'] == 'cache' + assert first.json['meta']['source'] == feed + + assert len(calls) == len(feeds)