exporter.py 文件源码

def fetch(url: str,
                session: aiohttp.ClientSession,
                timeout: float = None,
                loop: AbstractEventLoop = None) -> Dict[Any, Any]:
    ''' Fetch JSON format data from a web resource and return a dict '''
    try:
        logger.debug('fetching %s', url)
        async with session.get(url, timeout=timeout) as resp:
            if not resp.status == 200:
                raise Exception('Fetch failed {}: {}'.format(resp.status, url))
            data = await resp.json()
            return data
    except asyncio.TimeoutError:
        raise Exception('Request timed out to {}'.format(url)) from None
    except aiohttp.ClientError as exc:
        raise Exception('Client error {}, {}'.format(exc, url)) from None