diff options
| -rw-r--r-- | .gitmodules | 3 | ||||
| -rw-r--r-- | api.py | 195 | ||||
| m--------- | joblib | 0 | ||||
| -rw-r--r-- | queries/match.sql | 22 | ||||
| -rw-r--r-- | queries/participant.sql | 54 | ||||
| -rw-r--r-- | queries/player.sql | 34 | ||||
| -rw-r--r-- | queries/roster.sql | 31 |
7 files changed, 179 insertions, 160 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 @@ -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 5dfa692..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 d1d81ce..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 "playerId", +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 |