Advertisement
Guest User

Untitled

a guest
Apr 25th, 2019
91
0
Never
Not a member of Pastebin yet? Sign Up, it unlocks many cool features!
Python 1.65 KB | None | 0 0
  1. import traceback
  2. import asyncio
  3.  
  4. from aiohttp.web import (Request, Response,
  5.                          HTTPUnauthorized, HTTPUnprocessableEntity,)
  6. from aiohttp import web
  7. from aiohttp.web import middleware
  8.  
  9. from cerberus import Validator
  10.  
  11.  
  12. @middleware
  13. async def response_middleware(request: Request, handler) -> Response:
  14.     message = dict()
  15.     try:
  16.         response = await handler(request)
  17.  
  18.         message['success'] = True
  19.         message['result'] = response
  20.     except web.HTTPException as error:
  21.         message['success'] = False
  22.         message['error'] = error.text
  23.     except Exception:
  24.         traceback_message = traceback.format_exc()
  25.  
  26.         message['success'] = False
  27.         message['error'] = traceback_message
  28.     finally:
  29.         return web.json_response(message)
  30.  
  31.  
  32. @middleware
  33. async def auth_middleware(request: Request, handler) -> str:
  34.     if request.headers.get('auth') is None:
  35.         raise HTTPUnauthorized()
  36.  
  37.     return await handler(request)
  38.  
  39.  
  40. async def handler(request: Request) -> str:
  41.     args_schema: dict = {
  42.         'foo': {'type': 'integer', 'coerce': int, 'required': True, },
  43.         'bar': {'type': 'float', 'coerce': float, 'required': False,
  44.                 'default': 0.0, },
  45.     }
  46.     validator = Validator(args_schema)
  47.     if not validator.validate(dict(request.query)):
  48.         raise HTTPUnprocessableEntity(text=str(validator.errors))
  49.  
  50.     await asyncio.sleep(1.0)
  51.     return 'bar'
  52.  
  53.  
  54. def main() -> None:
  55.     app = web.Application(middlewares=[response_middleware, auth_middleware])
  56.     app.add_routes([web.get('/', handler)])
  57.     web.run_app(app)
  58.  
  59.  
  60. if __name__ == '__main__':
  61.     main()
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement