Decorators with aiohttp response and some utils

responses.py

import json
from bson import ObjectId
import datetime
from aiohttp import web
from aiohttp.client_exceptions import ClientConnectorError
import logging
from uuid import uuid4
import pandas as pd
from io import BytesIO
import hashlib


def _hash_from_dict(o):
    """Returns sha1 for dict object."""
    sha1 = hashlib.sha1()

    def _update(v):
        if isinstance(v, str):
            sha1.update(v.encode())
        elif isinstance(v, (int, float)):
            _update(str(v))
        elif isinstance(v, (tuple, list)):
            for e in v:
                _update(e)
        elif isinstance(v, dict):
            keys = v.keys()
            for k in sorted(keys):
                _update(k)
                _update(v[k])

    _update(o)
    return sha1.hexdigest()


def try_or(fn, default):
    # Universal crutch :)
    try:
        return fn()
    except:
        return default

  
async def data_to_xlsx(data, header=True, columns=None):
    """Returns Excel file in memory from raw data."""
    file = BytesIO()
    df = pd.DataFrame(data)
    df.to_excel(file, index=False, header=header, columns=columns)
    del df
    file.seek(0)
    return file


class JSONEncoder(json.JSONEncoder):
    """JSONEncoder."""

    def default(self, o):
        """Json defaults."""
        if isinstance(o, ObjectId):
            return str(o)
        elif isinstance(o, datetime.datetime):
            return o.timestamp()
        return json.JSONEncoder.default(self, o)


def json_response(fn):
    async def decorator(*args, **kwargs):
        headers = {'Access-Control-Allow-Origin': '*'}
        result = None
        try:
            result = await fn(*args, **kwargs)
        except ClientConnectorError as e:
            result = {'error': e}
            logging.error(e)
        responce = json.dumps(result, cls=JSONEncoder)
        return web.json_response(body=responce, headers=headers)
    return decorator


def text_response(fn):
    async def decorator(*args, **kwargs):
        result = await fn(*args, **kwargs)
        headers = {'Access-Control-Allow-Origin': '*'}
        return web.Response(text=str(result), headers=headers)
    return decorator


def xlsx_response(filename='report.xlsx', header=True, columns=None):
    def _response(fn):
        async def decorator(*args, **kwargs):
            result = await fn(*args, **kwargs)
            xlsx_file = await data_to_xlsx(result, header, columns)
            xlsx_file.name = str(uuid4()) + '.xlsx'
            headers = {
                'Access-Control-Allow-Origin': '*',
                'Content-Type': 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
                'Content-Disposition': 'attachment; filename=' + filename,
            }
            resp = web.StreamResponse(headers=headers)
            resp.content_length = xlsx_file.getbuffer().nbytes
            resp.content_type = 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet'
            await resp.prepare(*args)
            await resp.write(xlsx_file.getvalue())
            return resp
        return decorator
    return _response