asyncio web client and server

asyncio_sample.py

import json
import logging
import logging.config
import os
import time
from config import LOGGING, db
from datetime import datetime, timezone

import aiohttp
from aiohttp import web

import asyncio
from datadog import initialize as statsd_initialize
from datadog import statsd
from models import FacebookPage

logger = logging.getLogger(__name__)

FETCH_INTERVAL = 10

ENV = os.getenv('ENV', 'dev')
PROD = ENV == 'prod'

FACEBOOK_ACCESS_KEY = os.getenv('FACEBOOK_ACCESS_KEY')

logging.config.dictConfig(LOGGING)


if PROD:
    statsd_initialize(
        statsd_host=os.getenv('STATSD_HOST', 'localhost'),
        statsd_port=8125,
    )


class FacebookPageTask:
    def __init__(self, page_id, name, url):
        self.page_id = page_id
        self.name = name
        self.url = url

    def is_valid(self, result):
        return 'error' not in result

    def clean_data(self, result):
        return result['fan_count']


loop = asyncio.get_event_loop()


async def fetch(session, url):
    with aiohttp.Timeout(10):
        async with session.get(url) as response:
            return await response.text(), datetime.now(timezone.utc)


async def fetch_stats(loop):
    # Fetch pages to process
    db.connect()
    query = FacebookPage.select()  # not async
    pages = (list(query))
    db.close()

    start_time = time.time()
    tasks = []
    fetched_stats = {}

    for page in pages:
        url = 'https://graph.facebook.com/v2.9/{}?fields=fan_count&access_token={}'.format(
              page.page_id, FACEBOOK_ACCESS_KEY)
        tasks.append(FacebookPageTask(
            name=page.name,
            page_id=page.page_id,
            url=url
        ))

    async with aiohttp.ClientSession(loop=loop) as session:
        results = await asyncio.gather(
            *[fetch(session, task.url) for task in tasks],
            return_exceptions=True  # default is false, that would raise
        )

        for idx, task in enumerate(tasks):
            result, timestamp = results[idx]
            if isinstance(result, Exception) or not task.is_valid(result):
                logger.error('Fetch {} FAILED'.format(task.name))
                continue

            fetched_stats[task.name] = {
                'followers': task.clean_data(json.loads(result)),
                'page_id': task.page_id,
                'fetched': timestamp.isoformat()
            }

    # Push followers metrics to statsd.
    for (page_name, stats) in fetched_stats.items():
        statsd.gauge('stats.{}.followers'.format(page_name))

    logger.info('{} page(s) stats fetched in {}s'.format(
        len(tasks),
        round(time.time() - start_time, 1))
    )


async def index_handler(request):
    return web.Response(text='ok')


async def health_handler(request):
    return web.Response(text=json.dumps({
        'name': 'facebook_followers_fetcher',
        'git_hash': os.getenv('GIT_HASH'),
        'hostname': os.getenv('HOSTNAME'),
    }))


async def start_background_tasks(app):
    while True:
        await asyncio.sleep(FETCH_INTERVAL)
        await fetch_stats(loop)


app = web.Application()

app.on_startup.append(start_background_tasks)

app.router.add_get('/healthz', health_handler)
app.router.add_get('/', index_handler)

web.run_app(app, host='0.0.0.0', port=8080, loop=loop)