trading_agent.requests.user_request_store

Saves the desks' requests to the user, one JSON file each in the desk's requests folder.

Typical usage example:

store = UserRequestStore(layout) store.save(request) store.pending()

UserRequestError

Bases: ValueError

A request does not exist, or is no longer waiting for a response.

UserRequestStore

UserRequestStore(layout)

Saves, finds and lists requests, and marks lapsed ones as expired.

Creates the store.

Parameters:

Name	Type	Description	Default
layout	WorkspaceLayout	The workspace whose desks' requests folders hold the requests.	required

Source code in src/trading_agent/requests/user_request_store.py

def __init__(self, layout: workspace_layout.WorkspaceLayout):
    """Creates the store.

    Args:
        layout (workspace_layout.WorkspaceLayout): The workspace whose desks' requests folders hold the requests.
    """
    self._layout = layout

save

save(request)

Writes a request, replacing any earlier version of it.

Parameters:

Name	Type	Description	Default
request	UserRequest	The request.	required

Source code in src/trading_agent/requests/user_request_store.py

def save(self, request: user_request_module.UserRequest) -> None:
    """Writes a request, replacing any earlier version of it.

    Args:
        request (user_request_module.UserRequest): The request.
    """
    directory = self._layout.requests_directory(request.desk)
    directory.mkdir(parents=True, exist_ok=True)
    path = directory / f'{request.request_id}.json'
    path.write_text(
        json.dumps(request.to_json_dictionary(), indent=2),
        encoding='utf-8',
    )

all

all()

Every request from every desk, oldest first, with lapsed ones shown as expired.

A lapsed request is only shown as expired here; expire_lapsed saves the change, so that the desk is told exactly once.

Returns:

Type	Description
list[UserRequest]	The requests.

Source code in src/trading_agent/requests/user_request_store.py

def all(self) -> list[user_request_module.UserRequest]:
    """Every request from every desk, oldest first, with lapsed ones shown as expired.

    A lapsed request is only shown as expired here; expire_lapsed saves the change, so that the desk is told exactly once.

    Returns:
        list[user_request_module.UserRequest]: The requests.
    """
    now = datetime.datetime.now(datetime.UTC)
    requests = []
    for request in self._saved():
        if request.status == user_request_module.PENDING and request.is_expired(now):
            request = dataclasses.replace(
                request,
                status=user_request_module.EXPIRED,
            )
        requests.append(request)
    return requests

expire_lapsed

expire_lapsed()

Saves every lapsed pending request as expired.

Returns:

Type	Description
list[UserRequest]	The requests that have just expired.

Source code in src/trading_agent/requests/user_request_store.py

def expire_lapsed(self) -> list[user_request_module.UserRequest]:
    """Saves every lapsed pending request as expired.

    Returns:
        list[user_request_module.UserRequest]: The requests that have just expired.
    """
    now = datetime.datetime.now(datetime.UTC)
    expired = []
    for request in self._saved():
        if request.status == user_request_module.PENDING and request.is_expired(now):
            lapsed = dataclasses.replace(
                request,
                status=user_request_module.EXPIRED,
            )
            self.save(lapsed)
            expired.append(lapsed)
    return expired

pending

pending()

The requests still waiting for a response, oldest first.

Returns:

Type	Description
list[UserRequest]	The pending requests.

Source code in src/trading_agent/requests/user_request_store.py

def pending(self) -> list[user_request_module.UserRequest]:
    """The requests still waiting for a response, oldest first.

    Returns:
        list[user_request_module.UserRequest]: The pending requests.
    """
    waiting = []
    for request in self.all():
        if request.status == user_request_module.PENDING:
            waiting.append(request)
    return waiting

get

get(request_id)

Finds one request.

Parameters:

Name	Type	Description	Default
request_id	str	Its identifier.	required

Returns:

Type	Description
UserRequest	The request.

Raises:

Type	Description
UserRequestError	No request has that identifier.

Source code in src/trading_agent/requests/user_request_store.py

def get(self, request_id: str) -> user_request_module.UserRequest:
    """Finds one request.

    Args:
        request_id (str): Its identifier.

    Returns:
        user_request_module.UserRequest: The request.

    Raises:
        UserRequestError: No request has that identifier.
    """
    for request in self.all():
        if request.request_id == request_id:
            return request
    raise UserRequestError(f'No request with id {request_id!r}')

pending_request

pending_request(request_id)

Finds a request that is still waiting for a response.

Parameters:

Name	Type	Description	Default
request_id	str	Its identifier.	required

Returns:

Type	Description
UserRequest	The request.

Raises:

Type	Description
UserRequestError	It does not exist, or was already answered, dismissed or expired.

Source code in src/trading_agent/requests/user_request_store.py

def pending_request(self, request_id: str) -> user_request_module.UserRequest:
    """Finds a request that is still waiting for a response.

    Args:
        request_id (str): Its identifier.

    Returns:
        user_request_module.UserRequest: The request.

    Raises:
        UserRequestError: It does not exist, or was already answered, dismissed or expired.
    """
    request = self.get(request_id)
    if request.status != user_request_module.PENDING:
        raise UserRequestError(
            f'Request {request_id} is already {request.status}'
        )
    return request