diff options
| -rw-r--r-- | api.py | 13 | ||||
| -rw-r--r-- | crawler.py | 13 |
2 files changed, 19 insertions, 7 deletions
@@ -47,8 +47,8 @@ class Apigrabber(object): async with con.transaction(): await con.execute(""" INSERT INTO crawljobs(start_date, end_date, finished, region) - SELECT '2017-02-01T00:00:00Z'::TIMESTAMP, - '2017-02-01T00:00:00Z'::TIMESTAMP, + SELECT '2017-02-14T00:00:00Z'::TIMESTAMP, + '2017-02-14T00:00:00Z'::TIMESTAMP, TRUE, region FROM JSONB_TO_RECORDSET($1::JSONB) AS jsn(region TEXT) @@ -134,7 +134,7 @@ class Apigrabber(object): async with con.transaction(isolation="serializable"): # select us our job - jobdate, delta_minutes = await con.fetchrow(""" + row_res = await con.fetchrow(""" SELECT start_date, -- new job's end date LEAST(EXTRACT(EPOCH FROM (start_date-previous_end))/60, $2) -- gap in minutes or default if smaller @@ -149,6 +149,12 @@ class Apigrabber(object): WHERE start_date-previous_end>INTERVAL '0' -- get gaps ORDER BY start_date DESC LIMIT 1 """, region, default_diff) + if row_res == None: + logging.warn("%s: no jobs available. idling.", region) + await asyncio.sleep(60) # a minute TODO make this smarter + asyncio.ensure_future(self.crawl_region(region)) + + jobdate, delta_minutes = row_res delta = datetime.timedelta(minutes=delta_minutes) # store our job as pending jobid = await con.fetchval(""" @@ -197,6 +203,7 @@ class Apigrabber(object): # TODO: respawn a worker if it dies because of connection issues # TODO: insert API version (force update if changed) # TODO: create database indices (id & shardId & type) + # TODO: make workers switch regions flexibly to meet demand? for region in self.regions: await self.request_update(region) @@ -1,6 +1,7 @@ #!/usr/bin/python import asyncio +import logging import aiohttp TOKEN = "aaa.bbb.ccc" @@ -32,10 +33,14 @@ class Crawler(object): "Accept": "application/vnd.api+json", "Content-Encoding": "gzip" } - async with session.get(self._apiurl + path, headers=headers, - params=params) as response: - assert response.status == 200 - return await response.json() + try: + async with session.get(self._apiurl + path, headers=headers, + params=params) as response: + assert response.status == 200 + return await response.json() + except aiohttp.errors.ClientResponseError: + logging.warning("error connecting to API, retrying") + return await self._req(session, path, params) async def version(self): """Returns the current API version.""" |