API

HttpUser

Bases: User

Source code in src/aiolocust/users/http.py

class HttpUser(User):
    session_kwargs: dict[str, Any] = {}
    """
    Extra arguments to pass to aiohttp.ClientSession, e.g.
    ```
    class TimeoutUser(HttpUser):
        session_kwargs = {
            "timeout": aiohttp.ClientTimeout(0.0001),
            "skip_auto_headers": {"User-Agent"},
        }
        ...
    ```
    """

    def __init__(self, runner: Runner | None = None, base_url=None):
        super().__init__(runner)
        self.base_url = base_url
        self.client: LocustClientSession  # type: ignore[assignment] # always set in cm

    @asynccontextmanager
    async def cm(self):
        async with LocustClientSession(self.runner, self.base_url, **self.session_kwargs) as self.client:
            yield

session_kwargs = {} class-attribute instance-attribute

Extra arguments to pass to aiohttp.ClientSession, e.g.

class TimeoutUser(HttpUser):
    session_kwargs = {
        "timeout": aiohttp.ClientTimeout(0.0001),
        "skip_auto_headers": {"User-Agent"},
    }
    ...

User

Bases: ABC

Source code in src/aiolocust/__init__.py

class User(ABC):
    def __init__(self, runner: Runner | None = None, **kwargs):
        self.runner: Runner = runner  # pyright: ignore[reportAttributeAccessIssue] # always set outside of unit testing
        self.running = True

    @abstractmethod
    async def run(self): ...

    @asynccontextmanager
    async def cm(self):
        """Override this method if you need an async context manager around the run method"""
        yield

cm() async

Override this method if you need an async context manager around the run method

Source code in src/aiolocust/__init__.py

@asynccontextmanager
async def cm(self):
    """Override this method if you need an async context manager around the run method"""
    yield

datatypes

SafeCounter

A thread-safe counter.

Source code in src/aiolocust/datatypes.py

class SafeCounter:
    """A thread-safe counter."""

    def __init__(self, limit: int | None = None):
        self.value = 0
        self.limit = limit if limit is not None else sys.maxsize
        self.lock = threading.Lock()

    def increment(self) -> bool:
        with self.lock:
            if self.value < self.limit:
                self.value += 1
                return False
            return True

main

is_user_class(item)

Check if a variable is a runnable (non-abstract) User class

Source code in src/aiolocust/main.py

def is_user_class(item) -> bool:
    """
    Check if a variable is a runnable (non-abstract) User class
    """
    return bool(inspect.isclass(item)) and issubclass(item, User) and not inspect.isabstract(item)

users

http

HttpUser

Bases: User

Source code in src/aiolocust/users/http.py

class HttpUser(User):
    session_kwargs: dict[str, Any] = {}
    """
    Extra arguments to pass to aiohttp.ClientSession, e.g.
    ```
    class TimeoutUser(HttpUser):
        session_kwargs = {
            "timeout": aiohttp.ClientTimeout(0.0001),
            "skip_auto_headers": {"User-Agent"},
        }
        ...
    ```
    """

    def __init__(self, runner: Runner | None = None, base_url=None):
        super().__init__(runner)
        self.base_url = base_url
        self.client: LocustClientSession  # type: ignore[assignment] # always set in cm

    @asynccontextmanager
    async def cm(self):
        async with LocustClientSession(self.runner, self.base_url, **self.session_kwargs) as self.client:
            yield

session_kwargs = {} class-attribute instance-attribute

Extra arguments to pass to aiohttp.ClientSession, e.g.

class TimeoutUser(HttpUser):
    session_kwargs = {
        "timeout": aiohttp.ClientTimeout(0.0001),
        "skip_auto_headers": {"User-Agent"},
    }
    ...

request_hook(span, params)

Request hook for renaming spans based on name parameter (passed via context vars)

Typical usage:

AioHttpClientInstrumentor().instrument(request_hook=aiolocust.users.http.request_hook)

Source code in src/aiolocust/users/http.py

def request_hook(span: Span, params: aiohttp.TraceRequestStartParams):  # noqa: ARG001
    """
    Request hook for renaming spans based on name parameter (passed via context vars)

    Typical usage:

    ```
    AioHttpClientInstrumentor().instrument(request_hook=aiolocust.users.http.request_hook)
    ```
    """
    if custom_name := context.get_value(SPAN_NAME_KEY):
        span.update_name(str(custom_name))

pw

LocustPage

A wrapper for the Playwright Page object to automatically generate OTel spans.

Source code in src/aiolocust/users/pw.py

class LocustPage:
    """A wrapper for the Playwright Page object to automatically generate OTel spans."""

    def __init__(self, page: Page):
        self._page = page

    async def goto(self, url: str, **kwargs):
        with tracer.start_as_current_span("playwright.goto") as span:
            span.set_attribute("browser.url", url)
            try:
                result = await self._page.goto(url, **kwargs)
                stats.request(Request(url, 1, 1, None))
            except Exception as e:
                span.record_exception(e)
                stats.request(Request(url, 1, 1, e))
                raise
            return result

    async def click(self, selector: str, **kwargs):
        with tracer.start_as_current_span("playwright.click") as span:
            span.set_attribute("browser.selector", selector)
            try:
                result = await self._page.click(selector, **kwargs)
                stats.request(Request(selector, 1, 1, None))
            except Exception as e:
                span.record_exception(e)
                stats.request(Request(selector, 1, 1, e))
                raise
            return result