How to use an iterator/generator to paginate HTTP request responses

[chuckwondo]

How can I go about iterating through "pages" of HTTP request responses?

For example, I want to query an HTTP API that pages results, and I want to create an iterator (perhaps a generator function) to perform the pagination (i.e., make as many HTTP requests as required to get all of the "pages") so that the caller of the function doesn't have to "manually" perform pagination.

In other words, a user should be able to paginate through multiple HTTP requests, similar to the following:

class Env:
    # The `read_timeout` parameter of `aiohttp.ClientSession` is deprecated, but setting
    # its value to `aiohttp.helpers.sentinel` silences the `DeprecationWarning`.
    # See https://github.com/suned/pfun/issues/107
    http = HTTP(read_timeout=aiohttp.helpers.sentinel)

for page in paginate().run(Env()):
    # parse page (a Response instance), etc.

Ignoring details for the moment, I imagine a function similar to the following for fetching a page of results (in a Response, but ignoring parsing for the moment):

def get_page() -> Effect[HasHTTP, ClientError, Response]:
    return depend(HasHTTP).and_then(
        lambda env: env.http.make_request(
            method="get",
            url=...,
            headers=...,
            params=...,
        )
    )

In addition, the paginate function that the user would call might look something like the following (again, ignoring some details):

def paginate() -> Effect[HasHTTP, ClientError, Iterator[Response]]:
    def pages(env: HasHTTP) -> Iterator[Response]:
        response = get_page().run(env)
        yield response

        # For now, ignore how we halt iteration, and how we cause `get_page`
        # to get subsequent pages (rather than fetching the same page each time).
        while <some condition>:
            response = get_page().run(env)
            yield response

    return depend(HasHTTP).map(pages)

Unfortunately, attempting to iterate through multiple Responses with an implementation similar to the above results in the following error:

TypeError: object generator can't be used in 'await' expression

I realize this has to do with using aiohttp under the covers and making use of an event loop, but my limited understanding of asyncio and FP concepts makes it hard for me to determine how to refactor things to allow such iteration that keeps the client iteration logic relatively clean and simple.

I've also thought that perhaps the signature for paginate should "invert" the position of Iterator and Effect, like so:

def paginate() -> Iterator[Effect[HasHTTP, ClientError, Response]]:
    ...

Then, the caller might end up doing something like so, instead:

env = Env()

for page in paginate():
    response = page.run(env)
    # parse response, etc.

One of the details not shown is that in order to obtain the "next" page of results, I need to supply a "token" of sorts, and the "token" is provided in the response (as a header) from the "previous" page of results.

To be more concrete, and to allow experimentation, in particular, I want to page through results of queries made to this public URL: https://cmr.uat.earthdata.nasa.gov/search/collections.umm_json

If you open a browser to that URL, you will get a JSON response, which by default, returns the first page of 10 items. In order to obtain the next page of 10 items, the CMR-Search-After header in the response must be supplied as a header in the next request, which will then return a response with a new value for the header, to supply in the next response, and so on, until the last page is fetched, in which case the header will not be in the response (thus indicating the end of the pages of results).

For reference on this HTTP API, see Earthdata CMR Search API. In particular, see the Search After section.

How can I implement iteration to abstract away some of these details from the user, and allow the user to iterate through pages in a relatively clean and simple manner?

[suned]

In addition, the paginate function that the user would call might look something like the following (again, ignoring some details)

One issue with this implementation is that its not obviously stack safe (ie might lead to RecursionError). The issue is that you're calling run on an effect inside of a continuation function (the one passed to map). When calling run on the effect returned by map, run is called recursively in pages, and pfun cant trampoline this recursion for you. I can elaborate further if you want (might be a bit long winded), but the gist is: don't call run inside continuation functions.

Unfortunately, attempting to iterate through multiple Responses with an implementation similar to the above results in the following error

I think this might be a bug in pfun actually. The issue is that map currently uses asyncio.iscoroutine to test if the return value of the continuation function is a coroutine, in which case its awaited. Apparently asyncio.iscoroutine mistakes the generator returned by your map function for a coroutine. It might make more sense to use asyncio.iscoroutinefunction to test if the result of the continuation should be awaited 🤔

I've also thought that perhaps the signature for paginate should "invert" the position of Iterator and Effect

The norm in pure FP programs is to only run effects at the very edge of your program, which this approach would make difficult to do. As such I think the first approach would be more natural to most functional programmers.

In general, I think your pain here stems from trying to use imperative tools (a while loop) in functional style. I would use something like this as a starting point (not tested):

from pfun import http, success, Effect, operator as op
from typing import Callable, List, TypeVar

A = TypeVar('A')

HTTPEffect = Effect[http.HasHTTP, http.ClientError, A]

class Page:
    ...


def parse(response: http.Response) -> Page:
    ...

def has_next_page(page: Page) -> bool:
    ...

def get_next_page(p: Page) -> HTTPEffect[Page]:
    ...


def get_while_has_next_page(current_page: Page) -> HTTPEffect[List[Page]]:
        if has_next_page(current_page):
            return (get_next_page(current_page)
                    .and_then(get_while_has_next_page)
                    .map(op.add([current_page])))
        return success([current_page])
    


def paginate(url: str) -> HTTPEffect[List[Page]]:
    first_page = http.get(url).map(parse)
    return first_page.and_then(get_while_has_next_page)

There's probably a generalization of get_while_has_next_page that could solve this for any scenario where you want to iterate a function with the results of the previous iteration until some condition which could be a valuable addition to pfun itself. Maybe it would look something like:

def gather_until(p: Callable[[A], bool],
                 next_: Callable[[A], Effect[R, E, A]],
                 first: Effect[R, E, A]) -> Effect[R, E, List[A]]:
    def iterate(x: A) -> Effect[R, E, List[A]]:
        if p(x):
            return next_(x).and_then(iterate).map(lambda rs: [x] + rs)
        return success([x])
    return first.and_then(iterate)

(Discovered a bug in the mypy pfun.Intersection plugin while doing this, but nevermind 😅 )

Let me know if you come up with something! Also, a general pagination library for pfun.http would be super cool 😄

[chuckwondo]

Thanks for your detailed answer. There's plenty of code for me to study and learn from.

However, the biggest part of this (and why I included it in the title of this discussion) is the desire to use a generator. I'm specifically looking for a way to not read every page into a list, and then return the list to the caller.

Since the number of pages might be quite large, I want to avoid holding every page in memory at once. I want to use a generator so that I can process a page at a time, effectively processing a "stream" of pages in order to avoid the possibility of an out-of-memory error.

That's also why I'm using a loop. I need to repeatedly fetch a page, then yield it. I'm not aware of a way to implement a generator function without an explicit loop somewhere in the mix, but please correct me if I'm missing something. Of course, I could write a recursive generator to eliminate the explicit loop, but I then run the risk of a stack overflow.

Alternatively, I suppose we could drop the generator function and design another way to implement such "lazy" page fetching, so I'm certainly willing to explore other avenues. I'm not tied to using generators, but rather to the "laziness," to avoid potential memory issues. Using a generator function was simply my "go to" approach for laziness.

[suned]

You might be able to something like:

def get_while_has_next_page(current_page: Page) -> HTTPEffect[Iterator[Page]]:
    def recursive_yield(g: Iterator[Page]) -> Iterator[Page]:
        yield current_page
        yield from g

    if has_next_page(current_page):
        return (get_next_page(current_page)
                .and_then(get_while_has_next_page)
                .map(recursive_yield))
    return success((p for p in [current_page]))

But that's getting to be a bit convoluted.

If you want to use a loop, you might be better off by using the aiohttp session exposed by http directly:

from typing import Iterator

from pfun import http
import aiohttp

def paginate(url: str) -> Effect[http.HasHTTP, http.ClientError, Iterator[Page]]:
    async def pages(session: aiohttp.ClientSession) -> Iterator[Page]:
        async with session.get(url) as response:
            page = await parse(response)
        while True:
            yield page
            if not has_next_page(page):
                return
            page = await get_next_page(session, page)
        
    return http.get_session().map(pages)

At first glance this looks impure, but since pages is not called until paginate(...).run(Env()) is called, it is actually pure (to the extent possible anyway).

[chuckwondo]

Unfortunately, the issue is that we cannot use Iterator, but rather must use AsyncIterator when directly dealing with the session in this manner, so paginate must be typed as follows (I've simplified things such that paginate will only yield a single page, to avoid multiple HTTP requests for the moment):

import asyncio
from typing import AsyncIterator

import aiohttp
from pfun import http
from pfun.effect import Effect

_A = TypeVar("_A")

HTTPEffect = Effect[http.HasHTTP, aiohttp.ClientError, _A]

class Env:
    http = http.HTTP()


def paginate(url: str) -> HTTPEffect[AsyncIterator[str]]:
    async def pages(session: aiohttp.ClientSession) -> AsyncIterator[str]:
        async with session.request("get", url=url) as response:
            page = await response.text()
        yield page

    return http.get_session().map(pages)


async def main(url: str) -> None:
    async for page in paginate(url).run(Env()):
        print(page)


if __name__ == "__main__":
    asyncio.run(main("https://reqbin.com/echo/get/json"))

The problem is that since running paginate produces an AsyncIterator, we must use async for, and thus must do so within an async function (main, in this example), which requires use of asyncio.run, producing the following exception when we run this:

RuntimeError: asyncio.run() cannot be called from a running event loop
sys:1: RuntimeWarning: coroutine 'CEffect.__call__' was never awaited

I haven't dug into the details of pfun, but I can only imagine the invoking Effect.run ends up attempting to invoke asyncio.run, but given that I must invoke asyncio.run at the top level, we get the error above.

Do you have any ideas on how to address this?

[chuckwondo]

Aha! I figured it out. The main needs to look like so:

async def main(pages: AsyncIterator[str]) -> None:
    async for page in pages:
        print(page)


if __name__ == "__main__":
    url = "https://reqbin.com/echo/get/json"
    paginate(url).map(main).run(Env())

That works, which is great, but I'm a bit confused by this mypy error I'm seeing on the very last line above:

Argument 1 to "map" of "Effect" has incompatible type "Callable[[AsyncIterator[str]], Coroutine[Any, Any, None]]";
expected "Callable[[AsyncIterator[str]], Awaitable[<nothing>]]"

Any idea on why that might be? (I do have the pfun plugin configured.)

[suned]

Hmm, it seems to be an issue with mypy itself actually, involving Union, TypeVar and Awaitable. This produces a similar error:

from typing import Union, TypeVar, Awaitable


A = TypeVar('A')


def f(a: Union[A, Awaitable[A]]) -> A:
    pass


a: Awaitable[str]
f(a)

If f and Effect.map is changed to use @overload instead of Union, it works. I'll fix that in master for now and consider submitting an issue with mypy.