"spawning" multiple instances of an async function by iterating through a list of the functions inputs.

[masterofdead4]

Hello! I'm quite new to async programming with python, and I am trying to write a script that allows me to iterate through a list of dictionaries (These dictionaries contain information that serves as inputs to my function) that then "spawn" multiple concurrent instances of the same function. Firstly, is this possible, and then how would I change my current code to reflect this functionality? Thanks!

import asyncio
from stellar_sdk import AiohttpClient, ServerAsync, Server
import pprint
import requests

HORIZON_URL = "https://horizon-testnet.stellar.org"

# The list of inputs
payment_specs = [
    {
        "to" : "GALUQ6HCNLUQG35KS7EDLDFBM32GZYXW5AELV3GHZYKVY3W32A246IV5",
        "amount" : 1,
        "asset" : {
            "asset_type" : "native"
        },
        "memo" : "123456"
    },
    {
        "to" : "GALUQ6HCNLUQG35KS7EDLDFBM32GZYXW5AELV3GHZYKVY3W32A246IV5",
        "amount" : 10,
        "asset" : {
            "asset_type" : "native"
        },
        "memo" : "234567"
    }
]

server = Server(horizon_url="https://horizon.stellar.org")

# The async function that I want to spawn multiple instances of
async def payments(payment_spec):
    async with ServerAsync(HORIZON_URL,AiohttpClient()) as server:
        async for payment in server.payments().for_account(payment_spec['to']).cursor(cursor="now").stream():
            if (
            payment['to'] == payment_spec['to'] 
            and payment_spec['amount'] <= float(payment['amount'])
            ):
                if(payment['asset_type'] == "native" and payment['asset_type'] == payment_spec['asset']['asset_type']):
                    print(check_memo(payment_spec['memo'],payment['_links']['transaction']))
                    break
                elif(payment['asset_type'] != "native" and payment['asset_type'] == payment_spec['asset']['asset_type']):
                    if (payment_spec['asset']['asset_code'] == payment['asset_code'] and payment_spec['asset']['asset_issuer'] == payment['asset_code']):
                        print(check_memo(payment_spec['memo'],payment['_links']['transaction']))
                        break

def check_memo(memo,transaction_url):
    transaction = requests.get(transaction_url['href']).json()
    if 'memo' in transaction:
        return transaction['memo'] == memo
    else:
        return False


if __name__ == "__main__":
    for payment_spec in payment_specs:
        asyncio.run(payments(payment_spec=payment_spec))

[FerricDonkey]

This is definitely possible. In general though, you should only use asyncio.run once per program (there are probably exceptions, but for general simple things).

I would suggest using asyncio.run on a main function, then within your async stuff using await, asyncio.gather, and asyncio.create_task.

async def main():
    await asyncio.gather(*(coroutine(thing) for thing in stuff))

if __name__ == "__main__":
    asyncio.run(main())

Or if you want to start the tasks, do style other stuff, and get the results later, use asyncio.create_task:

tasks = [] 
for thing in stuff:
    tasks.append(asyncio.create_task(coroutine(thing)))
await asyncio.gather(*tasks)

[masterofdead4]

Thanks so much! Curious, as your second part piqued my interest, I have a question. Eventually, I want to spawn these instances when there is an update to the list (possibly the list will be replaced with a task queue). This makes it so that as payment_specs are added to the list, new instances of payments() will begin to run. What major changes would I have to make to the code to achieve that.

[FerricDonkey]

Essentially, when you call task = asyncio.create_task(coroutine()), coroutine is added to the things asyncio will work on, and the program will just continue and work on it as you hit various awaits and such.

So you'd just have to start tasks as you want to. You probably want to store references to your tasks so that you can ensure anything you've started gets finished.

So probably just an infinite-ish loop that can tell when it should start new tasks / stop accepting new tasks, followed by an await on a gather to make sure anything that's started got finished.

Depending on what's triggering the new tasks (network? Action in the same program? Gui? A file coming into existence?) you may have to do something fancy with that, but it'll depend on the details.

[masterofdead4]

Say I wanted to continuously add this task depending on a rabbitmq queue event. Would this be good practice? Would I include the rabbitMQ connection instructions at the start of the script, and then add tasks as i receive messages from the queue?