As mentioned in the last article, I wish I could stick with my notes on learning Fastapi for python Web development framework. Since there is a beginning, so also hope that they have an end, even if there is no end ~ also at least a process! So continue sorting before I run into some problems using the process again!

Fastapi Base add-on

The introduction of relevant knowledge in the last article mainly refers to the number of examples provided on the official website, which is just a practical exercise based on the above operation. However, in practical application, there are still many problems involved. In fact, there are many points that need to be paid attention to.

1 Fastapi App object initialization configuration items

app = FastAPI()
Copy the code

From instantiating an application, our FastAPI() just uses the default parameters, but there are still a lot of configuration items involved. Let me add something here.

From the configuration items, we can learn the following:

• 1: How to describe the app object
• 2: How to configure our API document address and close this document address
• 3: How to add notification of APP startup and shutdown event callback

First we enter the internal source code to see:

class FastAPI(Starlette):
    def __init__(
        self,
        *,
        debug: bool = False,
        routes: Optional[List[BaseRoute]] = None,
        title: str = "FastAPI",
        description: str = "",
        version: str = "0.1.0",
        openapi_url: Optional[str] = "/openapi.json",
        openapi_tags: Optional[List[Dict[str, Any]]] = None,
        servers: Optional[List[Dict[str, Union[str, Any]]]] = None,
        dependencies: Optional[Sequence[Depends]] = None,
        default_response_class: Type[Response] = Default(JSONResponse),
        docs_url: Optional[str] = "/docs",
        redoc_url: Optional[str] = "/redoc",
        swagger_ui_oauth2_redirect_url: Optional[str] = "/docs/oauth2-redirect",
        swagger_ui_init_oauth: Optional[Dict[str, Any]] = None,
        middleware: Optional[Sequence[Middleware]] = None,
        exception_handlers: Optional[
            Dict[
                Union[int, Type[Exception]],
                Callable[[Request, Any], Coroutine[Any, Any, Response]],
            ]
        ] = None,
        on_startup: Optional[Sequence[Callable[[], Any]]] = None,
        on_shutdown: Optional[Sequence[Callable[[], Any]]] = None,
        openapi_prefix: str = "",
        root_path: str = "",
        root_path_in_servers: bool = True,
        responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None,
        callbacks: Optional[List[BaseRoute]] = None,
        deprecated: Optional[bool] = None,
        include_in_schema: bool = True,
        **extra: Any,
    ) -> None:
    `````
    `````
Copy the code

There are a lot of parameters involved, but all of them are non-mandatory and have default values.

Here I will simply comment on some parameters:

• Debug: bool = False Indicates whether to enable debugging

• Routes: Optional[List[BaseRoute]] = None List of routes attached to the APP object

• Title: STR = “FastAPI”, define a title text for the document of our app, which will correspond to the title on our document

• Description: STR = “” App document description

• Version: STR = “0.1.0”, app application document version

• Openapi_prefix: STR = “” Access prefix of the API address accessed

• Openapi_url: Optional[STR] = “/openapi.json”

PS: We can import this directly into Swagger later, and then we can see all the interfaces of our application. Swagger for openAPI specification document format

The following example shows our API documentation:

return get_swagger_ui_html( openapi_url=openapi_url, title="Swagger UI", oauth2_redirect_url=oauth2_redirect_url, init_oauth=app.swagger_ui_init_oauth, Swagger_js_url = "https://cdn.jsdelivr.net/npm/[email protected]/swagger-ui-bundle.js", Swagger_css_url = "https://cdn.jsdelivr.net/npm/[email protected]/swagger-ui.css",)Copy the code

• Openapi_tags: Optional[List[Dict[STR, Any]] = None Specifies the label of each API document interface

• Servers: Optional[List[Dict[STR, Union[STR, Any]]]] = None

• Dependencies: Optional[Sequence[Depends]] = None

• Default_response_class: Type[Response] = Default(JSONResponse) The Default Response message returns a Response in application/ JSON format. This is the response that FastAPI returns by default.

• Docs_url: Optional[STR] = “/docs” Swagger API

If set to None, the API document does not exist

• Redoc_url: Optional[STR] = “/redoc”

• Swagger_ui_init_oauth: Optional[Dict[STR, Any]] = None, Swagger UI API document authorization initialization

• Swagger_ui_oauth2_redirect_url: Optional[STR] = “/docs/ oAuth2-redirect “Swagger_UI_oAuth2_redirect_URL: Optional[STR] = “/docs/ oAuth2-redirect

• Middleware: Optional[Sequence[middleware]] = None Specifies the middleware of an application

• exception_handlers: Optional[ Dict[ Union[int, Type[Exception]], Callable[[Request, Any], Coroutine[Any, Any, Response]], ]] = None The applied exception error handler object that can pass in the corresponding exception handling for the function

Example:

from starlette.responses import Response async def Not404Fount(request: Request, exc): response = Response('{"Error": "500"}') return response async def Not500Fount(request: Request, exc): msg = '' if isinstance(exc, ZeroDivisionError): MSG = 'ZeroDivisionError' response = response (MSG) return response # Not404Fount) # add 404 app.add_Exception_Handler (500, Not500Fount)Copy the code

• On_startup: Optional[Sequence[Callable[[], Any]]] = None Callback function used when the application is started

• On_shutdown: Optional[Sequence[Callable[[], Any]]] = None Callback function when the application exits

Examples of starting and closing callbacks:

Def f_startup(): print(" start ") pass def f_shutdown(): Add_event_handler ('startup', f_startup) # print('shutdown', f_shutdown)Copy the code

• Root_path: STR = “”, the root directory of the current application

• root_path_in_servers: bool = True,

• Deprecated: Optional[bool] = None Indicates whether the current version of the interface document is available. If True:

• Include_in_schema: bool = True indicates whether to enable API document viewing, False indicates that all generated OpenAPI schemas (and automatic document systems) are cleared automatically

2 Parameters deprecated and include_IN_SCHEMA Other descriptions

In the app object our deprecated and include_in_SCHEMA are global, but what if we only exclude interfaces from documentation or mark the availability of the current version?

Sample code:

import uvicorn from fastapi import FastAPI, File, UploadFile app = FastAPI() from fastapi import BackgroundTasks, FastAPI, Depends app = FastAPI(title="XXXX message management system ", description=' this system is mainly used for XXXX', Version ='v1.0.0') import threading @app.get("/sync3") def sync3(): Print ('sync- current thread name :', threading.current_thread().getName()) return "Sync task completed!" @app.get("/sync",include_in_schema=False) def sync(): @app.get("/sync",include_in_schema=False) def sync(): Print ('sync- current thread name :', threading.current_thread().getName()) return "Sync task completed!" @app.get("/async",deprecated=True) async def asyncAction (): Print (app.routes) print(' asyncAction ', threading.current_thread().getName()) return "Async task completed!" if __name__ == '__main__': # equal to the uvicorn command line uvicorn script name :app object start service: Uvicorn XXX :app --reload uvicorn. Run ('frun:app', host="127.0.0.1", port=8000, debug=True, reload=True)Copy the code

Document presentation results:

3 Write an API description

PS: In order to see the document, the first time must be the Internet drop yo! Of course, you can use custom access to the static page /docs required resources, the corresponding document access mode, related to the resource address request to local:

Replace get_swagger_UI_html with get_swagger_ui_html

• swagger_js_url
• swagger_favicon_url
• oauth2_redirect_url

Get_swagger_ui_html source code:

def get_swagger_ui_html(
    *,
    openapi_url: str,
    title: str,
    swagger_js_url: str = "https://cdn.jsdelivr.net/npm/swagger-ui-dist@3/swagger-ui-bundle.js",
    swagger_css_url: str = "https://cdn.jsdelivr.net/npm/swagger-ui-dist@3/swagger-ui.css",
    swagger_favicon_url: str = "https://fastapi.tiangolo.com/img/favicon.png",
    oauth2_redirect_url: Optional[str] = None,
    init_oauth: Optional[Dict[str, Any]] = None,
) -> HTMLResponse:
Copy the code

3.1 Basic Interface Document Description

We know that the swagger API document is very useful, but we need to configure some API description, otherwise we do not know the interface why! In fact, to put it bluntly, I am a newbie in English and need Chinese to interpret! Ha ha

This interface, for example, has no idea what it’s for!

A good API interface documentation is still needed!

Example:

#! /usr/bin/evn python # coding=utf-8 import uvicorn from fastapi import FastAPI, File, UploadFile app = FastAPI() from FastAPI import BackgroundTasks, FastAPI, Depends app = FastAPI(title="XXXX ", Import threading @app.get("/sync3") def sync3(): Print ('sync- current thread name :', threading.current_thread().getName()) return "Sync task completed!" @app.get("/sync", include_in_schema=False) def sync(): @app.get("/sync", include_in_schema=False) def sync(): Print ('sync- current thread name :', threading.current_thread().getName()) return "Sync task completed!" @app.get("/async", deprecated=True) async def asyncAction (): Print (app.routes) print(' asyncAction ', threading.current_thread().getName()) return "Async task completed!" @app.get(path='/ API /v1/get/user', summary=' getuser ', description=' this interface is used to add users ', tags=[' user module ']) def getUser (): Return {"code": 0, "MSG ":" get success ", "data": None} @app.post(path='/ API /v1/add/user', summary=' adduser ', description=' this interface is used to get user information ', tags=[' user module ']) def adduser(): Return {"code": 0, "MSG ":" added successfully ", "data": None} @app.put(path='/ API /v1/updata/user', summary=' update user', description=' this interface is used to update user information ', Tags = [' user modules']) def updatauser () : return {" code ": 0," MSG ":" success ", "data" : None} @app.put(path='/ API /v1/delete/user', summary=' delete user', description=' this interface is used to delete user information ', Tags = [' user modules']) def deleteuser () : return {" code ": 0," MSG ":" delete the success ", "data" : None} @app.put(path='/ API /v1/add/data', summary=' adddata ', description=' this interface is used to adddata ', tags=[' data module ']) def addDatas (): Return {" code ": 0," MSG ":" delete the success ", "data" : None} if __name__ = = "__main__ ': # equal to the uvicorn command line uvicorn script name :app object start service: Uvicorn XXX :app --reload uvicorn. Run ('frun:app', host="127.0.0.1", port=8000, debug=True, reload=True)Copy the code

Examples of document effects:

3.2 Function of internal API comments description

Example of internal comments for a function:

@app.get("/asyncio/task") async def do_task(username: STR): param username: username: return: Asyncio.create_task (write_log_2()) return "Hello, async block under coroutine!"Copy the code

Where the corresponding document displays comments:

3.3 Describe the interface document using APIRouter

Examples of this code are as follows:

From fastAPI import APIRouter rstest = APIRouter(tags=["APIRouter related "]) @rstest. Post ("/rstest", Name =" new example ",description=" new example ") async def create(): return "hello, async blocking under coroutine!" app.include_router(prefix="/v1", router=rstest) if __name__ == "__main__": Uvicorn. Run (' main: app, host = "127.0.0.1", the port = 8000, reload = True)Copy the code

An example of the result is as follows:

3.4 Interface Parameters Description

Check out our source code

Query: the source code

def Query( # noqa: N802 default: Any, *, alias: Optional[str] = None, title: Optional[str] = None, description: Optional[str] = None, gt: Optional[float] = None, ge: Optional[float] = None, lt: Optional[float] = None, le: Optional[float] = None, min_length: Optional[int] = None, max_length: Optional[int] = None, regex: Optional[str] = None, example: Any = Undefined, examples: Optional[Dict[str, Any]] = None, deprecated: Optional[bool] = None, **extra: Any, based on Query module default value - STR = None - Default parameter declaration - Form: q: STR = Query(" Query ") # Equivalent to q: STR = "Query" - Mandatory parameter declaration - form: q: STR = Query(...) -title: Parameter title - Description: parameter description - Deprecated: Indicates that a parameter is about to expire or will not be available. Special additional information: -alias: indicates the alias of the parameterCopy the code

The Body source code:

def Body( # noqa: N802 default: Any, *, embed: bool = False, media_type: str = "application/json", alias: Optional[str] = None, title: Optional[str] = None, description: Optional[str] = None, gt: Optional[float] = None, ge: Optional[float] = None, lt: Optional[float] = None, le: Optional[float] = None, min_length: Optional[int] = None, max_length: Optional[int] = None, regex: Optional[str] = None, example: Any = Undefined, examples: Optional[Dict[str, Any]] = None, **extra: Any, ) -> Any: -title: parameter title -description: parameter description information -deprecated: Indicates that the parameter is about to expire or is unavailable Special additional information: - alias: parameter aliases example: examples examples: list of examplesCopy the code

Their parameters are passed in with a description of the parameters and an example.

So we can use the above to show the description and explanation of our parameters.

Example:

import uvicorn from fastapi import FastAPI, File, UploadFile app = FastAPI() from fastapi import BackgroundTasks, FastAPI, Depends, Body, Query from pydantic import BaseModel, Field from typing import Optional app = FastAPI(title="XXXX message management system ", description=' this system is mainly used for XXXXXX', Version ='v1.0.0') import threading @app.get("/sync3") def sync3(): Print ('sync- current thread name :', threading.current_thread().getName()) return "Sync task completed!" @app.get("/sync", include_in_schema=False) def sync(): @app.get("/sync", include_in_schema=False) def sync(): Print ('sync- current thread name :', threading.current_thread().getName()) return "Sync task completed!" @app.get("/async", deprecated=True) async def asyncAction (): Print (app.routes) print(' asyncAction ', threading.current_thread().getName()) return "Async task completed!" @app.get(path='/ API /v1/get/user', summary=' getuser ', description=' this interface is used to add users ', tags=[' user module ']) def getUser (): Return {"code": 0, "MSG ":" get success ", "data": None} @app.post(path='/ API /v1/add/user', summary=' adduser ', description=' this interface is used to get user information ', tags=[' user module ']) def adduser(): Return {"code": 0, "MSG ":" added successfully ", "data": None} @app.put(path='/ API /v1/updata/user', summary=' update user', description=' this interface is used to update user information ', Tags = [' user modules']) def updatauser () : return {" code ": 0," MSG ":" success ", "data" : None} @app.delete(path='/ API /v1/delete/user', summary=' delete user', description=' this interface is used to delete user information ', Tags = [' user modules']) def deleteuser () : return {" code ": 0," MSG ":" delete the success ", "data" : None} @app.put(path='/ API /v1/add/data', summary=' adddata ', description=' this interface is used to adddata ', tags=[' data module ']) def addDatas (): Return {"code": 0, "MSG ":" new data", "data": None} @app.get("/ API /v1/ceshi/data", summary=' view data ', description=' this interface is used to add data ', Tags =[' data module ']) async def read_items(uerid: STR = Query(... , min_length=3, max_length=50, title=' user uerID ', description=' Alias ='UID', example="UUU323423385783785"), uerName: STR = Query(... , min_length=3, max_length=50, title=' uername', description=' uername', alias='UNAME', example= 'uername'),): results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]} print(uername) if uerid: results.update({"uerid": uerid}) return results class Item(BaseModel): name: str description: STR = Field(None, title=" description ", description=" description ", max_length=300) price: float = Field(... , gt=0, title=" price ", description=" price ") tax: float = None # class Item(BaseModel): # name: STR # description: str = None # price: float # tax: Float = None @app.post("/ API /v1/udaoay/data", summary=' change data ', description=' change data ', Tags =[' data module ']) async def udData (item: item = Body(... , embed=False, example={'para1': "required, format required string ", 'name':" required, format required string ", 'description': "optional, format required string, default is None", 'price': "Float are mandatory, format requirements", "tax", "floating point are optional, format requirements"})) : return {" code ": 0," MSG ":" success ", "data" : the item} if __name__ = = "__main__ ': # equal to the uvicorn command line uvicorn script name :app object start service: Uvicorn XXX :app --reload uvicorn. Run ('frun:app', host="127.0.0.1", port=8000, debug=True, reload=True)Copy the code

Examples of results for query parameters are described as follows:

Example result descriptions of parameters in the request body are as follows:

3.5 Example of using model Config to describe interface parameters

class Item(BaseModel): name: str description: STR = Field(None, title=" description ", description=" description ", max_length=300) price: float = Field(... Tax: float = None class Config: schema_extra={"example": tax: float = None class Config: schema_extra={ {"name":" description details ", "price": }} @app.post("/ API /v1/udaoay/data", summary=' modify data ', description=' this interface is used to add data ', Tags =[' data module ']) async def udData (item: item = Body(... , embed=False, example={'para1': "required, format required string ", 'name':" required, format required string ", 'description': "optional, format required string, default is None", 'price': "Float are mandatory, format requirements", "tax", "floating point are optional, format requirements"})) : return {" code ": 0," MSG ":" success ", "data" : the item}Copy the code

4 API security Issues

API documentation for our project is all the entry point, security is certainly not to be ignored. The security schemes are as follows:

• 1: in the online environment, close the API file and cancel the ADDRESS of the API file
• 2: globally close all interfaces of the API document through the global include_in_SCHEMA close
• 3: Only one instance of the API visible on the Intranet is supported
• 4: Enable Base Auth through Nginx to implement user authentication or restrict the corresponding IP access to an API address
• 5: Restrict the source IP address of the request on the APP
• 6: User recognition mechanism is also enabled on App applications

An easy way to supplement your own user and password awareness within your App:

4.1 Restrictions on Global Access Authentication

import uvicorn from fastapi import Depends from fastapi import FastAPI, File, UploadFile from config import settings from fastapi.security import HTTPBasic, HTTPBasicCredentials from fastapi.exceptions import HTTPException from fastapi.openapi.docs import get_swagger_ui_html From starlette. Status import HTTP_401_UNAUTHORIZED app = FastAPI(title="XXXX message management system ", description=' this system is mainly in XXXXXX', Version =settings.API_V1_STR, Docs_url =None) security = HTTPBasic() @app.get("/docs",include_in_schema=False) async def is required get_documentation(credentials: HTTPBasicCredentials = Depends(security)): print("sssssssssssssssssss") if credentials.username ! = "xiaozhongtongxue" or credentials.password ! =" xiaozhongtongxue": raise HTTPException(status_code=HTTP_401_UNAUTHORIZED, detail="y user and password incorrect ！！！！！" , headers={"WWW-Authenticate": "Basic"}, ) else: return get_swagger_ui_html(openapi_url="/openapi.json", title="doc") if __name__ == '__main__': # equal to the uvicorn command line uvicorn script name :app object start service: Uvicorn XXX :app --reload uvicorn. Run ('frun:app', host="127.0.0.1", port=8000, debug=True, reload=True)Copy the code

Docs_url =None; docs_url=None

4.2 Limiting Authentication on an Interface

The above way is open, as long as the authentication passed once, I can view all, of course, if the basis of that and then for a certain interface restrictions, it is also possible, just need we need to limit our access to the interface to do the relevant dependency injection restrictions can be! , but it feels like subsequent access to the API connection doesn’t make sense. Compare our interface is for access!

import secrets import uvicorn from fastapi import Depends, FastAPI, HTTPException, status from fastapi.security import HTTPBasic, HTTPBasicCredentials app = FastAPI() # HTTPBasicCredentials app = FastAPI() # Def get_current_username(HTTPBasicCredentials = validity (security)) def get_current_username(HTTPBasicCredentials = validity (security)) Correct_username = secrets.compare_digest(credentials.username, "admin") correct_password = secrets.compare_digest(credentials.password, "admin") if not (correct_username and correct_password): Raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=" the user and password are incorrect ！！！！！" , headers={"WWW-Authenticate": "Basic"}, ) return credentials.username @app.get("/users/msg") def read_current_user(username: str = Depends(get_current_username)): return {"username": username} if __name__ == "__main__": Uvicorn. Run (' main: app, host = "127.0.0.1", the port = 8000)Copy the code

When you need to view the API documentation for a and interface:

5 TYPE of API response packets

For response messages, usually our write API must return the json format, but it does not rule out that I simply need to return string, or return HTML template, return data stream, return XML format, etc.

So there is a need for customization, but our Fastapi is basically packaged, so we don’t need to deal with it ourselves.

PlainTextResponse must be used to return a string similar to a custom error check.

The following is an example:

@app.exception_handler(RequestValidationError) async def validation_exception_handler(request, exc): # return JSONResponse({'mes':' RequestValidationError ') # return JSONResponse({'mes':' RequestValidationError ') '%(STR (exc))}) # TypeError: 'STR' object is not callable return" Need to return using the return PlainTextResponse (" Where's little Sister!" ) to correctCopy the code

5.1 responses type

from fastapi.responses import (HTMLResponse, JSONResponse, ORJSONResponse, Response, PlainTextResponse, UJSONResponse,StreamingResponse,FileResponse)
Copy the code

• Response requires its own reextension of the base Response body
• HTMLResponse Returns an HTML template
• JSONResponse jison format
• ORJSONResponse uses the JSON format of another library
• UJSONResponse uses the JSON format of another library
• StreamingResponse Format of the output stream
• FileResponse Indicates how the file responds to the body

5.2 Customizing Response

An example is returning a response in XML format:

from fastapi import FastAPI, Response app = FastAPI() @app.get("/legacy/") def get_legacy_data(): data = """<? The XML version = "1.0"? > <shampoo> <Header> Apply shampoo here. </Header> <Body> You'll have to use soap here. </Body> </shampoo> """ return Response(content=data, media_type="application/xml")Copy the code

5.3 Return the string PlainTextResponse

Example:

@app.get("/PlainTextResponse1/") async def PlainTextResponse1(): Return PlainTextResponse(' hello ') @app.get("/PlainTextResponse2/", Response_class =PlainTextResponse) async def PlainTextResponse2(): return "Hello!"Copy the code

5.4 Returning the JSON Data Format

Example:

@app.post("/ API /v1/udaoay/data") async def udData (): return {"code": 0, "MSG ":" modified ", "data": Return JSONResponse(status_code=404, content={"message": "Item not found"})Copy the code

5.5 return HTMLResponse

Example:

from fastapi import FastAPI from fastapi.responses import HTMLResponse app = FastAPI() def generate_html_response(): html_content = """ <html> <head> <title>Some HTML in here</title> </head> <body> <h1>Look ma! HTML! </h1> </body> </html> """ return HTMLResponse(content=html_content, status_code=200) @app.get("/items/", response_class=HTMLResponse) async def read_items(): return generate_html_response()Copy the code

5.6 Returning RedirectResponse temporary Redirection

Example:

@app.get("/RedirectResponse/", response_class=HTMLResponse)
async def RedirectResponse():
   return RedirectResponse("https://wwww.baifuw.com")
Copy the code

5.7 Return stream information StreamingResponse

Example 1 is taking an asynchronous generator or a plain generator/iterator and streaming the response body.

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()


async def fake_video_streamer():
    for i in range(10):
        yield b"some fake video bytes"


@app.get("/")
async def main():
    return StreamingResponse(fake_video_streamer()
Copy the code

Example 2 such as: If you have a file-like object (for example, by open()), you can in StreamingResponse.

This includes many libraries that interact with cloud storage, video processing, and more.

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

some_file_path = "large-video-file.mp4"
app = FastAPI()


@app.get("/")
def main():
    file_like = open(some_file_path, mode="rb")
    return StreamingResponse(file_like, media_type="video/mp4")
Copy the code

5.8 return FileResponse

Asynchronously treat the file as a response flow.

Instantiate with a different set of parameters than the other response types:

• Path – The file path of the file to stream.
• Headers – Any custom headers to include as a dictionary.
• Media_type – Displays a string of media types. If not set, the filename or path will be used to infer the media type.
• Filename – If set, this will be included in the reply content-disposition.

The file response will include the appropriate Content-Length, Last-Modified, and ETag headers.

from fastapi import FastAPI
from fastapi.responses import FileResponse

some_file_path = "large-video-file.mp4"
app = FastAPI()


@app.get("/")
async def main():
    return FileResponse(some_file_path)
Copy the code

6 BaseSettings configuration file read

In the old days of writing Flask, we would write some configuration information into a configuration file, and then read the relevant configuration values according to the test environment or the formal environment.

Define the location file Congfig. Py

from functools import lru_cache import secrets from pydantic import AnyHttpUrl, BaseSettings import pprint pp = pprint.PrettyPrinter(indent=4) class Settings(BaseSettings): API_V1_STR: str = "/api/v1" SECRET_KEY: str = secrets.token_urlsafe(32) JWT_SECRET_KEY: str = '' JWT_ALGORITHM: str = '' JWT_EXPIRES: int = 60 JWT_REFRESH_EXPIRES_DAYS: int = 1 class Config: env_file = ".env" case_sensitive = True env_file_encoding = 'utf-8' @lru_cache() def get_settings(): return Settings() print("config.py > settings : ..." ) settings = Settings() pp.pprint(settings.dict())Copy the code

Run output:

config.py > settings : ...
{   'API_V1_STR': '/api/v1',
    'JWT_ALGORITHM': '',
    'JWT_EXPIRES': 60,
    'JWT_REFRESH_EXPIRES_DAYS': 1,
    'JWT_SECRET_KEY': '',
    'SECRET_KEY': 'pGR3SgBuuMq-nbmjRTJHh-px4huaPpTpgQwnM5MDcik'}
Copy the code

And then when we initialize our App we’ll introduce our single column object above, Settings

import uvicorn from fastapi import BackgroundTasks, FastAPI, Depends, Body, Query from fastapi.responses import JSONResponse,ORJSONResponse,UJSONResponse,PlainTextResponse,HTMLResponse,StreamingResponse,RedirectResponse,FileResponse ,Response from fastapi.exceptions import RequestValidationError from fastapi import FastAPI, File, UploadFile from config import Settings app = FastAPI() UploadFile from config import Settings app = FastAPI() version=settings.API_V1_STR )Copy the code

7 Fastapi Background Task supplement (Coroutine mode)

Originally wanted to add to the last to go, but the word count exceeded! Here’s another way to add background tasks (non-threaded) :

7.1 Background Tasks in Threading Mode

import uvicorn from fastapi import FastAPI, File, UploadFile app = FastAPI() from fastapi import BackgroundTasks, FastAPI app = FastAPI() def tasks(msg): Print (' run ', MSG) import time time.sleep(5) print(' run ', msg) @app.post("/backgroundTasks") async def runBackgroundTasks(msg: str, background_tasks: BackgroundTasks): Background_tasks. add_task(tasks, MSG) return "The task is being processed!" if __name__ == '__main__': # equal to the uvicorn command line uvicorn script name :app object start service: Uvicorn XXX :app --reload uvicorn. Run ('frun:app', host="127.0.0.1", port=8000, debug=True, reload=True)Copy the code

7.2 Coroutine Mode – Background Task

PS: the way of coroutine background, if your asynchronous task is super time-consuming, it is not recommended to use this method! Comparisons are single threaded!

import uvicorn from fastapi import FastAPI, Request import asyncio import threading app = FastAPI() from fastapi.openapi.docs import get_swagger_ui_html import time Async def write_log(): print(threading.current_thread().getname ()) Async def write_log_2(): Print (threading.current_thread().getName()) @app.get("/syncio/task") async def do_task(): Asyncio.create_task (write_log()) return "Hello, synchronization block under coroutine!" @app.get("/asyncio/task") async def do_task(): asyncio.create_task(write_log_2()) return "Hello, async blocking under coroutine!" Request: request @app.get("/request/ MSG ") def read_request(item_id: STR, request: request): scope = request.scope import json print(scope) client = request.client print(client) return get_swagger_ui_html() if __name__ = = "__main__" : uvicorn. Run (' main: app, host = "127.0.0.1", the port = 8000, reload = True)Copy the code

8 Extract client request information from Fastapi request

Firstly, when I used flask before, the request extraction corresponding to flask contained a lot of relevant information, and all the parameter sources of our requests were directly extracted through request. In fact, our FastAPI can also extract relevant parameter information. Even if you want to introduce other validators into FastAPI, you usually need to fetch the data from Request ourselves.

Let’s see what data our FastAPI Request gets:

PS is different from flask request, fastAPI request needs to be passed by our routing process! Cannot exist independently of routing functions.

Request source code:

HTTPConnection source (omitting some initialization functions and other magic functions) :

As you can see from the above source code, our Fastapi request can extract a lot of information:

• The following information can be extracted from request.scope:

- 'type' : 'HTTP' - 'asgi: {' version' : '3.0', 'spec_version' : '2.1'}, - 'http_version' : '1.1' - 'server' : (' 127.0.0.1, 8000) - 'client' : (' 127.0.0.1, 50863), - 'scheme' : 'HTTP' - 'method' : 'GET' - 'root_path' : '', - 'path': '/request/msg', - 'raw_path': b'/request/msg', - 'query_string': b'item_id=55555555', - 'headers': [- (b 'host', b '127.0.0.1:8000'), and (b 'connection', b 'keep alive -'), and (b 'accept', b 'application/json). - (b'user-agent', B 'Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, Like Gecko) Chrome/70.0.3538.25 Safari/537.36 Core/1.70.3870.400 QQBrowser/10.8.4405.400'), - (b'referer', B 'http://127.0.0.1:8000/docs') - (b' accept - encoding ', b 'gzip, deflate, br'), - (b 'accept - language, B 'useful - CN, useful; q = 0.9')] - 'fastapi_astack: < contextlib. AsyncExitStack object at 0 x000001be71fa19b0 >,' app ': <fastapi.applications.FastAPI object at 0x000001BE71F91D68>, - 'router': <fastapi.routing.APIRouter object at 0x000001BE71FA1320>, - 'endpoint': <function read_request at 0x000001BE71F76A60>, - 'path_params': {}Copy the code

• Type: indicates whether the request protocol is HTTP or HTTPS
• Asgi: Version information of the ASGI Web service being used
• Http_version HTTP version used
• Server: IP address and port of the request server (required for TCP quad objects)
• Client: IP address and port from which the client originated (a requirement for TCP quad objects)
• Scheme: protocol scheme used by the client
• Method: indicates the client request submission method
• Root_path: indicates the directory
• Path: indicates the path address of the client
• Raw_path: address of the client’s path (bytes)
• Headers: A list of information about the request headers from which the client requests
  • request.headers.get(“Authorization”)
• Query_string: query parameters submitted by the client
• Path_params: the path parameter submitted by the client request
• Url: indicates the URL of the client
  • request.url.path
  • request.url.port
  • request.url.scheme
• base_url
• The body client can use request.body() if it calls byte data
• If the data submitted by the JSON client is in JSON format, it can be directly converted into field data
• The form data submitted by the form client
• Cookies Indicates the cookies submitted by the client
• auth
• session
• client

conclusion

This is the basis of only the second, purely personal study notes, if there is a mistake also hope to give more advice.

In the future, I plan to conduct special studies on security and database. See you next time!

END

Let students | article | welcome learning exchange together 】 【 original 】 【 QQ: 308711822