Skip to content

Instantly share code, notes, and snippets.

@jbn
Created April 7, 2016 12:24
Show Gist options
  • Save jbn/fc90e3ddbc5c60c698d07b3df30004c8 to your computer and use it in GitHub Desktop.
Save jbn/fc90e3ddbc5c60c698d07b3df30004c8 to your computer and use it in GitHub Desktop.
An example showing how to stream HTML in a aiohttp server.
import asyncio
from aiohttp import web
import subprocess
async def uptime_handler(request):
# http://HOST:PORT/?interval=90
interval = int(request.GET.get('interval', 1))
# Without the Content-Type, most (all?) browsers will not render
# partially downloaded content. Note, the response type is
# StreamResponse not Response.
resp = web.StreamResponse(status=200,
reason='OK',
headers={'Content-Type': 'text/html'})
# The StreamResponse is a FSM. Enter it with a call to prepare.
await resp.prepare(request)
while True:
try:
# Technically, subprocess blocks, so this is a dumb call
# to put in an async example. But, it's a tiny block and
# still mocks instantaneous for this example.
resp.write(b"<strong>")
resp.write(subprocess.check_output('uptime'))
resp.write(b"</strong><br>\n")
# Yield to the scheduler so other processes do stuff.
await resp.drain()
# This also yields to the scheduler, but your server
# probably won't do something like this.
await asyncio.sleep(interval)
except Exception as e:
# So you can observe on disconnects and such.
print(repr(e))
raise
return resp
async def build_server(loop, address, port):
# For most applications -- those with one event loop --
# you don't need to pass around a loop object. At anytime,
# you can retrieve it with a call to asyncio.get_event_loop().
# Internally, aiohttp uses this pattern a lot. But, sometimes
# "explicit is better than implicit." (At other times, it's
# noise.)
app = web.Application(loop=loop)
app.router.add_route('GET', "/uptime", uptime_handler)
return await loop.create_server(app.make_handler(), address, port)
if __name__ == '__main__':
loop = asyncio.get_event_loop()
loop.run_until_complete(build_server(loop, 'localhost', 9999))
print("Server ready!")
try:
loop.run_forever()
except KeyboardInterrupt:
print("Shutting Down!")
loop.close()
@userappgate
Copy link

I made a couple of changes, so it actually works with aiohttp 3

import asyncio
from aiohttp import web
import subprocess


async def uptime_handler(request):
    # http://HOST:PORT/?interval=90
    interval = 1

    # Without the Content-Type, most (all?) browsers will not render
    # partially downloaded content. Note, the response type is
    # StreamResponse not Response.
    resp = web.StreamResponse(status=200,
                              reason='OK',
                              headers={'Content-Type': 'text/html'})

    # The StreamResponse is a FSM. Enter it with a call to prepare.
    await resp.prepare(request)



    while True:
        try:
            # Technically, subprocess blocks, so this is a dumb call
            # to put in an async example. But, it's a tiny block and
            # still mocks instantaneous for this example.
            await resp.write(b"<strong>")
            await resp.write(subprocess.check_output('uptime'))
            await resp.write(b"</strong><br>\n")

            # Yield to the scheduler so other processes do stuff.
            await resp.drain()

            # This also yields to the scheduler, but your server
            # probably won't do something like this.
            await asyncio.sleep(interval)
        except Exception as e:
            # So you can observe on disconnects and such.
            print(repr(e))
            raise

    return resp


async def build_server(loop, address, port):
    # For most applications -- those with one event loop --
    # you don't need to pass around a loop object. At anytime,
    # you can retrieve it with a call to asyncio.get_event_loop().
    # Internally, aiohttp uses this pattern a lot. But, sometimes
    # "explicit is better than implicit." (At other times, it's
    # noise.)
    app = web.Application(loop=loop)
    app.router.add_route('GET', "/uptime", uptime_handler)

    return await loop.create_server(app.make_handler(), address, port)


if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(build_server(loop, 'localhost', 9999))
    print("Server ready!")

    try:
        loop.run_forever()
    except KeyboardInterrupt:
        print("Shutting Down!")
        loop.close()

@MagicMagnate
Copy link

For any deprecation warning (ver. 3.4.4) I removed await resp.drain() and in build_server part changed to:

   app = web.Application()
   app.router.add_route('GET', "/uptime", uptime_handler)
   runner = web.AppRunner(app)
   await runner.setup()

   site = web.TCPSite(runner, address, port)
   await site.start()

@nuzhatgawhar
Copy link

I made a couple of changes, so it actually works with aiohttp 3

import asyncio
from aiohttp import web
import subprocess


async def uptime_handler(request):
    # http://HOST:PORT/?interval=90
    interval = 1

    # Without the Content-Type, most (all?) browsers will not render
    # partially downloaded content. Note, the response type is
    # StreamResponse not Response.
    resp = web.StreamResponse(status=200,
                              reason='OK',
                              headers={'Content-Type': 'text/html'})

    # The StreamResponse is a FSM. Enter it with a call to prepare.
    await resp.prepare(request)



    while True:
        try:
            # Technically, subprocess blocks, so this is a dumb call
            # to put in an async example. But, it's a tiny block and
            # still mocks instantaneous for this example.
            await resp.write(b"<strong>")
            await resp.write(subprocess.check_output('uptime'))
            await resp.write(b"</strong><br>\n")

            # Yield to the scheduler so other processes do stuff.
            await resp.drain()

            # This also yields to the scheduler, but your server
            # probably won't do something like this.
            await asyncio.sleep(interval)
        except Exception as e:
            # So you can observe on disconnects and such.
            print(repr(e))
            raise

    return resp


async def build_server(loop, address, port):
    # For most applications -- those with one event loop --
    # you don't need to pass around a loop object. At anytime,
    # you can retrieve it with a call to asyncio.get_event_loop().
    # Internally, aiohttp uses this pattern a lot. But, sometimes
    # "explicit is better than implicit." (At other times, it's
    # noise.)
    app = web.Application(loop=loop)
    app.router.add_route('GET', "/uptime", uptime_handler)

    return await loop.create_server(app.make_handler(), address, port)


if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(build_server(loop, 'localhost', 9999))
    print("Server ready!")

    try:
        loop.run_forever()
    except KeyboardInterrupt:
        print("Shutting Down!")
        loop.close()

Hi if I want to use it for videos how can I do that?

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment