api: send requests synchronously

3 files changed, 63 insertions, 24 deletions

diff --git a/api.py b/api.py
index c679df9..1686204 100644
--- a/api.py
+++ b/api.py
@@ -3,13 +3,20 @@
 import asyncio
 import database
 import crawler
+import datetime
 
 
 async def main():
     db = database.Database()
     await db.connect("postgres://vgstats@localhost/vgstats")
     api = crawler.Crawler()
-    matches = await api.matches()
+    try:
+        await db.meta("last_match_update")
+    except KeyError:
+        await db.meta("last_match_update",
+                      datetime.datetime(1, 1, 1).isoformat())
+    matches = await api.matches_since(
+        await db.meta("last_match_update"))
     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"))
 
diff --git a/crawler.py b/crawler.py
index 7ec99b0..96eb8ce 100644
--- a/crawler.py
+++ b/crawler.py
@@ -1,7 +1,6 @@
 #!/usr/bin/python
 
 import asyncio
-import datetime
 import aiohttp
 
 TOKEN = "aaa.bbb.ccc"
@@ -13,7 +12,6 @@ class Crawler(object):
         """Sets constants."""
         self._apiurl = APIURL
         self._token = TOKEN
-        self._lastquery = datetime.datetime(1, 1, 1)
         self._pagelimit = 50
 
     async def _req(self, session, path, params=None):
@@ -36,8 +34,7 @@ class Crawler(object):
         }
         async with session.get(self._apiurl + path, headers=headers,
                                params=params) as response:
-            if response.status != 200:
-                return None
+            assert response.status == 200
             return await response.json()
 
     async def matches(self, region="na", params=None):
@@ -55,35 +52,30 @@ class Crawler(object):
             params = dict()
         params["page[limit]"] = self._pagelimit
         params["page[offset]"] = 0
-        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):
+            while True:
                 params["page[offset]"] += params["page[limit]"]
-                task = asyncio.ensure_future(
-                    self._req(
-                        session,
-                        "shards/" + region + "/matches",
-                        params))
-                tasks.append(task)
+                try:
+                    res = await self._req(session,
+                                          "shards/" + region + "/matches",
+                                          params)
+                except AssertionError:
+                    break
 
-            results = await asyncio.gather(*tasks)
-            for res in results:
-                if res is None:
-                    continue
                 data += res["data"] + res["included"]
 
         return data
 
-    async def matches_new(self, region="na"):
-        """Queries the API for new matches since the last query.
+    async def matches_since(self, date, region="na"):
+        """Queries the API for new matches since the given date.
 
         :param region: see `matches`
         :type region: str
+        :param date: Start date in ISO8601 format.
+        :type date: str
+        :return: Processed API response
+        :rtype: list of dict
         """
-        lastdate = self._lastquery.replace(microsecond=0).isoformat() + "Z"
-        self._lastquery = datetime.datetime.now()
-        return self.matches(region, {"filter[createdAt-start]": lastdate})
+        return await self.matches(region, {"filter[createdAt-start]": date})
diff --git a/database.py b/database.py
index 8cdf1f8..8fad408 100644
--- a/database.py
+++ b/database.py
@@ -79,6 +79,46 @@ class Database(object):
             tasks.append(task)
         await asyncio.gather(*tasks)
 
+    async def meta(self, key, value=None):
+        """Sets or gets data into a dictionary-like database object.
+
+        :param key: ID of the value.
+        :type key: str
+        :param value: (optional) Value to set.
+        :type value: object
+        :return: If `value` is None, the value the database entry.
+        :rtype: object
+        """
+        async with self._pool.acquire() as conn:
+            async with conn.transaction():
+                await conn.execute("CREATE TABLE IF NOT EXISTS meta " +
+                                   "(key TEXT PRIMARY KEY, value json)")
+                if value is not None:
+                    value = json.dumps(value)
+                    await conn.execute("INSERT INTO meta(key, value) " +
+                                       "VALUES ($1, $2) " +
+                                       "ON CONFLICT (key) DO " +
+                                       "UPDATE SET value=$2",
+                                       key, value)
+                else:
+                    result = await conn.fetch("SELECT value FROM meta " +
+                                              "WHERE key='" + key + "'")
+                    if len(result) == 0:
+                        raise KeyError
+                    return result[0]["value"]
+
+    async def execute(self, query, args):
+        """Runs an SQL statement.
+
+        :param query: SQL query to execute.
+        :type query: str
+        :param args: Query arguments.
+        :type args: list of objects
+        """
+        async with self._pool.acquire() as conn:
+            async with conn.transaction():
+                await conn.execute(query, args)
+
     async def select(self, query):
         """Returns the result of an SQL query.