diff options
| -rw-r--r-- | api.py | 22 | ||||
| -rw-r--r-- | crawler.py | 4 | ||||
| -rw-r--r-- | insert.sql | 8 | ||||
| m--------- | joblib | 0 |
4 files changed, 20 insertions, 14 deletions
@@ -52,7 +52,7 @@ class Worker(object): with open(root + "/insert.sql", "r", encoding="utf-8-sig") as file: self._insertquery = file.read() - async def _execute_job(self, jobid, payload): + async def _execute_job(self, jobid, payload, priority): """Finish a job.""" api = crawler.Crawler(self._apitoken) logging.debug("%s: getting matches from API", jobid) @@ -60,20 +60,26 @@ class Worker(object): async for data in api.matches(region=payload["region"], params=payload["params"]): logging.debug("%s: inserting into database", jobid) - ids = await con.fetch(self._insertquery, json.dumps(data)) - logging.info("%s: inserted %s objects", jobid, len(ids)) - object_ids = [i["id"] for i in ids] - for object_id in object_ids: + objects = await con.fetch(self._insertquery, json.dumps(data)) + logging.info("%s: inserted %s", jobid, + {t: len([s for s in objects if s["type"] == t]) + for t in set([e["type"] for e in objects])} + ) + for obj in objects: await self._queue.request(jobtype="process", - payload={"id": object_id}) + priority=priority, + payload={ + "id": obj["id"], + "type": obj["type"] + }) async def _work(self): """Fetch a job and run it.""" - jobid, payload = await self._queue.acquire(jobtype="grab") + jobid, payload, priority = await self._queue.acquire(jobtype="grab") if jobid is None: raise LookupError("no jobs available") logging.debug("%s: starting job", jobid) - await self._execute_job(jobid, payload) + await self._execute_job(jobid, payload, priority) await self._queue.finish(jobid) logging.debug("%s: finished job", jobid) @@ -12,7 +12,7 @@ class Crawler(object): """Sets constants.""" self._apiurl = APIURL self._token = token - self._pagelimit = 50 + self._pagelimit = 5 async def _req(self, session, path, params): """Sends an API request and returns the response dict. @@ -69,7 +69,7 @@ class Crawler(object): yield res - if len(res["data"]) < 50: + if len(res["data"]) < self._pagelimit: # asked for 50, got less -> exhausted break params["page[offset]"] += params["page[limit]"] @@ -68,15 +68,15 @@ insert_matches AS ( INSERT INTO match SELECT * FROM linked_matches ON CONFLICT(id) DO NOTHING - RETURNING id -- TODO conflict shouldn't happen in prod + RETURNING type, id -- TODO conflict shouldn't happen in prod ), insert_players AS ( INSERT INTO player SELECT * FROM linked_players ON CONFLICT(id) DO UPDATE SET attributes=EXCLUDED.attributes WHERE (player.attributes->'stats'->>'played')::int < (EXCLUDED.attributes->'stats'->>'played')::int - RETURNING id + RETURNING type, id ) - SELECT * FROM insert_matches + SELECT type, id FROM insert_matches UNION - SELECT * FROM insert_players + SELECT type, id FROM insert_players diff --git a/joblib b/joblib -Subproject fa1932a5ac4dd16c743ae55315aa7419297058b +Subproject f7b858a53e0b3233e04186d18d309f8403ea99d |