diff options
| -rw-r--r-- | api.py | 18 | ||||
| -rw-r--r-- | crawler.py | 53 | ||||
| -rw-r--r-- | database.py | 75 |
3 files changed, 90 insertions, 56 deletions
@@ -1,10 +1,18 @@ #!/usr/bin/python +import asyncio import database import crawler -db = database.Database("dbname=vgstats user=vgstats") -api = crawler.Crawler() -#db.upsert("na", api.matches(params={"filter[playerNames]": "fallingcomets"}), True) -db.upsert("na", api.matches(), True) -print(db.select("SELECT data->'attributes'->>'name' as name FROM player WHERE CAST(data->'attributes'->'stats'->>'winStreak' AS integer) > 10")) + +async def main(): + db = database.Database() + await db.connect("postgres://vgstats@localhost/vgstats") + api = crawler.Crawler() + matches = await api.matches() + await db.upsert("na", matches, True) + print(await db.select("SELECT data->'attributes'->>'name' as name FROM player WHERE CAST(data->'attributes'->'stats'->>'winStreak' AS integer) > 3")) + +loop = asyncio.get_event_loop() +loop.run_until_complete(main()) +loop.close() @@ -1,11 +1,13 @@ #!/usr/bin/python -import requests +import asyncio import datetime +import aiohttp TOKEN = "aaa.bbb.ccc" APIURL = "https://api.dc01.gamelockerapp.com/" + class Crawler(object): def __init__(self): """Sets constants.""" @@ -14,9 +16,11 @@ class Crawler(object): self._lastquery = datetime.datetime(1, 1, 1) self._pagelimit = 50 - def _req(self, path, params=None): + async def _req(self, session, path, params=None): """Sends an API request and returns the response dict. + :param session: aiohttp client session. + :type session: :class:`aiohttp.ClientSession` :param path: URL path. :type path: str :param params: (optional) Request parameters. @@ -30,12 +34,13 @@ class Crawler(object): "Accept": "application/vnd.api+json", "Content-Encoding": "gzip" } - http = requests.get(self._apiurl + path, headers=headers, - params=params) - http.raise_for_status() - return http.json() + async with session.get(self._apiurl + path, headers=headers, + params=params) as response: + if response.status != 200: + return None + return await response.json() - def matches(self, region="na", params=None): + async def matches(self, region="na", params=None): """Queries the API for matches and their related data. :param region: (optional) Region where the matches were played. @@ -48,20 +53,32 @@ class Crawler(object): """ if params is None: params = dict() - resp = [] params["page[limit]"] = self._pagelimit params["page[offset]"] = 0 - while True: - # go one page forward until 404 - try: - json = self._req("shards/" + region + "/matches", params) - except requests.exceptions.HTTPError: - break - resp += json["data"] + json["included"] - params["page[offset]"] += params["page[limit]"] - return resp + batchsize = 100 # FIXME We don't know how long we can paginate + + tasks = [] + data = [] + # fire a lot of http requests + async with aiohttp.ClientSession() as session: + for _ in range(0, batchsize): + params["page[offset]"] += params["page[limit]"] + task = asyncio.ensure_future( + self._req( + session, + "shards/" + region + "/matches", + params)) + tasks.append(task) + + results = await asyncio.gather(*tasks) + for res in results: + if res is None: + continue + data += res["data"] + res["included"] + + return data - def matches_new(self, region="na"): + async def matches_new(self, region="na"): """Queries the API for new matches since the last query. :param region: see `matches` diff --git a/database.py b/database.py index 396701a..8cdf1f8 100644 --- a/database.py +++ b/database.py @@ -1,77 +1,85 @@ #!/usr/bin/python -import psycopg2 -import psycopg2.extras +import asyncio +import json +import asyncpg + class Database(object): """Database wrapper class""" - def __init__(self, connstring): + def __init__(self): + self._pool = None + + async def connect(self, connstring): """Connects to the database. :param connstring: Connection string containing user and database. :type connstring: str """ - self._connection = psycopg2.connect(connstring) - self._c = self._connection.cursor() + self._pool = await asyncpg.create_pool(connstring) - def upsert_type(self, shard, json, objtype, many=False): + async def upsert_type(self, shard, obj, objtype, many=False): """Upserts an object of given `objtype` into the corresponding database. :param shard: Region in which an object's id is unique. :type shard: str - :param json: Object to upsert. - :type json: dict + :param obj: Object to upsert. + :type obj: dict or list :param objtype: Object type and table name. :type objtype: str - :param many: (optional) Whether `json` is an array + :param many: (optional) Whether `obj` is a list of objects of the same objtype. :type many: bool """ if not many: - json = [json] - - arr = [{ - "id": shard + j["id"], - "data": psycopg2.extras.Json(j) - } for j in json] + obj = [obj] - self._c.execute("CREATE TABLE IF NOT EXISTS " + objtype + - " (id TEXT PRIMARY KEY, data json)") + arr = [[ + shard + j["id"], + json.dumps(j) + ] for j in obj] - self._c.executemany("INSERT INTO " + objtype + " " + - "VALUES (%(id)s, %(data)s) " + - "ON CONFLICT (id) DO " + - "UPDATE SET data=%(data)s", - arr) - self._connection.commit() + async with self._pool.acquire() as conn: + async with conn.transaction(): + await conn.execute("CREATE TABLE IF NOT EXISTS " + objtype + + " (id TEXT PRIMARY KEY, data json)") + await conn.executemany("INSERT INTO " + objtype + " " + + "VALUES ($1, $2) " + + "ON CONFLICT (id) DO " + + "UPDATE SET data=$2", + arr) - def upsert(self, shard, json, many=False): + async def upsert(self, shard, obj, many=False): """Upserts an object into the corresponding database. :param shard: Region in which an object's id is unique. :type shard: str - :param json: Object to upsert. - :type json: dict - :param many: (optional) Whether `json` is an array + :param obj: Object to upsert. + :type obj: dict + :param many: (optional) Whether `obj` is a list of objects. :type many: bool """ if not many: - json = [json] + obj = [obj] objectmap = dict() # figure out the type of each object and sort into map - for j in json: + for j in obj: objtype = j["type"] if objtype not in objectmap: objectmap[objtype] = [] objectmap[objtype].append(j) # execute bulk upsert for each type + tasks = [] for objtype, objects in objectmap.items(): - self.upsert_type(shard, objects, objtype, True) + task = asyncio.ensure_future( + self.upsert_type(shard, objects, objtype, True)) + tasks.append(task) + await asyncio.gather(*tasks) - def select(self, query): + async def select(self, query): """Returns the result of an SQL query. :param query: SQL query to execute. @@ -79,5 +87,6 @@ class Database(object): :return: List of results. :rtype: list of dict """ - self._c.execute(query) - return self._c.fetchall() + async with self._pool.acquire() as conn: + async with conn.transaction(): + return await conn.fetch(query) |