API

HttpUser

Bases: User

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

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

    ssl_context: None | ssl.SSLContext = None
    """
    Used to create a custom TCPConnector for LocustClientSession,
    instead of having to pass ssl=ssl_context on each request.

    For example, to use OS-managed trust you could do something like this:

    import truststore
    from aiolocust import HttpUser

    class MyUser(HttpUser):
        ssl_context = truststore.SSLContext(protocol=ssl.PROTOCOL_TLS_CLIENT)
        ...
    """

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

    @asynccontextmanager
    async def cm(self):
        async with LocustClientSession(
            self.runner,
            self.base_url,
            connector=aiohttp.TCPConnector(ssl=self.ssl_context) if self.ssl_context else None,
            **self.session_kwargs,
        ) as self.client:
            yield

session_kwargs = {'timeout': aiohttp.ClientTimeout(60.0)} 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"},
    }
    ...

ssl_context = None class-attribute instance-attribute

Used to create a custom TCPConnector for LocustClientSession, instead of having to pass ssl=ssl_context on each request.

For example, to use OS-managed trust you could do something like this:

import truststore from aiolocust import HttpUser

class MyUser(HttpUser): ssl_context = truststore.SSLContext(protocol=ssl.PROTOCOL_TLS_CLIENT) ...

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
    def run(self) -> AbcCoroutine[Any, Any, None]: ...

    @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

users

http

HttpUser

Bases: User

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

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

    ssl_context: None | ssl.SSLContext = None
    """
    Used to create a custom TCPConnector for LocustClientSession,
    instead of having to pass ssl=ssl_context on each request.

    For example, to use OS-managed trust you could do something like this:

    import truststore
    from aiolocust import HttpUser

    class MyUser(HttpUser):
        ssl_context = truststore.SSLContext(protocol=ssl.PROTOCOL_TLS_CLIENT)
        ...
    """

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

    @asynccontextmanager
    async def cm(self):
        async with LocustClientSession(
            self.runner,
            self.base_url,
            connector=aiohttp.TCPConnector(ssl=self.ssl_context) if self.ssl_context else None,
            **self.session_kwargs,
        ) as self.client:
            yield

session_kwargs = {'timeout': aiohttp.ClientTimeout(60.0)} 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"},
    }
    ...

ssl_context = None class-attribute instance-attribute

Used to create a custom TCPConnector for LocustClientSession, instead of having to pass ssl=ssl_context on each request.

For example, to use OS-managed trust you could do something like this:

import truststore from aiolocust import HttpUser

class MyUser(HttpUser): ssl_context = truststore.SSLContext(protocol=ssl.PROTOCOL_TLS_CLIENT) ...

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)
                await events.request.fire(Request(url, 1, 1, None))
            except Exception as e:
                span.record_exception(e)
                await events.request.fire(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)
                await events.request.fire(Request(selector, 1, 1, None))
            except Exception as e:
                span.record_exception(e)
                await events.request.fire(Request(selector, 1, 1, e))
                raise
            return result