Using Locust as a library

It is possible to start a load test from your own Python code, instead of running Locust using the locust command.

Start by creating an Environment instance:

from locust.env import Environment

env = Environment(user_classes=[MyTestUser])

The Environment instance’s create_local_runner, create_master_runner or create_worker_runner can then be used to start a Runner instance, which can be used to start a load test:

env.create_local_runner()
env.runner.start(5000, spawn_rate=20)
env.runner.greenlet.join()

It is also possible to bypass the dispatch and distribution logic, and manually control the spawned users:

new_users = env.runner.spawn_users({MyUserClass.__name__: 2})
new_users[1].my_custom_token = "custom-token-2"
new_users[0].my_custom_token = "custom-token-1"

The above example only works on standalone mode and is an experimental feature, meaning that it could be removed in future versions. But it’s useful if you want to have fine-grained control on the spawned users.

Note

Do not attempt to create a master runner and worker(s) in the same Python process. It does not work, and even if it did, it would not give you better performance than running a single LocalRunner. Every worker must run in its own process, there is no way around that.

We could also use the Environment instance’s create_web_ui method to start a Web UI that can be used to view the stats, and to control the runner (e.g. start and stop load tests):

env.create_local_runner()
env.create_web_ui()
env.web_ui.greenlet.join()

Full example

import gevent
from locust import HttpUser, task, between
from locust.env import Environment
from locust.stats import stats_printer, stats_history
from locust.log import setup_logging

setup_logging("INFO", None)


class User(HttpUser):
    wait_time = between(1, 3)
    host = "https://docs.locust.io"

    @task
    def my_task(self):
        self.client.get("/")

    @task
    def task_404(self):
        self.client.get("/non-existing-path")


# setup Environment and Runner
env = Environment(user_classes=[User])
env.create_local_runner()

# start a WebUI instance
env.create_web_ui("127.0.0.1", 8089)

# start a greenlet that periodically outputs the current stats
gevent.spawn(stats_printer(env.stats))

# start a greenlet that save current stats to history
gevent.spawn(stats_history, env.runner)

# start the test
env.runner.start(1, spawn_rate=10)

# in 60 seconds stop the runner
gevent.spawn_later(60, lambda: env.runner.quit())

# wait for the greenlets
env.runner.greenlet.join()

# stop the web server for good measures
env.web_ui.stop()