REST Client

In this example we will build a REST Client that fetches data from a fictional API that we’ll build ourselves.

The purpose of this tutorial is not to teach you how to build REST APIs, but to show you how to use DataService to fetch data from them, so feel free to go ahead and run the server code below to get the API up and running.

"""Fictional API server for demonstration purposes."""

from faker import Faker
from fastapi import FastAPI
from fastapi_pagination import Page, add_pagination, paginate

from examples.api.models import User

app = FastAPI()
fake = Faker()

users = [  # create some data
    User(name=fake.first_name(), surname=fake.last_name()) for _ in range(1000)
]


@app.get("/users")
async def get_users() -> Page[User]:
    return paginate(users)


add_pagination(app)

This is a simple FastAPI server that serves a list of Users using page based pagination provided by fastapi-pagination.

You can run it by executing the following commands:

$ pip install "fastapi[all]" fastapi-pagination faker

$ fastapi dev api_server.py

We can now proceed to build the REST Client that will fetch data from the API.

First we define a simple User model that will hold the user details. We are using a simple pydantic.BaseModel instead of BaseDataItem here because we are not expecting exceptions to be raised. Please note that the same model is used in the server code.

class User(BaseModel):  # define your model
    name: str = Field(..., examples=["Steve"])
    surname: str = Field(..., examples=["Rogers"])

Then we define the parse_users callback that will extract the user details from the API response and yield a new User object for each user.

def parse_users(response: Response):
    users = response.data["items"]
    for user in users:
        yield User(**user)

We then proceed by defining the paginate callback which will iterate over the total number of pages and yield a new Request object for each one.

def paginate(response: Response):
    pages = response.data["pages"]
    yield from parse_users(response)
    for p in range(2, pages + 1):
        yield Request(
            url=response.request.url,
            callback=parse_users,
            client=response.request.client,
            params={"page": p},
            content_type="json",
        )

We are starting from page 2 because the first page is already served by the initial request.

Finally we define the main function that will create the initial Request object and run the DataService.

Full code for the api_client example:

import logging

from dataservice import (
    DataService,
    HttpXClient,
    Request,
    Response,
    ServiceConfig,
    setup_logging,
)
from examples.api.models import User

logger = logging.getLogger("api_client")

setup_logging("api_client")


def parse_users(response: Response):
    users = response.data["items"]
    for user in users:
        yield User(**user)


def paginate(response: Response):
    pages = response.data["pages"]
    yield from parse_users(response)
    for p in range(2, pages + 1):
        yield Request(
            url=response.request.url,
            callback=parse_users,
            client=response.request.client,
            params={"page": p},
            content_type="json",
        )


def main():
    httpx_client = HttpXClient()
    start_requests = [
        Request(
            url="http://127.0.0.1:8000/users",
            callback=paginate,
            client=httpx_client,
            content_type="json",
        )
    ]
    data_service = DataService(
        start_requests, config=ServiceConfig(**{"limiter": {"max_rate": 10}})
    )
    data = tuple(data_service)
    for item in data:
        logger.info(item)


if __name__ == "__main__":
    main()

A few things to note in this example:

Each Request object is created with a content_type of json to tell the Client we are expecting a JSON response. The paginate callback is also yielding Request objects using the params argument to build the URL for each page. Finally, our fictional API is rate limited to 10 requests per minute, so we are using a limiter from aiolimiter.