diff options
| -rw-r--r-- | .gitignore | 91 | ||||
| -rw-r--r-- | process.py | 86 |
2 files changed, 177 insertions, 0 deletions
diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..9a05e2d --- /dev/null +++ b/.gitignore @@ -0,0 +1,91 @@ +# Byte-compiled / optimized / DLL files +__pycache__/ +*.py[cod] +*$py.class + +# C extensions +*.so + +# Distribution / packaging +.Python +env/ +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +*.egg-info/ +.installed.cfg +*.egg + +# PyInstaller +# Usually these files are written by a python script from a template +# before PyInstaller builds the exe, so as to inject date/other infos into it. +*.manifest +*.spec + +# Installer logs +pip-log.txt +pip-delete-this-directory.txt + +# Unit test / coverage reports +htmlcov/ +.tox/ +.coverage +.coverage.* +.cache +nosetests.xml +coverage.xml +*,cover +.hypothesis/ + +# Translations +*.mo +*.pot + +# Django stuff: +*.log +local_settings.py + +# Flask stuff: +instance/ +.webassets-cache + +# Scrapy stuff: +.scrapy + +# Sphinx documentation +docs/_build/ + +# PyBuilder +target/ + +# Jupyter Notebook +.ipynb_checkpoints + +# pyenv +.python-version + +# celery beat schedule file +celerybeat-schedule + +# dotenv +.env + +# virtualenv +.venv/ +venv/ +ENV/ + +# Spyder project settings +.spyderproject + +# Rope project settings +.ropeproject diff --git a/process.py b/process.py new file mode 100644 index 0000000..342d838 --- /dev/null +++ b/process.py @@ -0,0 +1,86 @@ +#!/usr/bin/python + +import os +import logging +import asyncio +import asyncpg + +source_db = { + "host": os.environ.get("POSTGRESQL_SOURCE_HOST") or "localhost", + "port": os.environ.get("POSTGRESQL_SOURCE_PORT") or 5532, + "user": os.environ.get("POSTGRESQL_SOURCE_USER") or "vgstats", + "password": os.environ.get("POSTGRESQL_SOURCE_PASSWORD") or "vgstats", + "database": os.environ.get("POSTGRESQL_SOURCE_DB") or "vgstats" +} + +dest_db = { + "host": os.environ.get("POSTGRESQL_DEST_HOST") or "localhost", + "port": os.environ.get("POSTGRESQL_DEST_PORT") or 5432, + "user": os.environ.get("POSTGRESQL_DEST_USER") or "vainsocial", + "password": os.environ.get("POSTGRESQL_DEST_PASSWORD") or "vainsocial", + "database": os.environ.get("POSTGRESQL_DEST_DB") or "vainsocial" +} + + +class Database(object): + async def connect(self, **connect_kwargs): + """Connect to database by arguments.""" + self._pool = await asyncpg.create_pool(**connect_kwargs) + + async def each(self, fetchquery, func, **func_args): + """Execute a function for each row that is fetched with query.""" + # TODO use iterator instead of callback + async with self._pool.acquire() as conn: + async with conn.transaction(): + tasks = [] + async for record in conn.cursor(fetchquery): + tasks.append( + asyncio.ensure_future(func(record, **func_args)) + ) + await asyncio.gather(*tasks) + + async def into(self, data, table): + """Insert a named tuple into a database.""" + 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( + table, "\", \"".join(keys), ", ".join(placeholders)) + + async with self._pool.acquire() as conn: + await conn.fetch(query, (*data)) + + +async def process(): + sdb = Database() + await sdb.connect(**source_db) + ddb = Database() + await ddb.connect(**dest_db) + + selqry = """ +SELECT + +(data->'attributes'->>'duration')::int AS "duration", +data->'attributes'->>'gameMode' AS "gameMode", +COALESCE(NULLIF(data->'attributes'->>'patchVersion', ''), '0')::int AS "patchVersion", +data->'attributes'->>'shardId' AS "shard", +data->'attributes'->'stats'->>'endGameReason' AS "result", +data->'attributes'->'stats'->>'queue' AS "queue", + +false AS "anyAFK", +'' AS "winningTeam", +0 AS "krakenCaptures", +0 AS "laneMinionsSlayed", +0 AS "jungleMinionsSlayed", +0 AS "heroDeaths" + +FROM match LIMIT 100 + """ + + await sdb.each(selqry, ddb.into, table="match") + +if __name__ == "__main__": + loop = asyncio.get_event_loop() + loop.run_until_complete( + process() + ) |