Create locust tasks dynamically?

Question

Basically I need to spawn 30 users and have 50 different tasks for them I need them to run in parallel. So I am trying to generate 50 tasks like following:

class UserSimulation(HttpUser):
    host = os.environ['BASE_URL']

    # time in seconds the simulated user will wait before proceeding to the next task
    wait_time = between(1, 2)

    for item_id in range(1, 51):

        @task(1)
        def view_items_with_different_item_ids(self, item_id=item_id):
            self.client.get(
                url=f"/my-url/item24_00{item_id}",
                verify=False,
                auth=(os.environ['USERNAME'], os.environ['PASSWORD'])

For obvious reasons this approach doesn't let me create 50 tasks dynamically, as only the last one gets saved. Any ideas of a work-around to it?

Answer 1

To do it the way you're trying, try creating different functions programmatically. I don't know how to use decorators with those, but if nothing else you can add the functions to the Locust task list when you create them. Just create a task list tasks = [] then tasks.append(view_items_with_different_item_ids_1) when the function is created.

However, I'm not positive if this is necessary, based on your description of what you want. If you only need 30 users to go make that call 50 times, you only need the one task and you can loop through the call in there.

class UserSimulation(HttpUser):
    host = os.environ['BASE_URL']

    # time in seconds the simulated user will wait before proceeding to the next task
    wait_time = between(1, 2)

    @task(1)
    def view_items_with_different_item_ids(self):
        for item_id in range(1, 51):
            self.client.get(
                url=f"/my-url/item24_00{item_id}",
                verify=False,
                auth=(os.environ['USERNAME'], os.environ['PASSWORD'])

If you need a random number instead of a sequential one but need to make sure each one is called once:

    import random

    item_ids = list(range(1,51))
    random.shuffle(item_ids)
    @task(1)
    def view_items_with_different_item_ids(self):
        for item_id in item_ids:
            self.client.get(
                url=f"/my-url/item24_00{item_id}",
                verify=False,
                auth=(os.environ['USERNAME'], os.environ['PASSWORD'])

If you just want to pull a random number all the time and don't care about repeats:

    import random

    item_ids = list(range(1,51))
    @task(1)
    def view_items_with_different_item_ids(self):
        random_id = random.choice(item_ids)
        self.client.get(
            url=f"/my-url/item24_00{random_id}",
            verify=False,
            auth=(os.environ['USERNAME'], os.environ['PASSWORD'])