# Python task runner with asyncio

I'm trying to a write super simple task runner with asyncio.

Basically, my goal is to run 100 requests to google as tasks, whilst also being be able to handle tasks being added at a later time.

from abc import ABC, abstractmethod
import asyncio
import socket
import time

import aiohttp
import requests

class MyScheduler:
def __init__(self, wait=2):
self.work = []
self.wait = wait

def set_initial_callback(self, callback, **kwargs):
self.initial_callback = callback
self.initial_callback_args = kwargs

async def _run(self):
while self.work:
if len(self.work) == 0:
await asyncio.sleep(self.wait)

async def set_things_up(self, callback, **kwargs):
await self._run()

def go(self):
asyncio.run(self.set_things_up(self.initial_callback, **self.initial_callback_args))

if n == 100:
return None

async with aiohttp.ClientSession() as session:
async with session.get('http://h...content-available-to-author-only...n.org/get') as resp:
print(resp.status)

t = time.time()
s = MyScheduler(wait=1)
s.go()
print(time.time() - t)



I benchmarked this against sequentally running requests, and I did see a massive speed up. It's still super rough, but I'd love some pointers on how I could improve my code in terms of readability/exploiting async stuff better.

I actually just started learning asyncio a couple days ago, so I won't be able to comment too deeply. I do see a few things though:

Disregarding asyncio for a sec, I think google could be set up better. You have the base case of the recursion as n == 100, and are incrementing n in each recursive call. To easily allow the caller to decide how many time to run, I'd reverse how n is being handled. I would decrement it each call, and set the base case as n <= 0. With how you have it now, say the caller wanted it to run 1000 times, they would need to call it as

google(-900)


which is a little wonky. I'd change the first bit to:

async def google(n):
if n <= 0:
return None

. . .


I'm not sure recursion is the cleanest tool for the job here. I'm also not sure entirely why you're using a job queue or why you're using a elaborate class here unless the goal is to be able to handle jobs being added at a later time.

If your goal is just to initiate many requests and wait on them at the same time, you could just gather them:

import aiohttp
import asyncio as a

# That is arguably beyond the responsibilities of a function intended to make requests
async with aiohttp.ClientSession() as session:
async with session.get('http://h...content-available-to-author-only...n.org/get') as resp:
print(resp.status)

async def start_requests(n_requests: int):
routines = [google() for _ in range(n_requests)]  # Create a list of reqeust-making coroutines
await a.gather(*routines)  # Unpack the routines into gather (since gather is var-arg)


Also, instead of doing timing using a single attempt and plain subtraction, it would be more accurate to use timeit:

from timeit import timeit

print("t:", timeit(lambda: a.run(start_requests(10)), number=20))  # number is the amount of tests to do


I'm assuming there's no issue using timeit for async code.

• thanks for the feedback! I should have added, yes the goal is to be able to handle jobs being added at a later time, as it offers more flexibility. :) Do you have any recommendations for how I could implement a graceful shutdown? – nz_21 Aug 29 '19 at 8:15