gavindiaz/PolyWeather
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Isolate observation collector from web API

Browse Source
This commit is contained in:
2569718930@qq.com
2026-06-08 13:19:52 +08:00
parent ec84243fb9
commit 721173aea3
9 changed files with 369 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files
deploy.sh
+3
View File
@@ -220,6 +220,9 @@ if [ "$FAILED_BACKEND" = "1" ]; then
exit 1
fi
echo "Updating observation collector..."
compose_up_retry "observation collector" -d --no-deps polyweather_collector
echo "Updating frontend..."
compose_up_retry "frontend" -d --no-deps polyweather_frontend
docker-compose.yml
+55 -1
View File
@@ -29,6 +29,7 @@ services:
- .env
environment:
POLYWEATHER_REDIS_URL: ${POLYWEATHER_REDIS_URL:-redis://polyweather_redis:6379/0}
POLYWEATHER_COLLECTOR_PATCH_ENDPOINT: ''
POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED: 'false'
POLYWEATHER_SCAN_TERMINAL_PREWARM_ENABLED: 'false'
POLYWEATHER_SERVICE_ROLE: bot
@@ -99,15 +100,21 @@ services:
env_file: *id001
environment:
POLYWEATHER_REDIS_URL: ${POLYWEATHER_REDIS_URL:-redis://polyweather_redis:6379/0}
POLYWEATHER_COLLECTOR_PATCH_ENDPOINT: ''
POLYWEATHER_CITY_DETAIL_BATCH_CONCURRENCY: ${POLYWEATHER_CITY_DETAIL_BATCH_CONCURRENCY:-1}
POLYWEATHER_CITY_DETAIL_BATCH_GLOBAL_CONCURRENCY: ${POLYWEATHER_CITY_DETAIL_BATCH_GLOBAL_CONCURRENCY:-1}
POLYWEATHER_CITY_DETAIL_BATCH_PARTIAL_TIMEOUT_MS: ${POLYWEATHER_CITY_DETAIL_BATCH_PARTIAL_TIMEOUT_MS:-3000}
POLYWEATHER_OBSERVATION_COLLECTOR_AMOS_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_AMOS_SEC:-60}
POLYWEATHER_OBSERVATION_COLLECTOR_AMSC_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_AMSC_SEC:-180}
POLYWEATHER_OBSERVATION_COLLECTOR_COWIN_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_COWIN_SEC:-60}
POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED: ${POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED:-true}
POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED: 'false'
POLYWEATHER_OBSERVATION_COLLECTOR_HKO_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_HKO_SEC:-600}
POLYWEATHER_OBSERVATION_COLLECTOR_INITIAL_DELAY_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_INITIAL_DELAY_SEC:-5}
POLYWEATHER_OBSERVATION_COLLECTOR_MADIS_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_MADIS_SEC:-300}
POLYWEATHER_OBSERVATION_COLLECTOR_TICK_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_TICK_SEC:-30}
POLYWEATHER_SCAN_TERMINAL_PREWARM_ENABLED: 'false'
POLYWEATHER_SERVICE_ROLE: web
UVICORN_WORKERS: ${UVICORN_WORKERS:-2}
healthcheck:
interval: 30s
retries: 3
@@ -129,6 +136,53 @@ services:
volumes:
- ${POLYWEATHER_RUNTIME_DATA_DIR:-/var/lib/polyweather}:/var/lib/polyweather
- ${POLYWEATHER_RUNTIME_DATA_DIR:-/var/lib/polyweather}:/app/data
polyweather_collector:
command: python -m web.observation_collector_worker
container_name: polyweather_collector
depends_on:
polyweather_redis:
condition: service_healthy
polyweather_web:
condition: service_started
logging:
driver: "json-file"
options:
max-size: "50m"
max-file: "3"
cpus: ${POLYWEATHER_COLLECTOR_CPUS:-0.75}
env_file: *id001
environment:
POLYWEATHER_COLLECTOR_PATCH_ENDPOINT: ${POLYWEATHER_COLLECTOR_PATCH_ENDPOINT:-http://polyweather_web:8000/api/internal/collector-patch}
POLYWEATHER_OBSERVATION_COLLECTOR_AMOS_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_AMOS_SEC:-60}
POLYWEATHER_OBSERVATION_COLLECTOR_AMSC_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_AMSC_SEC:-180}
POLYWEATHER_OBSERVATION_COLLECTOR_COWIN_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_COWIN_SEC:-60}
POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED: 'true'
POLYWEATHER_OBSERVATION_COLLECTOR_HKO_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_HKO_SEC:-600}
POLYWEATHER_OBSERVATION_COLLECTOR_INITIAL_DELAY_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_INITIAL_DELAY_SEC:-15}
POLYWEATHER_OBSERVATION_COLLECTOR_MADIS_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_MADIS_SEC:-300}
POLYWEATHER_OBSERVATION_COLLECTOR_TICK_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_TICK_SEC:-30}
POLYWEATHER_REDIS_URL: ${POLYWEATHER_REDIS_URL:-redis://polyweather_redis:6379/0}
POLYWEATHER_SCAN_TERMINAL_PREWARM_ENABLED: 'false'
POLYWEATHER_SERVICE_ROLE: collector
healthcheck:
interval: 60s
retries: 3
test:
- CMD
- python
- -c
- import sqlite3; c=sqlite3.connect('/var/lib/polyweather/polyweather.db');
c.execute('SELECT 1'); c.close()
timeout: 10s
image: ghcr.io/yangyuan-zhen/polyweather-backend:${IMAGE_TAG:-latest}
mem_limit: ${POLYWEATHER_COLLECTOR_MEM_LIMIT:-768m}
memswap_limit: ${POLYWEATHER_COLLECTOR_MEMSWAP_LIMIT:-1g}
pids_limit: 256
restart: unless-stopped
user: ${UID:-1000}:${GID:-1000}
volumes:
- ${POLYWEATHER_RUNTIME_DATA_DIR:-/var/lib/polyweather}:/var/lib/polyweather
- ${POLYWEATHER_RUNTIME_DATA_DIR:-/var/lib/polyweather}:/app/data
x-polyweather-base:
env_file: *id001
image: ghcr.io/yangyuan-zhen/polyweather-backend:${IMAGE_TAG:-latest}
src/database/db_manager.py
+62 -36
View File
@@ -39,6 +39,10 @@ class DBManager:
conn.execute("PRAGMA busy_timeout=5000")
return conn
@staticmethod
def _is_sqlite_locked_error(exc: sqlite3.OperationalError) -> bool:
return "database is locked" in str(exc).lower()
def _init_cache_key(self) -> str:
return os.path.abspath(self.db_path)
@@ -3015,19 +3019,30 @@ class DBManager:
pressure_hpa: Optional[float] = None,
obs_time: str,
) -> None:
with self._get_connection() as conn:
conn.execute(
"""
INSERT INTO airport_obs_log (icao, city, temp_c, wind_kt, pressure_hpa, obs_time)
VALUES (?, ?, ?, ?, ?, ?)
""",
(str(icao).strip().upper(), str(city).strip().lower(),
temp_c, wind_kt, pressure_hpa, str(obs_time)),
)
conn.execute(
"DELETE FROM airport_obs_log WHERE created_at < datetime('now', '-2 hours')"
)
conn.commit()
safe_icao = str(icao).strip().upper()
safe_city = str(city).strip().lower()
try:
with self._get_connection() as conn:
conn.execute(
"""
INSERT INTO airport_obs_log (icao, city, temp_c, wind_kt, pressure_hpa, obs_time)
VALUES (?, ?, ?, ?, ?, ?)
""",
(safe_icao, safe_city, temp_c, wind_kt, pressure_hpa, str(obs_time)),
)
conn.execute(
"DELETE FROM airport_obs_log WHERE created_at < datetime('now', '-2 hours')"
)
conn.commit()
except sqlite3.OperationalError as exc:
if self._is_sqlite_locked_error(exc):
logger.warning(
"airport obs log skipped because sqlite is locked icao={} city={}",
safe_icao,
safe_city,
)
return
raise
def get_airport_obs_recent(
self, icao: str, minutes: int = 30
@@ -3068,30 +3083,41 @@ class DBManager:
safe_otime = str(otime_utc or "").strip()
if not safe_icao or not safe_runway or not safe_otime:
return
with self._get_connection() as conn:
existing = conn.execute(
"SELECT id FROM runway_obs_log WHERE icao=? AND runway=? AND otime_utc=? LIMIT 1",
(safe_icao, safe_runway, safe_otime),
).fetchone()
if existing:
try:
with self._get_connection() as conn:
existing = conn.execute(
"SELECT id FROM runway_obs_log WHERE icao=? AND runway=? AND otime_utc=? LIMIT 1",
(safe_icao, safe_runway, safe_otime),
).fetchone()
if existing:
return
conn.execute(
"""
INSERT INTO runway_obs_log (
icao, city, runway,
tdz_temp, mid_temp, end_temp, target_runway_max,
wind_dir, wind_speed, rvr, mor, humidity,
otime_utc
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
safe_icao, safe_city, safe_runway,
tdz_temp, mid_temp, end_temp, target_runway_max,
wind_dir, wind_speed, rvr, mor, humidity,
safe_otime,
),
)
conn.commit()
except sqlite3.OperationalError as exc:
if self._is_sqlite_locked_error(exc):
logger.warning(
"runway obs log skipped because sqlite is locked icao={} city={} runway={}",
safe_icao,
safe_city,
safe_runway,
)
return
conn.execute(
"""
INSERT INTO runway_obs_log (
icao, city, runway,
tdz_temp, mid_temp, end_temp, target_runway_max,
wind_dir, wind_speed, rvr, mor, humidity,
otime_utc
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
safe_icao, safe_city, safe_runway,
tdz_temp, mid_temp, end_temp, target_runway_max,
wind_dir, wind_speed, rvr, mor, humidity,
safe_otime,
),
)
conn.commit()
raise
def get_runway_obs_recent(
self, icao: str, minutes: int = 60
tests/test_deployment_runtime_config.py
+39 -7
View File
@@ -52,7 +52,7 @@ def test_scan_terminal_prewarm_only_runs_for_web_service(monkeypatch):
assert app_factory._scan_terminal_prewarm_enabled() is True
def test_observation_collector_only_runs_for_web_service(monkeypatch):
def test_observation_collector_only_runs_for_collector_service(monkeypatch):
from web import app_factory
monkeypatch.delenv("POLYWEATHER_SERVICE_ROLE", raising=False)
@@ -63,22 +63,53 @@ def test_observation_collector_only_runs_for_web_service(monkeypatch):
assert app_factory._observation_collector_enabled() is False
monkeypatch.setenv("POLYWEATHER_SERVICE_ROLE", "web")
assert app_factory._observation_collector_enabled() is False
monkeypatch.setenv("POLYWEATHER_SERVICE_ROLE", "collector")
assert app_factory._observation_collector_enabled() is True
monkeypatch.setenv("POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED", "false")
assert app_factory._observation_collector_enabled() is False
def test_docker_compose_isolates_scan_terminal_prewarm_to_web_service():
def test_docker_compose_isolates_collector_from_web_and_bot_services():
compose = (ROOT / "docker-compose.yml").read_text(encoding="utf-8")
bot_block = compose.split(" polyweather:", 1)[1].split(
"\n polyweather_frontend:",
1,
)[0]
web_block = compose.split(" polyweather_web:", 1)[1].split(
"\n polyweather_collector:",
1,
)[0]
collector_block = compose.split(" polyweather_collector:", 1)[1].split(
"\nx-polyweather-base:",
1,
)[0]
assert "POLYWEATHER_SERVICE_ROLE: web" in compose
assert "POLYWEATHER_SERVICE_ROLE: bot" in compose
assert "POLYWEATHER_SCAN_TERMINAL_PREWARM_ENABLED: 'false'" in compose
assert "POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED: 'false'" in compose
assert "POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED: ${POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED:-true}" in compose
assert "POLYWEATHER_OBSERVATION_COLLECTOR_AMSC_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_AMSC_SEC:-180}" in compose
assert "POLYWEATHER_OBSERVATION_COLLECTOR_MADIS_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_MADIS_SEC:-300}" in compose
assert "POLYWEATHER_SERVICE_ROLE: collector" in collector_block
assert "POLYWEATHER_SCAN_TERMINAL_PREWARM_ENABLED: 'false'" in bot_block
assert "POLYWEATHER_SCAN_TERMINAL_PREWARM_ENABLED: 'false'" in web_block
assert "POLYWEATHER_SCAN_TERMINAL_PREWARM_ENABLED: 'false'" in collector_block
assert "POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED: 'false'" in bot_block
assert "POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED: 'false'" in web_block
assert "POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED: 'true'" in collector_block
assert "POLYWEATHER_CITY_DETAIL_BATCH_CONCURRENCY: ${POLYWEATHER_CITY_DETAIL_BATCH_CONCURRENCY:-1}" in web_block
assert "POLYWEATHER_CITY_DETAIL_BATCH_GLOBAL_CONCURRENCY: ${POLYWEATHER_CITY_DETAIL_BATCH_GLOBAL_CONCURRENCY:-1}" in web_block
assert "POLYWEATHER_CITY_DETAIL_BATCH_PARTIAL_TIMEOUT_MS: ${POLYWEATHER_CITY_DETAIL_BATCH_PARTIAL_TIMEOUT_MS:-3000}" in web_block
assert "UVICORN_WORKERS: ${UVICORN_WORKERS:-2}" in web_block
assert "POLYWEATHER_COLLECTOR_PATCH_ENDPOINT: ''" in bot_block
assert "POLYWEATHER_COLLECTOR_PATCH_ENDPOINT: ''" in web_block
assert (
"POLYWEATHER_COLLECTOR_PATCH_ENDPOINT: "
"${POLYWEATHER_COLLECTOR_PATCH_ENDPOINT:-http://polyweather_web:8000/api/internal/collector-patch}"
in collector_block
)
assert "command: python -m web.observation_collector_worker" in collector_block
assert "POLYWEATHER_OBSERVATION_COLLECTOR_AMSC_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_AMSC_SEC:-180}" in collector_block
assert "POLYWEATHER_OBSERVATION_COLLECTOR_MADIS_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_MADIS_SEC:-300}" in collector_block
def test_scan_terminal_backend_timeout_returns_before_next_proxy_abort():
@@ -152,6 +183,7 @@ def test_deploy_script_retries_compose_recreate_races():
assert "compose_up_retry()" in script
assert "removal of container .* is already in progress" in script
assert 'compose_up_retry "backend services" -d --no-deps polyweather_web polyweather' in script
assert 'compose_up_retry "observation collector" -d --no-deps polyweather_collector' in script
assert 'compose_up_retry "frontend" -d --no-deps polyweather_frontend' in script
tests/test_observation_collector.py
+32
View File
@@ -1,4 +1,5 @@
from concurrent.futures import ThreadPoolExecutor
import sqlite3
import time
@@ -130,3 +131,34 @@ def test_observation_collector_run_due_once_refreshes_panel_cache():
assert collector.run_due_once(now_ts=1180.0) == 1
assert calls == [("qingdao", False), ("qingdao", False)]
assert refreshed == ["qingdao", "qingdao"]
def test_observation_collector_worker_entrypoint_exists():
from web import observation_collector_worker
assert callable(observation_collector_worker.main)
def test_ephemeral_observation_log_writes_skip_sqlite_lock(monkeypatch, tmp_path):
from src.database.db_manager import DBManager
db = DBManager(str(tmp_path / "polyweather.db"))
def locked_connection():
raise sqlite3.OperationalError("database is locked")
monkeypatch.setattr(db, "_get_connection", locked_connection)
db.append_airport_obs(
icao="ZSQD",
city="qingdao",
temp_c=24.0,
obs_time="2026-06-08T04:00:00Z",
)
db.append_runway_obs(
icao="ZSQD",
city="qingdao",
runway="17/35",
target_runway_max=24.0,
otime_utc="2026-06-08T04:00:00Z",
)
tests/test_web_observability.py
+40 -1
View File
@@ -969,7 +969,45 @@ def test_chart_detail_payload_uses_threadpool_and_reuses_short_cache(monkeypatch
def test_city_detail_batch_partial_timeout_default_stays_below_proxy_budget(monkeypatch):
monkeypatch.delenv("POLYWEATHER_CITY_DETAIL_BATCH_PARTIAL_TIMEOUT_MS", raising=False)
assert city_api._city_detail_batch_partial_timeout_seconds() == 6.0
assert city_api._city_detail_batch_partial_timeout_seconds() == 3.0
def test_city_detail_batch_returns_busy_when_global_builder_slot_is_full(monkeypatch):
import asyncio
build_calls = 0
async def build_batch_item(city, **kwargs):
nonlocal build_calls
build_calls += 1
return city, {"city": city}
monkeypatch.setenv("POLYWEATHER_CITY_DETAIL_BATCH_GLOBAL_CONCURRENCY", "1")
monkeypatch.setattr(city_api, "_CITY_DETAIL_BATCH_BUILD_SEMAPHORE", None)
monkeypatch.setattr(city_api, "_CITY_DETAIL_BATCH_BUILD_SEMAPHORE_SIZE", 0)
monkeypatch.setattr(city_api.legacy_routes, "_assert_entitlement", lambda request: None)
monkeypatch.setattr(city_api.legacy_routes, "_normalize_city_or_404", lambda name: name.strip().lower())
monkeypatch.setattr(city_api, "_build_city_detail_batch_item_async", build_batch_item)
semaphore = city_api._city_detail_batch_build_semaphore()
assert semaphore.acquire(blocking=False) is True
try:
payload = asyncio.run(
city_api.get_city_detail_batch_payload(
object(),
cities="Paris,Shanghai",
resolution="10m",
limit=2,
)
)
finally:
semaphore.release()
assert payload["partial"] is True
assert payload["busy"] is True
assert payload["details"] == {}
assert payload["missing"] == ["paris", "shanghai"]
assert build_calls == 0
def test_city_detail_batch_endpoint_limits_backend_concurrency(monkeypatch):
@@ -1039,6 +1077,7 @@ def test_city_detail_batch_returns_completed_details_when_one_city_is_slow(monke
}
monkeypatch.setenv("POLYWEATHER_CITY_DETAIL_BATCH_PARTIAL_TIMEOUT_MS", "20")
monkeypatch.setenv("POLYWEATHER_CITY_DETAIL_BATCH_CONCURRENCY", "2")
monkeypatch.setattr(city_api.legacy_routes, "_assert_entitlement", lambda request: None)
monkeypatch.setattr(city_api.legacy_routes, "_normalize_city_or_404", lambda name: name.strip().lower())
monkeypatch.setattr(city_api, "_build_city_detail_batch_item_async", build_batch_item)
web/app_factory.py
+1 -1
View File
@@ -41,7 +41,7 @@ def _observation_collector_enabled() -> bool:
enabled = str(
os.getenv("POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED") or "true"
).strip().lower() in {"1", "true", "yes", "on"}
return enabled and _service_role() in {"web", "api", "backend"}
return enabled and _service_role() in {"collector"}
def create_app() -> FastAPI:
web/observation_collector_worker.py
+47
View File
@@ -0,0 +1,47 @@
"""Standalone observation collector process.
The FastAPI web worker must stay responsive for health checks and user-facing
API routes. High-frequency source polling runs here instead of inside the web
request process.
"""
from __future__ import annotations
import signal
import threading
from types import FrameType
from typing import Optional
from loguru import logger
from web.core import _weather
from web.observation_collector_service import start_observation_collector_loop
from web.services.city_runtime import _refresh_city_panel_cache
_STOP_EVENT = threading.Event()
def _handle_stop_signal(signum: int, _frame: Optional[FrameType]) -> None:
logger.info("observation collector worker stopping signal={}", signum)
_STOP_EVENT.set()
def main() -> None:
signal.signal(signal.SIGINT, _handle_stop_signal)
signal.signal(signal.SIGTERM, _handle_stop_signal)
thread = start_observation_collector_loop(
weather=_weather,
cache_refresher=lambda city: _refresh_city_panel_cache(city, force_refresh=False),
)
if thread is None:
logger.warning("observation collector worker started with collector disabled")
else:
logger.info("observation collector worker started thread={}", thread.name)
while not _STOP_EVENT.wait(3600):
pass
if __name__ == "__main__":
main()
web/services/city_api.py
+90 -54
View File
@@ -46,6 +46,9 @@ _CITY_DETAIL_BATCH_RESPONSE_CACHE: Dict[CityDetailBatchResponseCacheKey, Dict[st
_CITY_DETAIL_BATCH_RESPONSE_CACHE_TS: Dict[CityDetailBatchResponseCacheKey, float] = {}
_CITY_DETAIL_BATCH_RESPONSE_INFLIGHT: Dict[CityDetailBatchResponseCacheKey, "asyncio.Task[Dict[str, Any]]"] = {}
_CITY_DETAIL_BATCH_RESPONSE_LOCK = asyncio.Lock()
_CITY_DETAIL_BATCH_BUILD_SEMAPHORE: Optional[threading.BoundedSemaphore] = None
_CITY_DETAIL_BATCH_BUILD_SEMAPHORE_SIZE = 0
_CITY_DETAIL_BATCH_BUILD_SEMAPHORE_LOCK = threading.Lock()
def _city_detail_payload_cache_ttl() -> float:
@@ -933,20 +936,38 @@ async def _build_city_detail_batch_item_async(
def _city_detail_batch_concurrency() -> int:
try:
value = int(os.getenv("POLYWEATHER_CITY_DETAIL_BATCH_CONCURRENCY", "3") or "3")
value = int(os.getenv("POLYWEATHER_CITY_DETAIL_BATCH_CONCURRENCY", "1") or "1")
except ValueError:
value = 3
return max(1, min(6, value))
value = 1
return max(1, min(4, value))
def _city_detail_batch_global_concurrency() -> int:
try:
value = int(os.getenv("POLYWEATHER_CITY_DETAIL_BATCH_GLOBAL_CONCURRENCY", "1") or "1")
except ValueError:
value = 1
return max(1, min(4, value))
def _city_detail_batch_build_semaphore() -> threading.BoundedSemaphore:
global _CITY_DETAIL_BATCH_BUILD_SEMAPHORE, _CITY_DETAIL_BATCH_BUILD_SEMAPHORE_SIZE
size = _city_detail_batch_global_concurrency()
with _CITY_DETAIL_BATCH_BUILD_SEMAPHORE_LOCK:
if _CITY_DETAIL_BATCH_BUILD_SEMAPHORE is None or _CITY_DETAIL_BATCH_BUILD_SEMAPHORE_SIZE != size:
_CITY_DETAIL_BATCH_BUILD_SEMAPHORE = threading.BoundedSemaphore(size)
_CITY_DETAIL_BATCH_BUILD_SEMAPHORE_SIZE = size
return _CITY_DETAIL_BATCH_BUILD_SEMAPHORE
def _city_detail_batch_partial_timeout_seconds() -> Optional[float]:
try:
timeout_ms = int(
os.getenv("POLYWEATHER_CITY_DETAIL_BATCH_PARTIAL_TIMEOUT_MS", "6000")
or "6000"
os.getenv("POLYWEATHER_CITY_DETAIL_BATCH_PARTIAL_TIMEOUT_MS", "3000")
or "3000"
)
except ValueError:
timeout_ms = 6000
timeout_ms = 3000
if timeout_ms <= 0:
return None
return max(0.001, min(60.0, timeout_ms / 1000.0))
@@ -991,58 +1012,73 @@ async def get_city_detail_batch_payload(
detail_scope = _normalize_city_detail_scope(scope)
async def _build_uncached_payload() -> Dict[str, Any]:
semaphore = asyncio.Semaphore(_city_detail_batch_concurrency())
build_semaphore = _city_detail_batch_build_semaphore()
if not build_semaphore.acquire(blocking=False):
return {
"cities": city_names,
"details": {},
"errors": {},
"missing": list(city_names),
"partial": True,
"busy": True,
"stale_reason": "city detail batch builder is busy",
}
async def _build_with_limit(city: str) -> Tuple[str, Dict[str, Any]]:
async with semaphore:
return await _build_city_detail_batch_item_async(
city,
force_refresh=force_refresh,
market_slug=market_slug,
target_date=target_date,
resolution=resolution,
detail_scope=detail_scope,
timing_recorder=timer,
)
try:
semaphore = asyncio.Semaphore(_city_detail_batch_concurrency())
task_by_city = {
city: asyncio.create_task(_build_with_limit(city))
for city in city_names
}
task_city_lookup = {task: city for city, task in task_by_city.items()}
done, pending = await timer.measure_async(
"build_details",
lambda: asyncio.wait(
task_by_city.values(),
timeout=_city_detail_batch_partial_timeout_seconds(),
),
)
details: Dict[str, Any] = {}
errors: Dict[str, str] = {}
missing: List[str] = []
for task in done:
city = task_city_lookup[task]
try:
result_city, payload = task.result()
except Exception as exc:
errors[city] = str(exc)
continue
details[result_city] = payload
async def _build_with_limit(city: str) -> Tuple[str, Dict[str, Any]]:
async with semaphore:
return await _build_city_detail_batch_item_async(
city,
force_refresh=force_refresh,
market_slug=market_slug,
target_date=target_date,
resolution=resolution,
detail_scope=detail_scope,
timing_recorder=timer,
)
for task in pending:
city = task_city_lookup[task]
missing.append(city)
task.cancel()
task_by_city = {
city: asyncio.create_task(_build_with_limit(city))
for city in city_names
}
task_city_lookup = {task: city for city, task in task_by_city.items()}
done, pending = await timer.measure_async(
"build_details",
lambda: asyncio.wait(
task_by_city.values(),
timeout=_city_detail_batch_partial_timeout_seconds(),
),
)
details: Dict[str, Any] = {}
errors: Dict[str, str] = {}
missing: List[str] = []
for task in done:
city = task_city_lookup[task]
try:
result_city, payload = task.result()
except Exception as exc:
errors[city] = str(exc)
continue
details[result_city] = payload
missing_set = set(missing)
missing = [city for city in city_names if city in missing_set]
return {
"cities": city_names,
"details": details,
"errors": errors,
"missing": missing,
"partial": bool(missing or errors),
}
for task in pending:
city = task_city_lookup[task]
missing.append(city)
task.cancel()
missing_set = set(missing)
missing = [city for city in city_names if city in missing_set]
return {
"cities": city_names,
"details": details,
"errors": errors,
"missing": missing,
"partial": bool(missing or errors),
}
finally:
build_semaphore.release()
cache_ttl = _city_detail_batch_response_cache_ttl()
cache_key = _city_detail_batch_response_cache_key(