Backup-bdg/OpenHands

1import base642from typing import AsyncIterator, Callable3 4 5def offset_to_page_id(offset: int, has_next: bool) -> str | None:6    if not has_next:7        return None8    next_page_id = base64.b64encode(str(offset).encode()).decode()9    return next_page_id10 11 12def page_id_to_offset(page_id: str | None) -> int:13    if not page_id:14        return 015    offset = int(base64.b64decode(page_id).decode())16    return offset17 18 19async def iterate(fn: Callable, **kwargs) -> AsyncIterator:20    """Iterate over paged result sets. Assumes that the results sets contain an array of result objects, and a next_page_id"""21    kwargs = {**kwargs}22    kwargs['page_id'] = None23    while True:24        result_set = await fn(**kwargs)25        for result in result_set.results:26            yield result27        if result_set.next_page_id is None:28            return29        kwargs['page_id'] = result_set.next_page_id30