Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Proxy server by using aiohttp.

Async Proxy Server

Installation

pip install -r requirements.txt

Run

Run the backend server

python runbackend.py

Run the proxy server

python runproxy.py

Testing

Run the Apache Bench. Then you can see that the server will handle requestsa asynchronously.

ab -n 100 -c 100 http://127.0.0.1:8080/
import time
from wsgiref.util import setup_testing_defaults
from wsgiref.simple_server import make_server
def ok_app(environ, start_response):
setup_testing_defaults(environ)
status = '200 OK'
headers = [('Content-type', 'application/json; charset=utf-8')]
start_response(status, headers)
ret = [b'{"message": "OK"}\n']
time.sleep(5)
return ret
if __name__ == "__main__":
httpd = make_server('', 5000, ok_app)
print("Serving on 5000...")
httpd.serve_forever()
import logging
from urllib.parse import urljoin
import asyncio
import aiohttp
from aiohttp import web
TARGET_SERVER_BASE_URL = 'http://127.0.0.1:5000'
logger = logging.getLogger("runproxy")
@asyncio.coroutine
def proxy(request):
target_url = urljoin(TARGET_SERVER_BASE_URL, request.match_info['path'])
logger.info("Requesting to: %s", target_url)
res = yield from aiohttp.request('get', target_url)
raw = yield from res.text()
logger.info("Got a response from: %s", target_url)
return web.Response(text=raw, status=res.status, headers={"Content-Type": "application/json"})
if __name__ == "__main__":
logging.root.setLevel(logging.INFO)
logging.root.addHandler(logging.StreamHandler())
app = web.Application()
app.router.add_route('GET', '/{path:\w*}', proxy)
loop = asyncio.get_event_loop()
f = loop.create_server(app.make_handler(), '0.0.0.0', 8080)
srv = loop.run_until_complete(f)
print('serving on', srv.sockets[0].getsockname())
try:
loop.run_forever()
except KeyboardInterrupt:
pass
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment