Use this section if gapi.py is not avialable in your distribution
Inside the folder create the __init__.py and the endpoint file
A descriptive name works the best
On the endpoint file
Special attention to APIRouter
from fastapi import APIRouter, Body from fastapi.requests import Request
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'])
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
If you don’t import the router in the __init__.py, GAIA API won’t load it
from .simple import router
You can go to /es/docs or /es/redoc and check the
API documentation
Swagger (/es/docs)
ReDoc (/es/redoc)
Using an endpoint to wrap a pipeline gives some advantages when working on a project, specially if a separate team needs to consume said endpoint, and when compared to use the provided Pipeline endpoint
General Pipeline Endpoint
Specific Pipeline Endpoint
The following can be achieve by applying the concepts seen in the FastAPI documentation specifically Request Body, Query Parameters and String Validations and Path Parameters and Numeric Validations
We will use the Search Endpoint for this example
Create a file for the model in models/api
The __init__.py is already included in this package
on the model file
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
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
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
@router.post('/') async def execute_post_search(req: Request, payload: ApiSearchRequest):
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
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
Query object server the same function as the Field from Pydantic, but if from FastAPI
from fastapi.params import Query
@router.get('/') async def execute_get_search( req: Request, q: str = Query(default='*', description='String query to execute in the search engine'), size: int = Query(default=25, description='Number of hits to return per request'), # sort: Optional[SortEntry] = Query(description='Sort field and order'), start: Optional[int] = Query(default=None, alias='from', description='Start position for retrieving hits', ge=0), page: Optional[int] = Query(default=None, description='Start page for retrieving hits. Minimum page is 1. Not applicable ' 'when from, is being used', ge=1), fetch_fields: Optional[list[str]] = Query(default=[], description='List of fields to return in the response based on ' 'field values', title='Fetch Fields'), scroll: Optional[str] = Query(default=None, description='Period to retain the search context for scrolling'), default_operator: BoolOperation = Query(default=BoolOperation.OR, description='The default operator for query string query: AND or OR', title='Default Operator'), exclude_fields: Optional[list[str]] = Query(default=None, description='List of fields to exclude in the response based on field values.', title='Exclude Fields'), ): ''' GAIA API endpoint, working as a wrapper for a specific pipeline, which will execute with the provided parameters. The main goal of this endpoint is to serve as an example of how static, well-defined endpoints can be created with a flexible and configurable backend **NOTE**: The starting parameters provided by swagger are almost the minimum _(the minimum being just q)_, for more options please check the schema `Remove this endpoint or update the pipeline it is using, this endpoint is not intended for production` ''' pipeline: PipelineImpl = pipeline_manager.get_pipeline('search') payload: ApiSearchRequest = ApiSearchRequest(q=q, size=size, start=start, page=page, fetch_fields=fetch_fields, scroll=scroll, default_operator=default_operator, exclude_fields=exclude_fields)