diff options
Diffstat (limited to 'api.py')
| -rw-r--r-- | api.py | 21 |
1 files changed, 6 insertions, 15 deletions
@@ -54,11 +54,6 @@ class Worker(object): with open(path, "r", encoding="utf-8-sig") as file: self._queries[table] = file.read() logging.info("loaded query '%s'", table) - logging.info("creating index") - async with self._destpool.acquire() as con: - async with con.transaction(): - await con.execute("CREATE UNIQUE INDEX ON match(\"apiId\")") - await con.execute("CREATE UNIQUE INDEX ON player(\"apiId\")") async def _execute_job(self, jobid, payload, priority): """Finish a job.""" @@ -66,7 +61,6 @@ class Worker(object): explicit_player = payload["playername"] async with self._srcpool.acquire() as srccon: async with self._destpool.acquire() as destcon: - logging.debug("%s: processing '%s'", jobid, object_id) async with srccon.transaction(): async with destcon.transaction(): # 1 object in raw : n objects in web @@ -134,10 +128,10 @@ class Worker(object): VALUES ({1}, 'epoch'::timestamp) ON CONFLICT("apiId") DO UPDATE SET ("{0}") = ({1}) WHERE player.played < EXCLUDED.played - RETURNING id + RETURNING "apiId" """.format( "\", \"".join(keys), ", ".join(placeholders)) - obj_id = await conn.fetchval(query, *data.values()) + objid = await conn.fetchval(query, *data.values()) if do_upsert_date: # upsert lmcd because it was an explicit request @@ -146,7 +140,7 @@ class Worker(object): WHERE player."lastMatchCreatedDate" < $1 """, lmcd) - return obj_id + return objid async def _into(self, conn, data, table): """Insert a named tuple into a table. @@ -154,7 +148,7 @@ class Worker(object): items = list(data.items()) keys, values = [x[0] for x in items], [x[1] for x in items] placeholders = ["${}".format(i) for i, _ in enumerate(values, 1)] - query = "INSERT INTO {} (\"{}\") VALUES ({}) ON CONFLICT DO NOTHING RETURNING id".format( + query = "INSERT INTO {} (\"{}\") VALUES ({}) ON CONFLICT DO NOTHING RETURNING \"apiId\"".format( table, "\", \"".join(keys), ", ".join(placeholders)) return await conn.fetchval(query, (*data)) @@ -164,10 +158,8 @@ class Worker(object): jobtype="process") if jobid is None: raise LookupError("no jobs available") - logging.debug("%s: starting job", jobid) await self._execute_job(jobid, payload, priority) await self._queue.finish(jobid) - logging.debug("%s: finished job", jobid) async def run(self): """Start jobs forever.""" @@ -175,8 +167,7 @@ class Worker(object): try: await self._work() except LookupError: - logging.info("nothing to do, idling") - await asyncio.sleep(10) + await asyncio.sleep(1) async def start(self, number=1): """Start jobs in background.""" @@ -189,7 +180,7 @@ async def startup(): source_db, dest_db ) await worker.setup() - await worker.start(10) + await worker.start(4) logging.basicConfig(level=logging.DEBUG) loop = asyncio.get_event_loop() |