Use this section if gapi.py is not avialable in your distribution
Inside the folder create the __init__.py and the endpoint file
Tip |
---|
A descriptive name works the best |
On the endpoint file
Special attention to APIRouter
Code Block | ||||
---|---|---|---|---|
| ||||
from fastapi import APIRouter, Body from fastapi.requests import Request |
The prefix indicates the prefix of the endpoint
Info |
---|
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 |
Code Block | ||||
---|---|---|---|---|
| ||||
router = APIRouter(prefix='/Simple', tags=['Simple']) |
Tip |
---|
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 |
Code Block | ||||
---|---|---|---|---|
| ||||
@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}' |
Info |
---|
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
Warning |
---|
If you don’t import the router in the __init__.py, GAIA API won’t load it |
Code Block | ||||
---|---|---|---|---|
| ||||
from .simple import router |
You can go to /es/docs or /es/redocand 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 is needs to consume said endpoint, specially and when compared to use the provided Pipeline endpoint
General Pipeline Endpoint
Specific Pipeline Endpoint
Tip |
---|
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
Info |
---|
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
Code Block | ||||
---|---|---|---|---|
| ||||
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
Code Block | ||||
---|---|---|---|---|
| ||||
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') |
Note | ||
---|---|---|
We will not import the model in the __init__, because we want the route to be import like this
|
on the endpoint file
Note |
---|
Assume all steps for adding an endpoint have been done |
Import the model
Import the pipeline_manager
Code Block | ||||
---|---|---|---|---|
| ||||
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 |
Code Block | ||||
---|---|---|---|---|
| ||||
@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
Info |
---|
you can turn the model into a dictionary using .dict(), for more information regarding this method please visit Pydantic 1.10 |
Code Block | ||||
---|---|---|---|---|
| ||||
pipeline: PipelineImpl = pipeline_manager.get_pipeline('search') result = await pipeline.execute_pipeline(req, props=payload.dict(exclude_none=True)) |
Note |
---|
Pipeline executes are asynchronous, an await is need to get the response. And this can only be called from an async function |
Code Block | ||||
---|---|---|---|---|
| ||||
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
Code Block | ||||
---|---|---|---|---|
| ||||
from fastapi.params import Query |
Code Block | ||||
---|---|---|---|---|
| ||||
@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) |