How to execute list of functions and also pass data to appropriate functions being called with asyncio

Question

I previously was using requests, but I have since moved onto aiohttp + asyncio to run accounts in parallel, however I am having trouble putting the logic together in my head.

class Faked(object):
    def __init__(self):
        self.database = sqlite3.connect('credentials.db')

    async def query_login(self, email):
        print(email)
        cur = self.database.cursor()
        sql_q = """SELECT * from user WHERE email='{0}'""".format(email)
        users = cur.execute(sql_q)
        row = users.fetchone()
        if row is None:
            raise errors.ToineyError('No user was found with email: ' + email + ' in database!')

        self.logger().debug("Logging into account '{0}'!".format(row[0]))
        call_func = await self._api.login(data={'email': row[0],
                                                'password': row[1],
                                                'deviceId': row[2],
                                                'aaid': row[3]})
        return await call_func

    async def send_friend_request(self, uid):
        return await self._api.send_friend_request(uid)


def main(funcs, data=None):
    """
   todo: fill
  :rtype: object
  """
    tasks = []
    if isinstance(funcs, list):
        for func in funcs:
            tasks.append(func)
    else:
        tasks.append(funcs)
    print(tasks)
    loop = asyncio.get_event_loop()
    results = loop.run_until_complete(asyncio.gather(*tasks))
    for result in results:
        print(result)
    return results


if __name__ == '__main__':  # for testing purposes mostly
    emails = ['email@hotmail.com', 'email@outlook.com', 'email@gmail.com']

I essentially just want to know how to queue multiple functions, in this instance query_login and send_friend_request, whilst also passing the correct data to said functions, assuming I'd be running three accounts on say a social media app simultaneously, it really boggles my mind, though I have a tendency of overcomplicating things, any help would be greatly appreciated.

Answer 1

Alright so this is really cool, let me show you something:

loop = asyncio.get_event_loop()
api = MyAPIToSomeCoolChatProgram()

def my_done_callback(fut):
    exc = fut.exception()
    if exc:
        print(fut.my_custom_attribute, 'raised an exception!')
        import traceback
        traceback.print_exc(exc) # dumps a "Traceback (most recent call last):" message to stderr
    print(fut.my_custom_attribute, 'completed, returned', repr(fut.result()))

fut1 = asyncio.ensure_future(api.send_friend_request(my_best_friend))
fut1.my_custom_attribute = 'fut1 (add a friend)'
fut1.add_done_callback(my_done_callback)

fut2 = asyncio.ensure_future(api.post_text_message('Hello everybody!'))
fut2.my_custom_attribute = 'fut2 (send a message)'
fut2.add_done_callback(my_done_callback)

print('Done creating the futures')
loop.run_forever()

Output:

  Done creating the futures
  fut1 (add a friend request) completed, returned '200 OK'
  fut2 (send a message) completed, returned '200 OK'

Note that they may appear in any order. You can call coroutines from non asynchronous code by wrapping the coroutine (the return value from the coroutine function) in a future (or more accurately a Task which is a subclass of Future). This coroutine will now run in the background. You can add a callback to the future which will be called when it finishes, passed one argument: the future object itself. Look up futures in the asyncio documentation if you'd like to learn more about them (also check out Coroutines and Tasks).

Anyway, those callbacks can do anything you want, including starting other tasks.

def when_done_logging_in(self, fut):
    self.login_info = fut.result() # note: calling fut.result() if the login coroutine raised an exception will reraise the exception here.
    next_fut = asyncio.ensure_future(self.send_friend_request(fut.friend_request_to_send))
    # do something with next_fut here (or don't if you don't care about the result)

def login_and_send_friend_request(self, email, friend):
    fut = asyncio.ensure_future(self.query_login(email))
    fut.friend_request_to_send = friend
    fut.add_done_callback(self.when_done_logging_in)

Of course you could also do that with:

 async def login_and_send_friend_request(self, email, friend):
    self.login_info = await self.query_login(email)
    await self.send_friend_request(friend)

which would be better because any exceptions are actually handled properly instead of simply being ignored. You could also do this, if you know the email in advance (which you may not):

def __init__(self, whatever_args_you_might_have_here, email):
    ...
    self.login_info = None
    self.email = email

async def send_friend_request(self, uid):
    if self.login_info is None:
        await self.query_login(self.email) # if you end up doing this you should probably make this not take a parameter and just use self.email instead
    do_send_friend_request_stuff()

Of course you might not know the email until after the object is created, in which case you could either initialize it to None until some login function is called, or use one of the first two ways.

If you wanted to execute a list of functions in sequence, you could do the following:

    def execute_coros_in_sequence(list_of_coros):
        fut=asyncio.ensure_future(list_of_coros[0])
        if len(list_of_coros) > 1:
            # there is probably a better way to do this
            fut.remaining_coros=list_of_coros[1:]
            fut.add_done_callback(lambda fut: execute_coros_in_sequence(fut.remaining_coros))

but probably a better way to do it would be just to make an async def function call them all, because that way you get exception handling etc. without a lot of overcomplication. A better way to do it, if you wanted it as a future (which you can also store as an attribute of an object and query to see if it's done yet), would be this:

class API:
     async def login(self):
         pass
     async def logout(self):
         pass
     async def do_fun_stuff(self):
         pass


async def test_api(api):
    api.login()
    api.do_fun_stuff()
    api.logout()

fut=asyncio.create_task(test_part_of_api(API()))

(By the way, asyncio.ensure_future() first checks to see if its argument is already a future, and if not, calls asyncio.create_task().)

but the future api is really cool and I wanted to show it to you. There are uses for it, and I can pretty much guarantee you're going to need both of these approaches in order to do anything complex.

Sorry for the wall of text disorganized answer. I'm a bit new here. I just think asyncio is really cool.

Answer 2

Python is designed to make this fairly easy with the unpack operator * or using lambda. There are several good answers in this thread that have what you need:

Passing functions with arguments to another function in Python?

Let's walk through it.

callstack = [] # initialize a list to serve as our stack.
     # See also collections.deque for a queue.

Then we can define our function:

def somefunc(a, b, c): 
    do stuff...

Then add a call to the stack with the arguments as a list.

args = [a, b, c]
callstack.append((somefunc, args)) # append a tuple with the function
            # and its arguments list.

# calls the next item in the callstack
def call_next(callstack):
    func, args = callstack.pop() # unpack our tuple
    func(*args) # calls the func with the args unpacked

The * operator unpacks your list and provides them as arguments in order. You can also unpack keyword arguments with the double-star operator (**).

def call_next(callstack):
    func, args, kwargs = callstack.pop() # unpack our tuple
    func(*args, **kwargs) # calls the func with both args and kwargs unpacked.

An alternate way is to just make a lambda.

def add(a, b):
    return a + b

callstack = []

callstack.append(lambda: add(1, 2))
callstack.pop()() # pops the lambda function, then calls the lambda function, 
                  # which just calls the function as you specified it.

Voila! All credit to the authors in the other thread. There's a gotcha here: If you're passing an object as an argument, it will passed as a reference. Be careful because you can modify the object before it gets called in your stack.

def add(a, b, c):
    return a + b + c

badlist = [1,2,3]
callstack.append((somefunc, badlist))
badlist = [2, 4, 6]
callstack.append((somefunc, badlist))

while len(callstack) > 0:
    print(call_next(callstack))

# Prints:
12
12

You can get around this in the *args version with:

# make a shallow copy and pass that to the stack instead.
callstack.append((somefunc, list(badlist))) 

In lambda functions, the entire thing is evaluated at call time so even things that normally wouldn't be a reference behave like references. The above trick won't work, so do any copying as needed before creating your lambda.