You are viewing an old version of this page. View the current version.

Compare with Current View Page History

« Previous Version 7 Next »

With sapi.py (project manager)

Available since version 0.5

On the root of the Search API project execute on the terminal the following command:

python sapi.py new endpoint <name>

This will generate a bare minimum endpoint with all the steps and changes view on the manual section, and some route examples

Manual

Use this section if sapi.py is not avialable in your distribution

1) Create the Endpoint package

  • Create a folder in app/api/v1

  • Inside the folder create the __init__.py and the endpoint file

    A descriptive name works the best


On the endpoint file


2) Add the imports

  • Special attention to APIRouter

from fastapi import APIRouter, Body
from fastapi.requests import Request

3) Declare the endpoint Router

  • This is essentially a small branch of the app, which will hold all our related routes

  • The prefix indicates the prefix of the endpoint

    All endpoint have the same base path, /es/api/v1, from here we attach the prefix, in this case being /Simple, so /es/api/v1/Simple

router = APIRouter(prefix='/Simple', tags=['Simple'])

4) Declare the routes

  • At this point, all the routes are pure FastAPI
  • The content of each route depends entirely on the one implementing the endpoint

If you want more information on how to build routes, and what features are available please go to the FastAPI tutorial, whenever you see @app, just replace that with @router

@router.post('/post')
async def post_request(req: Request, body: dict = Body()):
    return body

@router.post('/put/{name}')
async def post_request(req: Request, name: str):
    return {"name": name}

@router.post('/get/{id}')
async def post_request(req: Request, id: int, param_a: int = 0):
    return f'GET Request for id {id}, with number {param_a}'

In this scenario the path for each route is:

/es/api/v1/Simple/post

/es/api/v1/Simple/put/lincon

/es/api/v1/Simple/get/42


on the __init__ file


5) Add the import for your router

  • Go to the  __init__.py file
  • Import the router from the API file

If you don’t import the router in the __init__.py, Search API won’t load it


from .simple import router

6) Verify Endpoint Status

  • Starting the server check the logs for the entry loading the endpoint
  • In case of error, the logs will show the issue
    • Either unable to load the endpoint or unable to initialize it
    • Otherwise you should see Endpoint <endpoint path> added

  • You can go to /es/docs or /es/redoc and check the
    API documentation

Swagger (/es/docs)

ReDoc (/es/redoc)

Endpoint as a Pipeline Wrapper

Using an endpoint to wrap a pipeline gives some advantages when working on a project, specially if a separate team is needs to consume said endpoint, specially when compared to use the provided Pipeline endpoint

General Pipeline Endpoint

  • Pipeline endpoint only accepts POST, without restriction of what parameters can accept
  • User needs to know the parameters each pipeline accepts
  • Prone to data input errors

Specific Pipeline Endpoint

  • FastAPI allow to pass a Pydantic model as the Body, meaning all data validation
  • Multiple methods (GET, PUT, POST, DELET) can apply to same pipeline
  • Can add query parameters or path Parameters
  • Customize pipeline response
  • API documentation & User guidance for data input

How To Use The Pipeline In An API

We will use the Search Endpoint for this example

1) Create a Model

  • Create a file for the model in models/api

    The __init__.py is already included in this package



on the model file


2) Add the imports

  • Since we are making a model from scratch we need to import BaseModel

from typing import Optional

from pydantic import BaseModel, Field, Extra

from framework.aggregations.utils import SelectionAgg
from models.engines import SortEntry
from models.utils import BoolOperation

3) Declare the model class

  • All models must inherit from BaseModel or from another model

  • Declare parameters as we did for Stages

class ApiSearchRequest(BaseModel):
    q: Optional[str] = Field(default='*', description='Query string')
    query: Optional[dict] = Field(default=None, description='Specific query object for search engine')
    knn: Optional[dict] = Field(default=None, description='Specific knn query object for search engine')
    size: Optional[int] = Field(default=25, description='Number of hits to return per request')
    sort: Optional[SortEntry] = Field(description='Sort field and order')
    start:  Optional[int] = Field(alias='from', description='Start position for retrieving hits', ge=0)
    page: Optional[int] = Field(description='Start page for retrieving hits. Minimum page is 1. Not applicable when start, is being used', ge=1)
    fetch_fields: Optional[list[str]] = Field(
        description='List of fields to return in the response based on field values', title='Fetch Fields')
    scroll: Optional[str] = Field(default=None, description='Period to retain the search context for scrolling')
    default_operator: BoolOperation = Field(BoolOperation.OR,
                                            description='The default operator for query string query: AND or OR',
                                            title='Default Operator')
    exclude_fields: Optional[list[str]] = Field(
        description='List of fields to exclude in the response based on field values.', title='Exclude Fields')
    aggs: Optional[list[SelectionAgg]] = Field(
        description='List of selected aggregations. Usable only when a DynamicAggStage exist in the pipeline',
        title='Aggregations')

We will not import the model in the __init__, because we want the route to be import like this

from models.api.api_search_request import ApiSearchRequest


on the endpoint file


4) Declare the model class


Assume all steps for adding an endpoint have been done

  • Import the model

  • Import the pipeline_manager

import os
from typing import Optional

from fastapi import APIRouter
from fastapi.params import Query
from starlette import status
from starlette.requests import Request
from starlette.responses import JSONResponse

from models.api.api_search_request import ApiSearchRequest # <---
from app.pipeline import PipelineImpl, pipeline_manager # <---
from models.utils import BoolOperation
from utils.data import find

5) Declare at least the required 2 parameters

  • The first one will be the req, which will be the request sent to the endpoint, this request is of type Request from the model starlette.requests or fastapi.requests (which is an alias for starlette)
  • The second parameter will be you body, in this case we name it payload, and the type will be the model you created
@router.post('/')
async def execute_post_search(req: Request, payload: ApiSearchRequest):

6) Use the pipeline

  • Get the pipeline implementation using the pipeline_manager

  • Then execute the pipeline calling the method execute_pipline, which accepts the request and the model as a dicitonary

    you can turn the model into a dictionary using .dict(), for more information regarding this method please visit Pydantic 1.10

pipeline: PipelineImpl = pipeline_manager.get_pipeline('search')
result = await pipeline.execute_pipeline(req, props=payload.dict(exclude_none=True))

Pipeline executes are asynchronous, an await is need to get the response. And this can only be called from an async function

7) Get the response and do whatever you want

  • In this case we are checking if the response of the stage named search is correct, if it has a status
  • Get the total from it and add it as a new field on the root response
  • and remove the search response entirely
    if 'status' in result['search'] and result['search']['status'] > 299:
        return JSONResponse(status_code=result['search']['status'], content=result['search']['error'])

    if 'status_code' in result['search'] and result['search']['status_code'] > 299:
        return JSONResponse(status_code=result['search']['status_code'], content=result['search']['error'])

    result['total'] = find(result, 'search.total')

    result.pop('search', None)

    return result
  • No labels