Merge pull request #22 from vainglorygame/rewrite

rewrite to use joblib (fixes #20), use updated raw schema, check for …

8 files changed, 179 insertions, 181 deletions

diff --git a/.gitmodules b/.gitmodules
new file mode 100644
index 0000000..52948d0
--- /dev/null
+++ b/.gitmodules
@@ -0,0 +1,3 @@
+[submodule "joblib"]
+	path = joblib
+	url = https://github.com/vainglorygame/joblib
diff --git a/api.py b/api.py
index 90b4fbf..5c7528d 100644
--- a/api.py
+++ b/api.py
@@ -1,11 +1,15 @@
-#!/usr/bin/python
+#!/usr/bin/python3
 
+import asyncio
 import os
 import glob
 import logging
-import asyncio
+import json
 import asyncpg
 
+import joblib.joblib
+
+
 source_db = {
     "host": os.environ.get("POSTGRESQL_SOURCE_HOST") or "vaindock_postgres_raw",
     "port": os.environ.get("POSTGRESQL_SOURCE_PORT") or 5532,
@@ -23,128 +27,119 @@ dest_db = {
 }
 
 
-class Processor(object):
+class Worker(object):
     def __init__(self):
+        self._queue = None
+        self._srcpool = None
+        self._destpool = None
         self._queries = {}
-        self._srcpool = self._destpool = None
+
+    async def connect(self, sourcea, desta):
+        """Connect to database."""
+        logging.info("connecting to database")
+        self._queue = joblib.joblib.JobQueue()
+        await self._queue.connect(**sourcea)
+        await self._queue.setup()
+        self._srcpool = await asyncpg.create_pool(**sourcea)
+        self._destpool = await asyncpg.create_pool(**desta)
 
     async def setup(self):
-        """Load .sql files from queries/ directory.
-        File name is the table to insert into."""
-        self._queries = {}
+        """Initialize the database."""
         scriptroot = os.path.realpath(
             os.path.join(os.getcwd(), os.path.dirname(__file__)))
-        # glob *.sql
-        for fp in glob.glob(scriptroot + "/queries/*.sql"):
+        for path in glob.glob(scriptroot + "/queries/*.sql"):
             # utf-8-sig is used by pgadmin, doesn't hurt to specify
-            with open(fp, "r", encoding="utf-8-sig") as file:
-                table = os.path.splitext(os.path.basename(fp))[0]
-                logging.info("loaded query for '%s'", table)
+            # file names: raw target table
+            table = os.path.splitext(os.path.basename(path))[0]
+            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\")")
 
-        # prepare worker queue
-        async with self._srcpool.acquire() as con:
-            await con.execute("""
-                CREATE TABLE IF NOT EXISTS processjobs
-                    (id SERIAL, tablename TEXT, finished BOOL)
-                """)
-            # purge failed jobs from last time and retry
-            await con.execute("DELETE FROM processjobs WHERE finished=false")
-
-    async def connect(self, source, dest):
-        """Connect to database by arguments."""
-        self._srcpool = await asyncpg.create_pool(**source)
-        self._destpool = await asyncpg.create_pool(**dest)
-
-    async def acquire_job(self):
-        """Reserve a job and return (id, table)."""
-        async with self._srcpool.acquire() as con:
-            while True:
-                try:
-                    async with con.transaction(isolation="serializable"):
-                        # select us our job
-
-                        table_name = await con.fetchval("""
-                        SELECT table_name
-                        FROM (
-                          SELECT table_name FROM information_schema.tables WHERE table_schema='public' AND
-                          (table_name ~ '^(match|roster|participant)_\w{2,3}_\d{4}_\d{2}_\d{2}$'
-                           OR table_name ~ '^(player|team)_\w{2,3}$')
-                        ) AS table_names
-                        WHERE table_name NOT IN
-                        (SELECT tablename FROM processjobs)
-                        """)
-                        if table_name == None:
-                            logging.warn("no jobs available")
-                            return (None, None)
-
-                        # store our job as pending
-                        jobid = await con.fetchval("""
-                            INSERT INTO processjobs(tablename, finished)
-                            VALUES ($1, FALSE)
-                            RETURNING id
-                        """, table_name)
-                        # exit loop
-                        return (jobid, table_name)
-                except asyncpg.exceptions.SerializationError:
-                    # job is being picked up by another worker, try again
-                    pass
-
-    async def run_job(self):
-        """Processes all data in table `table_name`."""
-        # reserve one job
-        jobid, table_name = await self.acquire_job()
-        assert jobid is not None
-        logging.info("%s: processing '%s'", jobid, table_name)
-
-        # process.
-        query_name = table_name.split("_")[0]  # `match`
+    async def _execute_job(self, jobid, payload):
+        """Finish a job."""
+        object_id = payload["id"]
+        explicit_player = payload["playername"]
         async with self._srcpool.acquire() as srccon:
             async with self._destpool.acquire() as destcon:
-                # load SQL
-                # replace generic `FROM` by `FROM partition`
-                query = self._queries[query_name].replace(
-                    "FROM \"" + query_name + "\"",
-                    "FROM \"" + table_name + "\""
-                )
-                logging.debug("%s: using query '%s'", jobid, query_name)
+                logging.debug("%s: processing '%s'", jobid, object_id)
                 async with srccon.transaction():
                     async with destcon.transaction():
-                        async for rec in srccon.cursor(query):
-                            # loop over src, insert into dest
-                            await self.into(destcon, rec, query_name)
+                        # 1 object in raw : n objects in web
+                        for table, query in self._queries.items():
+                            logging.debug("%s: running '%s' query",
+                                          jobid, table)
+                            # fetch from raw, converted to format for web table
+                            if table == "player":
+                                # upsert under special conditions
+                                data = await srccon.fetchrow(
+                                    query, object_id, explicit_player)
+                                await self._playerinto(destcon, data, table)
+                            else:
+                                data = await srccon.fetchrow(
+                                    query, object_id)
+                                # insert processed result into web table
+                                await self._into(destcon, data, table)
+                    data = await srccon.fetchrow(
+                        "DELETE FROM match WHERE id=$1", object_id)
 
-    async def run(self):
-        """Execute a function for each row that is fetched with query."""
-        async def worker():
-            """Spawn tasks forever."""
-            try:
-                await self.run_job()
-            except AssertionError:
-                logging.info("worker idling")
-                await asyncio.sleep(30)
-            asyncio.ensure_future(worker())
-
-        for _ in range(5):
-            asyncio.ensure_future(worker())
+    async def _playerinto(self, conn, data, table):
+        """Insert a player named tuple into a table.
+        Upserts specific columns."""
+        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 player ("{0}") VALUES ({1})
+            ON CONFLICT("apiId") DO UPDATE SET ("{0}") = ({1})
+            WHERE player."lastMatchCreatedDate" < EXCLUDED."lastMatchCreatedDate"
+        """.format(
+            "\", \"".join(keys), ", ".join(placeholders))
+        await conn.execute(query, (*data))
 
-    async def into(self, conn, data, table):
+    async def _into(self, conn, data, table):
         """Insert a named tuple into a table."""
         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 ({})".format(
+        query = "INSERT INTO {} (\"{}\") VALUES ({}) ON CONFLICT DO NOTHING".format(
             table, "\", \"".join(keys), ", ".join(placeholders))
         await conn.execute(query, (*data))
 
+    async def _work(self):
+        """Fetch a job and run it."""
+        jobid, payload, _ = await self._queue.acquire(jobtype="process")
+        if jobid is None:
+            raise LookupError("no jobs available")
+        logging.debug("%s: starting job", jobid)
+        await self._execute_job(jobid, payload)
+        await self._queue.finish(jobid)
+        logging.debug("%s: finished job", jobid)
+
+    async def run(self):
+        """Start jobs forever."""
+        while True:
+            try:
+                await self._work()
+            except LookupError:
+                logging.info("nothing to do, idling")
+                await asyncio.sleep(10)
+
 
-async def main():
-    pr = Processor()
-    await pr.connect(source_db, dest_db)
-    await pr.setup()
-    await pr.run()
+async def startup():
+    for _ in range(1):
+        worker = Worker()
+        await worker.connect(
+            source_db, dest_db
+        )
+        await worker.setup()
+        await worker.run()
 
 logging.basicConfig(level=logging.DEBUG)
 loop = asyncio.get_event_loop()
-loop.run_until_complete(main())
+loop.run_until_complete(startup())
 loop.run_forever()
diff --git a/joblib b/joblib
new file mode 160000
+Subproject f7b858a53e0b3233e04186d18d309f8403ea99d
diff --git a/queries/match.sql b/queries/match.sql
index a07652f..1e6d37f 100644
--- a/queries/match.sql
+++ b/queries/match.sql
@@ -1,15 +1,15 @@
 SELECT
 
-id AS "apiId",
-(attributes->>'createdAt')::timestamp as "createdAt",
-(attributes->>'duration')::int AS "duration",
-attributes->>'gameMode' AS "gameMode",
-COALESCE(NULLIF(attributes->>'patchVersion', ''), '0') AS "patchVersion",
-attributes->>'shardId' AS "shardId",
-attributes->'stats'->>'endGameReason' AS "endGameReason",
-attributes->'stats'->>'queue' AS "queue" ,
+match.id AS "apiId",
+(match.data->'data'->'attributes'->>'createdAt')::timestamp as "createdAt",
+(match.data->'data'->'attributes'->>'duration')::int AS "duration",
+match.data->'data'->'attributes'->>'gameMode' AS "gameMode",
+COALESCE(NULLIF(match.data->'data'->'attributes'->>'patchVersion', ''), '0') AS "patchVersion",
+match.data->'data'->'attributes'->>'shardId' AS "shardId",
+match.data->'data'->'attributes'->'stats'->>'endGameReason' AS "endGameReason",
+match.data->'data'->'attributes'->'stats'->>'queue' AS "queue",
 
-relationships->'rosters'->'data'->0->>'id' as "roster_1",
-relationships->'rosters'->'data'->1->>'id' as "roster_2"
+match.data->'relations'->0->'data'->>'id' as "roster_1",
+match.data->'relations'->1->'data'->>'id' as "roster_2"
 
-FROM "match"
+FROM match WHERE id=$1
diff --git a/queries/participant.sql b/queries/participant.sql
index b7685a9..e11179c 100644
--- a/queries/participant.sql
+++ b/queries/participant.sql
@@ -1,28 +1,34 @@
+WITH rosters AS (SELECT
+  JSONB_ARRAY_ELEMENTS(match.data->'relations') AS roster
+FROM match WHERE id=$1),
+participants AS (SELECT
+  JSONB_ARRAY_ELEMENTS(rosters.roster->'relations') AS participant
+FROM rosters)
 SELECT
 
-id AS "apiId",
-relationships->'player'->'data'->>'id' AS "player_apiId",
+participant->'data'->>'id' AS "apiId",
+participant->'relations'->0->'data'->>'id' AS "player_apiId",
 
-attributes->>'actor' AS "hero",
-(attributes->'stats'->>'assists')::int AS "assists",
-COALESCE(NULLIF(attributes->'stats'->>'crystalMineCaptures', ''), '0')::int AS "crystalMineCaptures",
-(attributes->'stats'->>'deaths')::int AS "deaths",
-(attributes->'stats'->>'farm')::float AS "farm",
-(attributes->'stats'->>'firstAfkTime')::float AS "firstAfkTime",
-COALESCE(NULLIF(attributes->'stats'->>'goldMineCaptures', ''), '0')::int AS "goldMineCaptures",
-COALESCE(NULLIF(attributes->'stats'->>'jungleKills', ''), '0')::int AS "jungleKills",
-COALESCE(NULLIF(attributes->'stats'->>'karmaLevel', ''), '0')::int AS "karmaLevel",
-COALESCE(NULLIF(attributes->'stats'->>'kills', ''), '0')::int AS "kills",
-COALESCE(NULLIF(attributes->'stats'->>'krakenCaptures', ''), '0')::int AS "krakenCaptures",
-COALESCE(NULLIF(attributes->'stats'->>'level', ''), '0')::int AS "level",
-COALESCE(NULLIF(attributes->'stats'->>'minionKills', ''), '0')::int AS "minionKills",
-COALESCE(NULLIF(attributes->'stats'->>'skillTier', ''), '0')::int AS "skillTier",
-COALESCE(NULLIF(attributes->'stats'->>'skinKey', ''), '0') AS "skinKey",
-(attributes->'stats'->>'wentAfk')::bool::bool AS "wentAfk",
-(attributes->'stats'->>'winner')::bool::bool AS "winner",
-attributes->'stats'->'itemGrants' AS "itemGrants",
-attributes->'stats'->'itemSells' AS "itemSells",
-attributes->'stats'->'itemUses' AS "itemUses",
-attributes->'stats'->'items' AS "items"
+participant->'data'->'attributes'->>'actor' AS "hero",
+(participant->'data'->'attributes'->'stats'->>'assists')::int AS "assists",
+COALESCE(NULLIF(participant->'data'->'attributes'->'stats'->>'crystalMineCaptures', ''), '0')::int AS "crystalMineCaptures",
+(participant->'data'->'attributes'->'stats'->>'deaths')::int AS "deaths",
+(participant->'data'->'attributes'->'stats'->>'farm')::float AS "farm",
+(participant->'data'->'attributes'->'stats'->>'firstAfkTime')::float AS "firstAfkTime",
+COALESCE(NULLIF(participant->'data'->'attributes'->'stats'->>'goldMineCaptures', ''), '0')::int AS "goldMineCaptures",
+COALESCE(NULLIF(participant->'data'->'attributes'->'stats'->>'jungleKills', ''), '0')::int AS "jungleKills",
+COALESCE(NULLIF(participant->'data'->'attributes'->'stats'->>'karmaLevel', ''), '0')::int AS "karmaLevel",
+COALESCE(NULLIF(participant->'data'->'attributes'->'stats'->>'kills', ''), '0')::int AS "kills",
+COALESCE(NULLIF(participant->'data'->'attributes'->'stats'->>'krakenCaptures', ''), '0')::int AS "krakenCaptures",
+COALESCE(NULLIF(participant->'data'->'attributes'->'stats'->>'level', ''), '0')::int AS "level",
+COALESCE(NULLIF(participant->'data'->'attributes'->'stats'->>'minionKills', ''), '0')::int AS "minionKills",
+COALESCE(NULLIF(participant->'data'->'attributes'->'stats'->>'skillTier', ''), '0')::int AS "skillTier",
+COALESCE(NULLIF(participant->'data'->'attributes'->'stats'->>'skinKey', ''), '0') AS "skinKey",
+(participant->'data'->'attributes'->'stats'->>'wentAfk')::bool AS "wentAfk",
+(participant->'data'->'attributes'->'stats'->>'winner')::bool AS "winner",
+participant->'data'->'attributes'->'stats'->'itemGrants' AS "itemGrants",
+participant->'data'->'attributes'->'stats'->'itemSells' AS "itemSells",
+participant->'data'->'attributes'->'stats'->'itemUses' AS "itemUses",
+participant->'data'->'attributes'->'stats'->'items' AS "items"
 
-FROM participant
+FROM participants
diff --git a/queries/player.sql b/queries/player.sql
index 992cb44..0619ee6 100644
--- a/queries/player.sql
+++ b/queries/player.sql
@@ -1,15 +1,29 @@
+WITH rosters AS (SELECT
+  match.data->'data'->'attributes'->>'createdAt' AS matchdate,
+  JSONB_ARRAY_ELEMENTS(match.data->'relations') AS roster
+FROM match WHERE id=$1),
+participants AS (SELECT
+  matchdate,
+  JSONB_ARRAY_ELEMENTS(rosters.roster->'relations') AS participant
+FROM rosters),
+players AS (SELECT
+  matchdate,
+  JSONB_ARRAY_ELEMENTS(participants.participant->'relations') AS player
+FROM participants)
+
 SELECT
 
-id AS "apiId",
-attributes->>'name' AS "name",
-(attributes->'stats'->>'level')::int AS "level",
-(attributes->'stats'->>'xp')::int AS "xp",
-(attributes->'stats'->>'played')::int AS "played",
-(attributes->'stats'->>'played_ranked')::int AS "playedRanked",
-(attributes->'stats'->>'played')::int - (attributes->'stats'->>'played_ranked')::int AS "playedCasual",
-(attributes->'stats'->>'wins')::int AS "wins",
-(attributes->'stats'->>'lifetimeGold')::float AS "lifetimeGold",
+players.player->'data'->>'id' AS "apiId",
+players.player->'data'->'attributes'->>'name' AS "name",
+(players.player->'data'->'attributes'->'stats'->>'level')::int AS "level",
+(players.player->'data'->'attributes'->'stats'->>'xp')::int AS "xp",
+(players.player->'data'->'attributes'->'stats'->>'played')::int AS "played",
+(players.player->'data'->'attributes'->'stats'->>'played_ranked')::int AS "playedRanked",
+(players.player->'data'->'attributes'->'stats'->>'played')::int - (players.player->'data'->'attributes'->'stats'->>'played_ranked')::int AS "playedCasual",
+(players.player->'data'->'attributes'->'stats'->>'wins')::int AS "wins",
+(players.player->'data'->'attributes'->'stats'->>'lifetimeGold')::float AS "lifetimeGold",
+CASE WHEN players.player->'data'->'attributes'->>'name'=$2 THEN matchdate ELSE 'epoch'::timestamp::text END AS "lastMatchCreatedDate",
 
 0 AS "streak"
 
-FROM "player"
+FROM players
diff --git a/queries/roster.sql b/queries/roster.sql
index eaf0ed6..c61967b 100644
--- a/queries/roster.sql
+++ b/queries/roster.sql
@@ -1,17 +1,18 @@
+WITH rosters AS (SELECT
+  JSONB_ARRAY_ELEMENTS(match.data->'relations') AS roster
+FROM match WHERE id=$1)
 SELECT
+roster->'data'->>'id' as "apiId",
+(roster->'data'->'attributes'->'stats'->>'acesEarned')::int AS "acesEarned",
+(roster->'data'->'attributes'->'stats'->>'gold')::int AS "gold",
+(roster->'data'->'attributes'->'stats'->>'heroKills')::int AS "heroKills",
+(roster->'data'->'attributes'->'stats'->>'krakenCaptures')::int AS "krakenCaptures",
+roster->'data'->'attributes'->'stats'->>'side' AS "side",
+roster->'data'->'attributes'->'stats'->>'side' AS "teamColor",
+(roster->'data'->'attributes'->'stats'->>'turretKills')::int AS "turretKills",
+(roster->'data'->'attributes'->'stats'->>'turretsRemaining')::int AS "turretsRemaining",
 
-id AS "apiId",
-(attributes->'stats'->>'acesEarned')::int AS "acesEarned",
-(attributes->'stats'->>'gold')::int AS "gold",
-(attributes->'stats'->>'heroKills')::int AS "heroKills",
-(attributes->'stats'->>'krakenCaptures')::int AS "krakenCaptures",
-attributes->'stats'->>'side' AS "side",
-attributes->'stats'->>'side' AS "teamColor",
-(attributes->'stats'->>'turretKills')::int AS "turretKills",
-(attributes->'stats'->>'turretsRemaining')::int AS "turretsRemaining",
-
-relationships->'participants'->'data'->0->>'id' AS "participant_1",
-relationships->'participants'->'data'->1->>'id' AS "participant_2",
-relationships->'participants'->'data'->2->>'id' AS "participant_3"
-
-FROM "roster"
+roster->'relations'->0->'data'->>'id' AS "participant_1",
+roster->'relations'->1->'data'->>'id' AS "participant_2",
+roster->'relations'->2->'data'->>'id' AS "participant_3"
+FROM rosters
diff --git a/ref_queries/update_roster_match_relationship.sql b/ref_queries/update_roster_match_relationship.sql
deleted file mode 100644
index a9c8cdc..0000000
--- a/ref_queries/update_roster_match_relationship.sql
+++ /dev/null
@@ -1,21 +0,0 @@
--- i am thinking how fast slow this will be and if we should do this in DB.
--- maybe we need a script to run a cursor and update records - dont know...
--- a few runs on 35K records was closing my server connection on my shitty laptop
-
--- this will update all roster with match_apiId
-update "roster" 
-set "match_apiId" = match."apiId"
-from match
-where (roster."apiId" = roster_1 or roster."apiId" = roster_2)
-
-
--- this is same as before however should only pick empty records
-update "roster" 
-set "match_apiId" = match."apiId"
-from match
-where roster."match_apiId" is not null and (roster."apiId" = roster_1 or roster."apiId" = roster_2)
-
-
-
--- this should always return a count of 2 for each match_apiId :)
-select "match_apiId", count(*) from roster group by "match_apiId"