INNER CODE UNIT · Python

_with_timeout

marc-shade/world-intel-mcp · src/world_intel_mcp/dashboard/app.py:188

    async def _with_timeout(name: str, coro, timeout: float = 45.0):
        effective = 90.0 if name in _SLOW_SOURCES else timeout
        try:
            return await asyncio.wait_for(coro, timeout=effective)
        except asyncio.TimeoutError:
            logger.warning("Dashboard fetch %s timed out after %.0fs", name, effective)
            return {"error": f"timeout after {effective}s", "_timeout": True}

    gathered = await asyncio.gather(
        *[_with_timeout(name, c) for name, c in zip(coros.keys(), coros.values())],
        return_exceptions=True,
    )

    result: dict = {}
    for name, data in zip(coros.keys(), gathered):
        if isinstance(data, Exception):
            logger.warning("Dashboard fetch %s failed: %s", name, data)
            result[name] = {"error": type(data).__name__ + ": " + str(data)[:120]}

View source record →