1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
|
#!/usr/bin/python
import asyncio
import os
import logging
import json
import asyncpg
import crawler
import joblib.joblib
class Worker(object):
def __init__(self, apitoken):
self._apitoken = apitoken
self._queue = None
self._pool = None
self._insertquery = ""
async def connect(self, **args):
"""Connect to database."""
logging.info("connecting to database")
self._queue = joblib.joblib.JobQueue()
await self._queue.connect(**args)
await self._queue.setup()
self._pool = await asyncpg.create_pool(**args)
async def setup(self):
"""Initialize the database."""
logging.info("initializing database")
async with self._pool.acquire() as con:
await con.execute("""
CREATE TABLE IF NOT EXISTS
match (id TEXT PRIMARY KEY, data JSONB)
""")
root = os.path.realpath(
os.path.join(os.getcwd(), os.path.dirname(__file__)))
with open(root + "/insert.sql", "r", encoding="utf-8-sig") as file:
self._insertquery = file.read()
async def _execute_job(self, jobid, payload, priority):
"""Finish a job."""
api = crawler.Crawler(self._apitoken)
logging.debug("%s: getting matches from API", jobid)
async with self._pool.acquire() as con:
async for data in api.matches(region=payload["region"],
params=payload["params"]):
logging.debug("%s: inserting into database", jobid)
matchids = await con.fetch(self._insertquery, json.dumps(data))
logging.info("%s: inserted %s matches", jobid, len(matchids))
for matchid in matchids:
await self._queue.request(jobtype="process",
priority=priority,
payload={
"id": matchid["id"],
})
async def _work(self):
"""Fetch a job and run it."""
jobid, payload, priority = await self._queue.acquire(jobtype="grab")
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."""
while True:
try:
await self._work()
except LookupError:
logging.info("nothing to do, idling")
await asyncio.sleep(10)
async def startup():
for _ in range(1):
worker = Worker(
apitoken=os.environ["VAINSOCIAL_APITOKEN"]
)
await worker.connect(
host=os.environ["POSTGRESQL_HOST"],
port=os.environ["POSTGRESQL_PORT"],
user=os.environ["POSTGRESQL_USER"],
password=os.environ["POSTGRESQL_PASSWORD"],
database=os.environ["POSTGRESQL_DB"]
)
await worker.setup()
await worker.run()
logging.basicConfig(level=logging.DEBUG)
loop = asyncio.get_event_loop()
loop.run_until_complete(startup())
loop.run_forever()
|
