API Routes

In this page we will go through all the routes we implemented with FastAPI.

Each route is basically a web URL that accept data in JSON format, performs some operations and then reply with a status code and a result.

The status code can be used to discriminate if a request has been executed successfully (2xx values) or if there was an internal error on the server (5xx values) or an error from the client (4xx values).

Data exchange

Before getting into details, we need to introduce how data is shared between the user and the application. Following the best practices, data has to be shared in a structured way (a JSON for example). FastAPI has some neat way to do this using a module called pydantic.

Pydantic defines the format of the data using Python classes, in this way it is possible to shape the information in a strict format making them easy to manage. In the following snippet of code, there are the classes we wrote to define the structure of the data. A brief explanation of what each class defines is written as a comment.

An example of a data definition class is shown below, full class implementation can be found in api/requests.py.

class TaskStatus(BaseModel):
    """Class that defines the output data of an inference."""
    task_id: str
    status : str
    type: str

This class will let the API produce a JSON containing three fields: task_id is an unique identifier for the task started; status defines the current status of task; type is an extra field that tells the user which kind of task has been started.

Routes

Now we do a brief explanation of each route implemented in out API.

/inference/start

@api.post('/inference/start')
async def schedule_inference(user_data: requests.UserData, db: Session = Depends(get_db)):
    """This is the endpoint used for schedule an inference."""
    crud.create_event(db, 'inference_start')
    ud = crud.create_user_data(db, user_data.dict())
    task: AsyncResult = inference.delay(ud.user_id)

    db_inf = crud.create_inference(db, task.task_id, task.status)
    return requests.TaskStatus(
        task_id=db_inf.task_id,
        status=db_inf.status,
        type='inference'
    )

/inference/start is used to start an inference task with the current active model.

This route receive as parameters:

• the user’s data which contains an ID and some features,

• a open session to a database generated by FastAPI.

First of all, we log the event in the database at line 54, then we store the received data. At line 6 we start the asynchronous process by telling Celery to spawn an inference task. user_id is then used to retrieve the data stored in the database at line 5.

At line 8 we insert a new entry in the database that help the user tracking the status of the task. Once the task is finished, the user can retrieve the predictions using another route.

The route, at the end, return the status of the task to the user.

/inference/status/{task_id}

@api.get('/inference/status/{task_id}', response_model=requests.TaskStatus)
async def get_inference_status(task_id: str, db: Session = Depends(get_db)):
    """This is the endpoint to get the results of an inference."""
    crud.create_event(db, 'status')
    
    task = AsyncResult(task_id)
    task_id, status = task.task_id, task.status

    db_inf = crud.get_inference(db, task_id=task_id)

    if db_inf is None:
        raise HTTPException(status_code=404, detail='Task not found')

    db_inf.status = status
    db_inf = crud.update_inference(db, db_inf.task_id, db_inf.status)

    if task.failed():
        raise HTTPException(status_code=500, detail='Task failed')

    if not task.ready():
        return requests.TaskStatus(
            task_id=db_inf.task_id,
            status=db_inf.status,
            type='inference',
        )

    return requests.TaskStatus(
        task_id=db_inf.task_id,
        status=db_inf.status,
        type='inference',
    )

/inference/status/{task_id} is used to check the status of the task running asynchronously.

This route accept as parameter the task_id associated to the process spawned by celery.

The extra code is similar to the previous route: it tracks the event and contains the logic to create the correct response.

/inference/results/{task_id}

@api.get('/inference/results/{task_id}')
async def get_inference_results(task_id: str, limit: int=10, db: Session = Depends(get_db)):
    """This is the endpoint to get the results with scores once the inference 
    task has been completed.
    
    Note: check the status of the task with the '/inference/status' endpoint.
    """
    crud.create_event(db, 'results')

    locations = crud.get_results_locations(db, task_id, limit)

    crud.mark_locations_as_shown(db, task_id, locations)

    return locations

/inference/results/{task_id} is used to retrieve the results of the inference once the response of /inference/status/{task_id} notify the successful completion of an inference.

Note how on line 10 we get from the database the locations with the score produced by the inference.

/inference/select/

@api.put('/inference/select/')
async def get_click(label: requests.LabelData, db: Session = Depends(get_db)):
    """This is the endpoint used to simulate a click on a choice.
    A click will be registered as a label on the data"""
    crud.create_event(db, 'selection')

    if label.location_id == -1:
        crud.create_event(db, 'bad_inference')
        return

    else:
        crud.create_event(db, 'good_inference')
        db_result = crud.update_result_label(db, label.task_id, label.location_id)

        if db_result is None:
            return HTTPException(404, f"Result not found: invalid task_id or location_id")

        return db_result

Since we want to simulate the act of a user that selects the desired location from a list, we use the /inference/select route for this purpose.

The routes expects to receive an id that indicates the selection of an user. This creates an history of labelled data that can be used to train a new model.

In the case no location has been selected, we register a bad_inference event on our log. Otherwise, we update the data in the database.

/train/start

@api.post('/train/start')
async def schedule_training(db: Session = Depends(get_db)):
    """This is the endpoint to start the training of a new model."""
    crud.create_event(db, 'training')

    task: AsyncResult = training.delay()

    db_model = crud.create_model(db, task.task_id, task.status)
    return requests.TaskStatus(
        task_id=db_model.task_id,
        status=db_model.status,
        type='training'
    )

The training of a new model is an operation that can be started manually only. This routes schedule the second type of task on Celery.

/content/…

The routes that begins with /content are used to debug what is stored in the database.

These routes are reported there as a demonstrative example and can be simply ignored.

@api.get('/content/info')
async def get_content_info(db: Session = Depends(get_db)):
    n_locations = crud.count_locations(db)
    n_users = crud.count_users(db)

    return requests.ContentInfo(
        locations=n_locations,
        users=n_users,
    )


@api.get('/content/location/{location_id}')
async def get_content_location(location_id: int, db: Session = Depends(get_db)):
    return crud.get_location(db, location_id)


@api.get('/content/locations')
async def get_content_locations(db: Session = Depends(get_db)):
    return crud.get_locations(db)


@api.get('/content/user/{user_id}')
async def get_content_user(user_id: int, db: Session = Depends(get_db)):
    return crud.get_user(db, user_id)


@api.get('/content/users')
async def get_content_users(db: Session = Depends(get_db)):
    return crud.get_users(db)


@api.get('/content/result/{result_id}')
async def get_content_result(result_id: int, db: Session = Depends(get_db)):
    return crud.get_result(db, result_id)


@api.get('/content/results')
async def get_content_result(task_id: str, db: Session = Depends(get_db)):
    return crud.get_results(db, task_id)