From b57c9a6f4e4572027aa2862bb684d6a0891ef801 Mon Sep 17 00:00:00 2001 From: schneefux Date: Sun, 5 Mar 2017 10:36:38 +0100 Subject: run jobs in batches (#57) --- api.py | 47 +++++++++++++++++++++++++---------------------- crawler.py | 1 + joblib | 2 +- 3 files changed, 27 insertions(+), 23 deletions(-) diff --git a/api.py b/api.py index f66b880..2e1c3de 100644 --- a/api.py +++ b/api.py @@ -50,33 +50,36 @@ class Apigrabber(joblib.worker.Worker): try: async for data in api.matches(region=payload["region"], params=payload["params"]): - matchids = await con.fetch(self._insertquery, json.dumps(data)) - logging.debug("%s: inserted %s matches from API into database", - jobid, len(matchids)) - for matchid in matchids: - await self._queue.request(jobtype="process", - priority=priority, - payload={ - "id": matchid["id"], - "playername": playername - }) + async with con.transaction(): + matchids = await con.fetch( + self._insertquery, json.dumps(data)) + logging.debug("%s: inserted %s matches from API into database", + jobid, len(matchids)) + payloads = [{ + "id": mat["id"], + "playername": playername + } for mat in matchids] + await self._queue.request(jobtype="process", + payload=payloads, + priority=priority) except crawler.ApiError as error: raise joblib.worker.JobFailed(error.args[0]) async def startup(): - worker = Apigrabber( - apitoken=os.environ["VAINSOCIAL_APITOKEN"] - ) - await worker.connect( - host=os.environ["POSTGRESQL_HOST"], - port=os.environ["POSTGRESQL_PORT"], - user=os.environ["POSTGRESQL_USER"], - password=os.environ["POSTGRESQL_PASSWORD"], - database=os.environ["POSTGRESQL_DB"] - ) - await worker.setup() - await worker.start(2) + for _ in range(1): + worker = Apigrabber( + apitoken=os.environ["VAINSOCIAL_APITOKEN"] + ) + await worker.connect( + host=os.environ["POSTGRESQL_HOST"], + port=os.environ["POSTGRESQL_PORT"], + user=os.environ["POSTGRESQL_USER"], + password=os.environ["POSTGRESQL_PASSWORD"], + database=os.environ["POSTGRESQL_DB"] + ) + await worker.setup() + await worker.start() logging.basicConfig( diff --git a/crawler.py b/crawler.py index 6a588c7..88bc93e 100644 --- a/crawler.py +++ b/crawler.py @@ -48,6 +48,7 @@ class Crawler(object): aiohttp.errors.ClientResponseError, aiohttp.errors.ClientOSError): # API bug? + logging.warning("API error, retrying") pass await asyncio.sleep(10) diff --git a/joblib b/joblib index 8d58a4f..7799048 160000 --- a/joblib +++ b/joblib @@ -1 +1 @@ -Subproject commit 8d58a4f5f754b6541af84478a2df3fba70db94be +Subproject commit 7799048a5927670b9492f80cdbbf2fd8d319c506 -- cgit v1.3.1