Since Fastapi may have been updated to the latest version, I don’t know if some of the descriptions on the previous nuggets are still useful, so HERE, I also take the trouble to record again. And the previous notes, some did not practice, this time to supplement!

PS: I originally wanted to wait for my own public number to finish all the concurrent related knowledge points, and then comb this, in view of the follow-up public number still need some time to sort out, so temporarily ahead of time to talk about this framework.

Original address: juejin.cn/post/684490…

Lead based

1 Fastapi installation

PIP installation:

PIP install fastAPI [all] PIP install uvicorn or PIP install fastAPI PIP install uvicornCopy the code

Or use the IDE installation directly

2 Cheapest Fastapi example

1) Write the code of frun.py module:

from fastapi import FastAPI import uvicorn app = FastAPI() @app.get('/') async def login(): Return 'uncle' @ app. Get ('/' s) async def index () : return 'hello' 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

Description of startup parameters:

• App object launched: requires module name +app object name
• Host Boot address
• Post the port number
• Debug Whether to enable debugging
• Reload Indicates whether hot update is enabled

2) Directly right-click to start the service:

INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [11708] using statreload
INFO:     Started server process [9140]
INFO:     Waiting for application startup.
INFO:     Application startup complete.

Copy the code

3) Then visit: http://127.0.0.1:8000 interface

4) open interface document: access to the default address: http://127.0.0.1:8000/docs#/

The framework foundation is used in detail

1 Multi-route mode

Multiple decorators decorate a function together

Example:

@app.post("/") @app.put("/") @app.delete("/") @app.get("/") @app.options("/") @app.head("/") @app.patch("/") @app.trace("/") async def login(): returnCopy the code

>PS: it doesn’t support methods for fixing like Flask does

Example:

@app.route('/', methods=['GET','POST'], name='asdasd', include_in_schema=True) async def loginjig(): returnCopy the code

This method will raise the parameter exception, say I am missing parameters! But I have specified according to the rules and still do not support.

TypeError: loginjig() takes 0 positional arguments but 1 was given
Copy the code

@app.route(‘/a’, methods=[‘GET’,’POST’], name=’a’); And the value our function returns must be a callable object.

The following example is correct:

@app.route('/a', methods=['GET','POST'], name='a') def loginjig(req:Request): print(req) return PlainTextResponseCopy the code

PS: However, the disadvantage of the route definition above is that the interface defined using the above form cannot be shown in the document.

2 Route Classification

• Static route: fixed request URL
• Dynamic routing: Variable request URLS can be customized

For example, the requested IP address is:

• http://127.0.0.1:8000/user/login

@app.get('/user/login') async def login(): returnCopy the code

For example, the requested address is:

• http://127.0.0.1:8000/user/loginbyuserid/6

@app.get('/user/ loginByUserId /{userid}') async def loginByUserId (userid:int): returnCopy the code

PS: the value of the input parameter must be an int, otherwise it cannot be passed.

• Wrong situation

• Normal condition

On the parameter verification related to the subsequent verification of the recitation.

3 Matching priority of the route URL with the same name

In the requested route, if the same routing address happens to exist, and one static, one dynamic, then what is its priority?

Here is an example:

from fastapi import FastAPI app = FastAPI() @app.get("/users/me") async def read_user_me(): return {"user_id": "The current user"} @app.get("/users/{user_id}") async def read_user(user_id: STR): return {" priority matched to: ": user_id}"Copy the code

The document interface

The requested address is:http://127.0.0.1:8000/users/me

Request the address is: http://127.0.0.1:8000/users/tytr

The following is an example of changing the routing order:

from fastapi import FastAPI app = FastAPI() @app.get("/users/{user_id}") async def read_user(user_id: str): Get ("/users/me") async def read_user_me(): return {"user_id": "the current user"}Copy the code

Request the address is: http://127.0.0.1:8000/users/me

Request the address is: http://127.0.0.1:8000/users/tytr

Note the same URL situation, the priority of route registration is very key!

4 Route parameter submission classification

4.1 Parameter types of the ROUTE URL

• Query Indicates the type of the Query parameter
• Path Path parameter type (dynamic Path parameter)
• Compound Query parameters (Path Path parameters +Query Query parameters)

4.2 Example of Query Parameter Types

Query parameters are automatically added to the url &xxx=xxx&ksakka= YYYY.

from fastapi import FastAPI import uvicorn app = FastAPI() from fastapi import FastAPI app = FastAPI() fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}] @app.get("/items/") async def read_item(skip: int = 0, limit: int = 10): return fake_items_db[skip: skip + limit] 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

Request the address

http://127.0.0.1:8000/items/?skip=0&limit=10
Copy the code

The document interface

Description:

• Skip: int = 0, limit: int = 10 Skip =0&limit=10 is appended to our request URL

• Skip: int = 0, limit: int = 10

Pass a non-int parameter

4.2 Compound Query Parameters (combination of Path and Query Query parameters)

The URL contains dynamic parameters, as well as parameters that need to be submitted by the & delimiter. This situation is also common in GET submission.

from fastapi import FastAPI app = FastAPI() @app.get("/users/{user_id}/items/{item_id}") async def read_user_item( user_id: int, item_id: str, q: str = None, short: bool = False ): item = {"item_id": item_id, "owner_id": user_id} if q: item.update({"q": q}) if not short: item.update( {"description": "This is an amazing item that has a long description"} ) return 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

Request the address

http://127.0.0.1:8000/users/123456/items/items_xinxiid/?q=assa&short=True
Copy the code

Request the address

http://127.0.0.1:8000/users/123456/items/items_xinxiid/?q=assa&short=False
Copy the code

4.3 Mandatory and Optional definition methods of Parameters

• The optional and mandatory parameters are mainly determined by whether to give default values
• If there is no default value, the parameter is mandatory and will be verified
• Parameters with default values are optional and will not be verified
• You can determine the type of the parameter by using Optional and set the default value

4.3.1 Verification examples of Mandatory Parameters

@app.get("/items/{item_id}")
async def read_user_item(item_id: str, needy: str):
    item = {"item_id": item_id, "needy": needy}
    return item
Copy the code

The underlying value in the above code does not have a default value, and an error will be displayed when this value is not submitted:

4.3.2 Use Optional to define the data types to be submitted

from typing import Optional

@app.get("/items/{item_id}")
async def read_user_item(item_id: str, limit: Optional[int] = None):
    item = {"item_id": item_id, "limit": limit}
    return item

Copy the code

Where parameters are defined:

• Item_id: STR is mandatory if there is no default value
• 栗 型 : STR without a default value is required

This is not a default value is mandatory parameters and our own definition of the function, a function if there is no default value, then call this function, it must be a parameter passed!

We set the query parameter limit to int, but it is optional and set to None:

4.3.2.1 Optional+Query and non-mandatory parameters

import uvicorn

from fastapi import FastAPI, Query,Body
from pydantic import BaseModel, Field
from typing import Optional

app = FastAPI()


@app.get("/items/{item_id}")
async def read_user_item(item_id: str, limit: Optional[str] = Query(None, min_length=3, max_length=50)):
    item = {"item_id": item_id, "limit": limit}
    return item
Copy the code

The above example can be extended to further define type + limit for our limit

4.3.2.2 Optional+Query And this parameter is mandatory

: About Optional and Query together and our parameter is mandatory:

SQL > alter table Query; SQL > alter table Query; As the first parameter

import uvicorn from fastapi import FastAPI, Query,Body from pydantic import BaseModel, Field from typing import Optional app = FastAPI() @app.get("/items/{item_id}") async def read_user_item(item_id: str, limit: Optional[str] = Query(... , min_length=3, max_length=50)): As the first argument item = {"item_id": item_id, "limit": limit} return itemCopy the code

The examples above are usually unnecessary for us to use the ones below

limit: Optional[str] = Query(... , min_length=3, max_length=50)Copy the code

Method to define mandatory parameters, and need more verification, can be simplified as:

limit: str = Query(... , min_length=3, max_length=50)Copy the code

4.4 Path Parameters with keys such as /

If there is a parameter value in our URL that needs to be passed to the path that needs to be carried, it will conflict with the ‘/’ in our URL. How to avoid this conflict?

Such as:

import uvicorn
from fastapi import FastAPI

app = FastAPI()

@app.get('/files/path/{filepath}')
def getfiles(filepath):

    return filepath


Copy the code

The requested address is:http://127.0.0.1:8000/files/path/%2Fyyytyu%2Fretr

Modify the code to add path to the participation:

@app.get('/files/path/{filepath:path}')
def getfiles(filepath):

    return filepath

Copy the code

Request the address is: http://127.0.0.1:8000/files/path/%2Fyyytyu%2Fretr

4.5 Enumeration of path parameters

An enumeration of path parameters defines an enumeration class that provides the default selection of inputs:

As the sample:

import uvicorn from fastapi import FastAPI from enum import Enum class ModelName(str, Enum): alexnet = "alexnet" resnet = "resnet" lenet = "lenet" app = FastAPI() @app.get("/model/{model_name}") async def get_model(model_name: ModelName): if model_name == ModelName.alexnet: return {"model_name": model_name, "message": "Deep Learning FTW!" } if model_name.value == "lenet": return {"model_name": model_name, "message": "LeCNN all the images"} return {"model_name": model_name, "message": "Have some residuals"}Copy the code

Sample documents

4.6 Introduce Query encapsulated in Fastapi to verify Query parameters

4.6.1 Single Parameter Value with the same name

Fastapi defines a Query class for our roadbed parameters to specify the details of parameter verification. The following is an example:

import uvicorn from fastapi import FastAPI, Query app = FastAPI() @app.get("/items/") async def read_items(q: str = Query(None, min_length=3, max_length=50), regex="^fixedquery$"): results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]} if q: results.update({"q": Q}) return results@app. get("/items2/") async def read_items2(xiaozhong: STR =' I ',,q: str = Query(None, min_length=3, max_length=50), regex="^fixedquery$"): results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]} if q: results.update({"q": q}) return results @app.get("/items3/") async def read_items3(xiaozhong:bool=False,,q: str = Query(None, min_length=3, max_length=50), regex="^fixedquery$"): results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]} if q: results.update({"q": q}) return results 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

The above parameters:

q:str = Query(None, min_length=3,max_length=50),regex="^fixedquery$")
Copy the code

It means:

• The q parameter is optional, but if entered, the maximum length must be less than 50, and the minimum length must be greater than 3: it must match the REgex

Of course None can be changed to another default value, such as:

q: q: str = Query('xiaozhong', min_length=3,max_length=50),regex="^fixedquery$")
Copy the code

Several validations for parameters:

Without passing q:

http://127.0.0.1:8000/items/

When q is transmitted and the length is greater than 50:

http://127.0.0.1:8000/items/

When q is transmitted and the length is less than 3:

http://127.0.0.1:8000/items/?q=4

Parameter regularization verification of Query parameter Query

4.6.2 Multiple Query Parameters with the same name -Query Parameter multi-value list

In general, it is rarely said in our interface to submit multiple values for the same parameter, such as:

http://localhost:8000/items/?q=foo&q=bar

However, we do not check for the existence of this condition, so we can also define our parameters like must be a list form:

from typing import List

from fastapi import FastAPI, Query

app = FastAPI()


@app.get("/items/")
async def read_items(q: List[str] = Query(["foo"."bar"])) :
    # <! List[STR]:-->
    query_items = {"q": q}
    return query_items

Copy the code

Default value:Non-default values:

4.7 Introduce Patht encapsulation in Fastapi to verify path parameters

Query can be used for Query parameters, and Path parameters can also be verified using the Path provided by Fastapi.

from fastapi import FastAPI, Path

app = FastAPI()


@app.get("/items/{item_id}")
async def read_items(
    q: str, item_id: int = Path(. , title="The ID of the item to get")
) :
    results = {"item_id": item_id}
    if q:
        results.update({"q": q})
    return results
Copy the code

Item_id can also be greater than or equal to item_id.

from fastapi import FastAPI, Path

app = FastAPI()

@app.get("/items/{item_id}")
async def read_items(
 *, item_id: int = Path(. , title="The ID of the item to get", ge=1), q: str ) :
    results = {"item_id": item_id}
    if q:
        results.update({"q": q})
    return results
Copy the code

Item_id must be the integer “g is greater than or equal to e is equal to 1” when ge = 1.

4.7 Verification of binding model under POST or PUT

The request body mainly involves the problem of how to correspond to the Basemodel, facilitate the model to parse, and carry out relevant serialization and anti-sequence operations.

PS usually uses the request header for non-form submissions:

Application/jsonCopy the code

4.7.1 Verification of Static Routes and Simple Request Bodies

For arguments submitted using POST and PUT, the usual request body is displayed on postman:

Generally, the Request Body is not submitted through GET. Parameters submitted by GET are called query parameters. Therefore, if the parameter information is submitted by POTS,PUT, etc., it will be submitted to the back end of the Request Body.

FastApi provides a form for how to receive and validate the request body using:

from pydantic import BaseModel
Copy the code

The following is an example:

from fastapi import FastAPI
from pydantic import BaseModel


class Item(BaseModel) :
    name: str
    description: str = None
    price: float
    tax: float = None


app = FastAPI()


@app.post("/items/")
async def create_item(item: Item) :
    return item
Copy the code

In the model above, I define what format it must be if an Item is submitted, for example

• Name is a mandatory field
• Description is optional and defaults to None
• Price is mandatory and needs to be of type float
• Tax is optional and defaults to None

How does the client submit the above parameters? Try submitting parameters without writing anything in case:

When submitting parameters in JSON format:

Intentionally submitting incorrect parameter format requests:

4.7.2 VERIFICATION of URL hybrid + simple request body

The combination of Request Body and Query and Path is inevitable in the design of some apis and may require some of the above mashup combinations, requiring simultaneous submission and retrieval of multiple parameters

So how do we normally receive this parameter? Example code is as follows:

from fastapi import FastAPI, Path
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel) :
    name: str
    description: str = None
    price: float
    tax: float = None


@app.put("/items/{item_id}")
async def update_item(
    *,
    item_id: int = Path(. , title="The ID of the item to get", ge=0, le=1000),
    q: str = None,
    item: Item = None.) :
    results = {"item_id": item_id}
    if q:
        results.update({"q": q})
    if item:
        results.update({"item": item})
    return results
Copy the code

Through the previous learning, it is actually very simple and the principle is the same, the above example request:

Where the above parameters:

• Q is the query type parameter
• Item_id is a pathtype parameter
• Item is the request body parameter

4.7.3 Verification of complex combination type request body

4.7.3.1 Single structure

In fact, there will be multi-body Boay submission for more complex business. In the mall order done before, the client may submit object information of multiple entities to the back end, such as order entity, address entity, commodity information entity, etc.

How do you accept multiple Body entities in Fastapi? Usually, in bottle, you can get the information submitted by the client directly by requesting. Body or request.json.

Fastapi assumes that the client submits parameters of the form:

{" item ": {" name" : "Foo", "description" : "The pretender", "price" : 42.0, "tax", 3.2}, "user" : {" username ": "dave", "full_name": "Dave Grohl" } }Copy the code

What about the reception processing?

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel) :
    name: str
    description: str = None
    price: float
    tax: float = None


class User(BaseModel) :
    username: str
    full_name: str = None


@app.put("/items/{item_id}")
async def update_item(*, item_id: int, item: Item, user: User) :
    results = {"item_id": item_id, "item": item, "user": user}
    return results
Copy the code

In this case, the client submits multiple entity objects. You can define multiple model objects. Fastapi will automatically process the extracted information for you.

4.7.3.2 Single structure + Single type Body parameter

If another assumption is made:

Fastapi assumes that the client submits an argument of the form: a single type Body argument

{" item ": {" name" : "Foo", "description" : "The pretender", "price" : 42.0, "tax", 3.2}, "user" : {" username ": "dave", "full_name": "Dave Grohl" }, "importance": 5 }Copy the code

In fact, this may not exist, how to read the importance parameter parsing? Since the parameters are Query and Path, the Body should also be present.

from fastapi import Body, FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel) :
    name: str
    description: str = None
    price: float
    tax: float = None


class User(BaseModel) :
    username: str
    full_name: str = None


@app.put("/items/{item_id}")
async def update_item(
    *, item_id: int, item: Item, user: User, importance: int = Body(. ,gt=0)
) :
    results = {"item_id": item_id, "item": item, "user": user, "importance": importance}
    return results
Copy the code

In the code above we introduce Body and in importance: int = Body(…) To process and extract:

4.7.3.3 Nested Model structures

What if, in addition, the client submits a more complex nested model? Hemp egg affirmation also can have such circumstance drop! Nesting has lists and entities in it.

Such as:

{" name ":" Foo ", "description" : "The pretender", "price" : 42.0, "tax" : 3.2, "tags" : [" rock ", "metal", "bar"], "image" : { "url": "http://example.com/baz.jpg", "name": "The Foo live" } }Copy the code

At this point, we need what is called child embedding:

from typing import Set

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Image(BaseModel) :
    url: str
    name: str


class Item(BaseModel) :
    name: str
    description: str = None
    price: float
    tax: float = None
    tags: Set[str] = []
    image: Image = None


@app.put("/items/{item_id}")
async def update_item(*, item_id: int, item: Item) :
    results = {"item_id": item_id, "item": item}
    return results

Copy the code

So in the code above, Item contains an Image, and it also contains a list definition of the tags type.

4.7.3.4 More complex nested model structures

Deeper nesting of MMP can also be defined as follows:

{ "name":"Foo", "description":"The pretender", "price":42, "items":[ { "name":"Foo", "description":"The pretender", "Price" : 42, "tax" : 3.2, "tags" : [" rock ", "metal", "bar"], "image" : {" url ":" http://example.com/baz.jpg ", "Name" : "The Foo live"}}, {" name ":" Foo2 ", "description" : "" The" 2 ", "price" : 422, "tax" : 3.2, "tags" : [" rock ", "metal", "bar" ], "image":{ "url":"http://example.com/baz.jpg", "name":"The Foo live" } } ] }Copy the code

The corresponding analysis is:

from typing import Set
from typing import List.Set

class Image(BaseModel) :
    url: str
    name: str
class Item(BaseModel) :
    name: str
    description: str = None
    price: float
    tax: float = None
    tags: Set[str] = []
    # images: List[Image] = None
    image: Image = None


class Offer(BaseModel) :
    name: str
    description: str = None
    price: float
    items: List[Item]


@app.post("/offers/")
async def create_offer(*, offer: Offer) :
    return offer
Copy the code

4.7.3.5 Field class for Request Body

The Field Field is the same as the Query, Path Field, and the Body Field is the same as the Body Field.

In other words. Field is used to standardize submitted Body parameter information.

Such as:

from fastapi import Body, FastAPI
from pydantic import BaseModel, Field

app = FastAPI()


class Item(BaseModel) :
    name: str
    description: str = Field(None, title="The title.",description="Error message text.", max_length=300)
    price: float= Field(... , gt=0, description="Error message text.")
    tax: float = None


@app.put("/items/{item_id}")
async def update_item(*, item_id: int, item: Item = Body(. , embed=True)) :
    results = {"item_id": item_id, "item": item}
    return results

Copy the code

The above meaning is the same as before defined parameter verification

Normal condition:Abnormal conditions:

The above documents show the specific definitions and related parameter descriptions:

For titles and descriptions of Field definitions, online Schemas:

Note: price: float = Field(… Gt =0, description=” error “)… The field marked as current is the required field

As follows: Notice the verification prompt marked with *

4.8 Submit the Form for verification

If your design front-end submits the form, you need to use the form to retrieve parameters. Fastapi handling of forms, if you are not a bucket setup, usually requires an additional setup of this processing to handle form submissions.

pip install python-multipart
Copy the code

Examples are as follows:

from fastapi import FastAPI, Form app = FastAPI() @app.post("/login/") async def login(username: str = Form(...) , password: str = Form(...) ): return {"username": username}Copy the code

Note that the request header submission should use:

Application/x - WWW - form - urlencoded wayCopy the code

4.9 Convert the input model into a dictionary after verification

If necessary, we need to convert our parameters in the output model to those of the transformation, Fastapi comes with a transformation handling:

Example:

import uvicorn from fastapi import FastAPI from pydantic import BaseModel from fastapi.encoders import jsonable_encoder app = FastAPI() class Item(BaseModel): name: str description: str = None price: float tax: float = None class User(BaseModel): username: str full_name: str = None @app.put("/items") async def update_item(item: Item): json_compatible_item_data = jsonable_encoder(item) print(json_compatible_item_data) return json_compatible_item_data 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

Print out L

{' name ':' string ', 'description' : 'string', 'price: 0.0,' tax: 0.0}Copy the code

4.10 Verification of request bodies of other types of parameters

Other Data Types ¶ Here are some other data types you can use (from the official documentation) :

• UUID:
  • A standard “universal unique identifier” commonly found in ids in many databases and systems.
  • In request and reply, will be represented as STR.
• datetime.datetime:
  • A Pythondatetime. Datetime.
  • In the request and reply, STR is represented in ISO 8601 format, such as 2008-09-15T15:53:00+05:00.
• datetime.date:
  • Pythondatetime.date.
  • In requests and replies, STR is represented in ISO 8601 format, such as 2008-09-15.
• datetime.time:
  • A Pythondatetime. Time.
  • In requests and replies, STR is represented in ISO 8601 format, such as 14:23:55.003.
• datetime.timedelta:
  • A Pythondatetime timedelta.
  • In request and reply, is represented as the total number of float seconds.
  • Pydantic also allows it to be expressed as “ISO 8601 time difference encoding”, see the documentation for more information. .
• frozenset:
  • In requests and replies, think of it as a set:
  • In the request, the list is read, the duplication is eliminated, and it is converted to a set.
  • In a reply, the set is converted to a list.
  • The generated schema will specify that the set value is unique (uniqueItems using JSONSchema).
• bytes:
  • Standard Pythonbytes.
  • The request and reply will be treated as STR.
  • The generated schema will specify that it is STR with binary “format”.
• Decimal:
  • Standard PythonDecimal.
  • In request and response, the handling is the same as float.

So I can also use other types of validation:

from datetime import datetime, time, timedelta
from uuid import UUID

from fastapi import Body, FastAPI

app = FastAPI()


@app.put("/items/{item_id}")
async def read_items(
    item_id: UUID,
    start_datetime: datetime = Body(None),
    end_datetime: datetime = Body(None),
    repeat_at: time = Body(None),
    process_after: timedelta = Body(None),
) :
    start_process = start_datetime + process_after
    duration = end_datetime - start_process
    return {
        "item_id": item_id,
        "start_datetime": start_datetime,
        "end_datetime": end_datetime,
        "repeat_at": repeat_at,
        "process_after": process_after,
        "start_process": start_process,
        "duration": duration,
    }

Copy the code

5 Operation of Headr

5.1 Obtaining request Headers

We mentioned in the previous form that we need to submit a special request type, but if there are some special requirements, our project needs to get the relevant parameters from the request header submission, such as some custom request headers, such as toekn, how to deal with that?

Fastapi also provides direct encapsulation for us, such as Query, Path, and the same class, import Header can be used.

Example:

import uvicorn from fastapi import FastAPI,Header from pydantic import BaseModel, Field from typing import Optional app = FastAPI() @app.get("/Header") async def Header_handel(ua: Optional[STR] = Header(None, convert_underscores=True) : # convert_underscores Auth_token === return UA if __name__ == '__main__' if auth_token === is converted to auth-toekn: # 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

Run example:

PS: When fetching the request header, we need to pay attention to the problem of writing the request header with a slash.

Before I Flask framework, encountered a custom request body is normal in other browsers, but in UC browser, it is blocked! Can’t get it, because in Python, a parameter such as user-agent is not considered a valid variable name. So in order to get the value of the request Header properly, the Header module converts the “_” in the parameter name to “-“. So it’s a slash conversion problem. The convert_underscores parameter in the examples above means whether to convert if the parameters in the request header are underlined

Auth_token == the assumptions that accompany auth_token convert_underscores =True and FalseCopy the code

Request headers with the same name as query parameters with multiple values are processed in the same way as query parameters with multiple values are processed in the same way as query parameters with multiple values are processed in the same way as query parameters with multiple values.

The following is an example:

import uvicorn from fastapi import FastAPI,Header,Cookie,Response from pydantic import BaseModel, Field from typing import List app = FastAPI() @app.get("/headerlist/") async def read_headerlist(x_token: List[str] = Header(None)): return {"X-Token values": x_token} 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

5.2 Setting the Response Header

There are request headers and there must be response headers:

How do we set up our custom response headers? Before doing Flask cross-domain processing, in fact, we try to add the response headers in the response body to support cross-domain request. So it’s important to learn how to write custom response headers.

For how do I write? According to the official document, show us two solutions. Scenario 1: Use Response

import uvicorn from fastapi import FastAPI, Response app = FastAPI() @app.get("/set_response_headers") def set_response_headers(response: Response): response.headers["X-Token"] = "888888888888888888888888888" response.headers["Access-Control-Allow-Methods"] = " GET, OPTIONS, HEAD, PUT, POST" return {"message": "OK"} 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

Scenario 2: Use JSONResponse

import uvicorn from fastapi import FastAPI, Response from fastapi.responses import JSONResponse app = FastAPI() @app.get("/set_response_headers") def set_response_headers(): content = {"message": "OK"} headers = { "X-Token": "888888888888888888888888888", "Access-Control-Allow-Methods": " GET, OPTIONS, HEAD, PUT, POST" } return JSONResponse(content=content, headers=headers) 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

6 Cookie operations

6.1 Cookie Operations – Write

For cookies that rely on front-end submission, the Cookie value written to us needs to be processed in the response message. Because this involves the processing of response messages, if we use the model to output directly, we usually need to encapsulate the processing.

For how do I write? According to the official document, show us two solutions.

Scenario 1: Use Response

import uvicorn from fastapi import FastAPI,Header,Cookie,Response from pydantic import BaseModel, Field from typing import Optional app = FastAPI() @app.get("/Cookie") async def Cookier_handel(cookie: Optional[str] = Cookie(None)): return cookie @app.get("/set_cookie/") def setcookie(response: Response): response.set_cookie(key="xiaozhong", Value = "sdasdasdaxxxxxxxxxxxxxxxxxxxda") return 'setting cookies success 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

Scenario 2: Use JSONResponse

import uvicorn from fastapi import FastAPI,Header,Cookie,Response from pydantic import BaseModel, Field from typing import Optional from fastapi.responses import JSONResponse app = FastAPI() @app.get("/Cookie") async def Cookier_handel(cookie: Optional[str] = Cookie(None)): return cookie @app.get("/set_cookie/") def setcookie(response: Response): response.set_cookie(key="xiaozhong", Value = "sdasdasdaxxxxxxxxxxxxxxxxxxxda") return 'setting cookies success' @ app. Getcookiebyjson def ("/getcookiebyjson/") setcookieyjson(): content = {"msg": } response = JSONResponse(content=content) response.set_cookie(key="xiaozhong", value="sdasdasdaxxxxxxxxxxxxxxxxxxxda") return response 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

6.2 Cookie Operations – Obtain

Usually when our front-end request is initiated, it usually submits the Cookie value of the front-end to do relevant verification. In fact, Fastapi also directly provides a good encapsulation for us, similar to Query, Path and the same class, import Cookie can be used, and obtain the request header the way.

Our cookies are actually submitted at the top of the request:

Example:

import uvicorn from fastapi import FastAPI,Header,Cookie from typing import Optional app = FastAPI() @app.get("/Cookie")  async def Cookier_handel(cookie: Optional[str] = Cookie(None)): return cookie 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

Use models to style responses

7.1 Use the response_model definition

When an interface is requested to return something visible to our client, it is called a response message, such as the response header, response code, response content, etc.

Users who are not usually that stupid return whatever they type. The following official website example is a pure demonstration:

from fastapi import FastAPI
from pydantic import BaseModel, EmailStr

app = FastAPI()



class UserIn(BaseModel) :
    username: str
    password: str
    email: str
    full_name: str = None


class UserOut(BaseModel) :
    username: str
    email: str
    full_name: str = None


@app.post("/user/", response_model=UserOut)
async def create_user(*, user: UserIn) :
    return user

Copy the code

After the request, we get the content information as UserOut:

Response_model is the response_model of our response, but typically if we’re doing API development, unless you have a particular interface, we’re going to return our response in the API format rather than directly output a specific model, Of course, if your model of your output is in your JSON format, this is also ok

class UserOut(BaseModel):
    code: str
    message: str
    data: list = None
Copy the code

Or we can return response_model as a dictionary:

from typing import Dict

from fastapi import FastAPI

app = FastAPI()


@app.get("/keyword-weights/", response_model=Dict[str.float])
async def read_keyword_weights() :
    return {"foo": 2.3."bar": 3.4}
Copy the code

7.2 include (exclude the rest) and exclude in response_model

• Response_model_exclude: if our response_model output has partial fields to filter
• Response_model_include: If the response_model of our output must be pure in a field, use it to include.
• Response_model_exclude_unset: does not return unset parameters (including if 0)
• Response_model_exclude_defaults: does not return fields that are default values
• Response_model_exclude_none: return response_model_exclude_none=True

from fastapi import FastAPI from pydantic import BaseModel, EmailStr import uvicorn app = FastAPI() class UserIn(BaseModel): username: str password: str email: str full_name: str = None class UserOut(BaseModel): username: str email: str full_name: STR = None # response@app. post("/user1/", response_model=UserOut, response_model_include={"username", "full_name"}) async def create_user1(*, user: UserIn): Response_model =UserOut, response_model_exclude={"email", }) async def create_user2(*, user: UserIn): return user 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.3 About the response status code status_code

Normally an interface request completes and returns 200 if there are no exceptions: as if the log is printed:

INFO:     127.0.0.1:58141 - "POST /user/ HTTP/1.1" 400
INFO:     127.0.0.1:58315 - "POST /user/ HTTP/1.1" 200
Copy the code

FastAPI runs our specified return status_code

7.3.1 – plan 1:

The following is an example:

@app.post("/user/", response_model=UserOut,status_code=500)
async def create_user(*, user: UserIn) :
    return user
Copy the code

Causes the requested interface to return:

6.3.2 – scheme 2:

You can even specify this by importing status:

from fastapi import FastAPI, status

app = FastAPI()


@app.post("/items/", status_code=status.HTTP_201_CREATED)
async def create_item(name: str) :
    return {"name": name}
    
Copy the code

7.3.3 – solution 3:

Can also be in our

@app.get("/set_response_headers")
def set_response_headers(response: Response):
    response.headers["X-Token"] = "888888888888888888888888888"
    response.headers["Access-Control-Allow-Methods"] = " GET, OPTIONS, HEAD, 
    PUT, POST"
    response.status_code = status.HTTP_201_CREATED
    return {"message": "OK"}
Copy the code

Sections 7.3.4 – solution 4:

Can also be in our

@app.get("/set_response_headers")
def set_response_headers():
    content = {"message": "OK"}

    headers = {
        "X-Token": "888888888888888888888888888",
        "Access-Control-Allow-Methods": " GET, OPTIONS, HEAD, PUT, POST"
    }

    return JSONResponse(status_code=status.HTTP_201_CREATED, content=content, headers=headers)
Copy the code

8 Troubleshooting exceptions

8.1 HTTPException Exception Thrown

There’s actually an HttpError exception class in Bottle, and there’s also an HTTPException in FastAPI.

As the sample:

from fastapi import FastAPI, HTTPException

app = FastAPI()

items = {"foo": "The Foo Wrestlers"}


@app.get("/items/{item_id}")
async def read_item(item_id: str) :
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"item": items[item_id]}
Copy the code

In the above code, a 404 error is actively thrown by checking if item_id exists in items

HTTPException and StarletteHTTPException are both inherited and Exception:

class HTTPException(StarletteHTTPException):
    def __init__(
        self, status_code: int, detail: Any = None, headers: dict = None
    ) -> None:
        super().__init__(status_code=status_code, detail=detail)
        self.headers = headers
Copy the code

Therefore, we can usually use raise directly to raise an exception.

8.2 HTTPException and return the new custom request header

from fastapi import FastAPI, HTTPException

app = FastAPI()

items = {"foo": "The Foo Wrestlers"}


@app.get("/items-header/{item_id}")
async def read_item_header(item_id: str) :
    if item_id not in items:
        raise HTTPException(
            status_code=404,
            detail="Item not found",
            headers={"X-Error": "There goes my error"},)return {"item": items[item_id]}
Copy the code

8.3 Customizing an HTTPException is Returned

Similar to the Bottle before, we add a custom global error to unify the processing return. FastAPI also provides a mechanism for customizing errors:

The official example is as follows:

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse


class UnicornException(Exception) :
    def __init__(self, name: str) :
        self.name = name


app = FastAPI()


@app.exception_handler(UnicornException)
async def unicorn_exception_handler(request: Request, exc: UnicornException) :
    return JSONResponse(
        status_code=418,
        content={"message": f"Oops! {exc.name} did something. There goes a rainbow..."},)@app.get("/unicorns/{name}")
async def read_unicorn(name: str) :
    if name == "yolo":
        raise UnicornException(name=name)
    return {"unicorn_name": name}
    
Copy the code

Observe request results:

When name == YOLO was thrown, we threw a UnicornException, and we, @app.Exception_handler (UnicornException) caught the exception and returned it.

8.4 FastAPI HTTPException and Starlette HTTPException

• FastAPI HTTPException is inherited from Starlette’s HTTPException.
• HTTPException and StarletteHTTPException inherit with Exception:

FastAPI HTTPException and Starlette HTTPException:

• FastAPI HTTPException allows you to add headers to a Response. Mainly used internally for OAuth 2.0 and some security-related functions.
• Starlette HTTPException could not add header information

So, we usually throw FastAPI HTTPException in our code.

However, when we register the exception handler, we should register it as Starlette HTTPException.

This way, when Starlette’s internal code or a Starlette extension throws a Starlette HTTPException, our processor can normally catch and handle the exception.

8.5 Override FastAPI’s default exception handling

A RequestValidationError is raised when a request contains invalid data, or when a parameter submission error occurs.

We can override our RequestValidationError with a custom exception:

For example, the default code does not add override handling:

from fastapi import FastAPI, HTTPException
from fastapi.exceptions import RequestValidationError
from fastapi.responses import PlainTextResponse
from starlette.exceptions import HTTPException as StarletteHTTPException

app = FastAPI()


@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request, exc) :
    return PlainTextResponse(str(exc.detail), status_code=exc.status_code)


# @app.exception_handler(RequestValidationError)
# async def validation_exception_handler(request, exc):
# return JSONResponse({'mes':' RequestValidationError ') # return JSONResponse({'mes':' RequestValidationError ') '%(str(exc))})



@app.get("/items/{item_id}")
async def read_item(item_id: int) :
    if item_id == 3:
        raise HTTPException(status_code=418, detail="Nope! I don't like 3.")
    return {"item_id": item_id}



if __name__ == '__main__':
    import uvicorn
    uvicorn.run(app='main4:app', host="127.0.0.1", port=8000, reload=True, debug=True)

Copy the code

Return a request with an exception:

When restoring coverage:

from fastapi import FastAPI, HTTPException
from fastapi.exceptions import RequestValidationError
from fastapi.responses import PlainTextResponse
from starlette.exceptions import HTTPException as StarletteHTTPException

app = FastAPI()


@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request, exc) :
    return PlainTextResponse(str(exc.detail), status_code=exc.status_code)


@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc) :
    return JSONResponse({'mes':RequestValidationError :%s your sister's error! '% (str(exc))})


@app.get("/items/{item_id}")
async def read_item(item_id: int) :
    if item_id == 3:
        raise HTTPException(status_code=418, detail="Nope! I don't like 3.")
    return {"item_id": item_id}



if __name__ == '__main__':
    import uvicorn
    uvicorn.run(app='main4:app', host="127.0.0.1", port=8000, reload=True, debug=True)

Copy the code

Request result:

The response code can also be modified as follows:

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        content=jsonable_encoder({"detail": exc.errors(), "body": exc.body}),
    )
Copy the code

Description:

From fastAPI import Depends, status, HTTPException # from starlette.exceptions import HTTPExceptionCopy the code

8.6 Determining the RequestValidationError type

If it is necessary to use the inverse function type to match the related error type, then we need to determine the RequestValidationError type, compare this error type more.

Such as:

from fastapi import FastAPI, HTTPException,status,Response from fastapi.exceptions import RequestValidationError from pydantic.errors import * from pydantic import ValidationError class RequestValidationErrorException(): @staticmethod def excaction(exc=RequestValidationError) ->JSONResponse: Print (" selfself", exc.errors()) print(" selfself", Exc. Errors () [0]. Get (" loc ") # error if path parameters' path 'in exc. Errors () [0]. Get (" loc ") : If isinstance(exc. Raw_errors [0]. Exc,IntegerError): Return CustomizeParameterException (MSG = '% s % s type error parameters, must be an Integer type' % (exc.errors()[0].get('loc'),exc.errors()[0].get('loc')[1])) elif isinstance(exc.raw_errors[0].exc,MissingError): Return CustomizeParameterException (MSG = '% s % s is missing, parameters is than the parameters' % (exc.errors()[0].get('loc'),exc.errors()[0].get('loc')[1])) elif isinstance(exc.raw_errors[0].exc,NumberNotLeError): Return CustomizeParameterException (MSG = '% s % s has limit, parameters must be less than or equal to % s' % (exc.errors()[0].get('loc'),exc.errors()[0].get('loc')[1],exc.errors()[0].get('ctx').get('limit_value'))) elif isinstance(exc.raw_errors[0].exc, NumberNotLtError): Return CustomizeParameterException (MSG = '% s % s has limit, parameters must be less than % s' % (exc.errors()[0].get('loc'),exc.errors()[0].get('loc')[1], exc.errors()[0].get('ctx').get('limit_value'))) else: CustomizeParameterException (MSG = 'path parameter errors, please check the parameter format and requirements') # body model parameters calibration type of error elif' body 'in exc. Errors () [0]. Get (" loc ") : pass if isinstance(exc.raw_errors[0].exc, ValidationError): # return CustomizeParameterException (MSG = '% s type error path parameters, must be an Integer type' % (exc. Errors () [0]. Get (" loc ") [1])) return CustomizeParameterException()Copy the code

8.7 Inherit JSONResponse and return the respective response body

We return our response body packets in flask the way I’m used to.

from typing import Any, The error response returned Dict # custom body information from fastapi. Responses import PlainTextResponse, JSONResponse class ApiResponse (JSONResponse) : Http_status_code = 200 # default success code = 0 data = None # return {} or [] MSG = 'success' def __init__(self,http_status_code=None, data=None,msg=None, **options): if data: self.data = data if msg: self.msg = msg if http_status_code: Body = dict(MSG =self.msg, code=self.code, data=self.data, ) super(ApiResponse, self).__init__(status_code=self.http_status_code,content=body, **options) class BadrequestException(ApiResponse): Http_status_code = 400 # error_code = 10032 code = 10032 ParameterException = 'error request' class ApiResponse: Http_status_code = 400 code = 400 MSG = 'error' class UnauthorizedException(ApiResponse) Http_status_code = 401 code = 401 MSG = 'unlicensed' class ForbiddenException(ApiResponse): Http_status_code = 403 code = 403 MSG = 'no access' class NotfoundException(ApiResponse): Http_status_code = 404 code = 404 MSG = 'access address does not exist' class MethodnotallowedException (ApiResponse) : Http_status_code = 405 code = 405 MSG = 'Submitting access using this method is not supported' class OtherException(ApiResponse): Http_status_code = 800 code = 800 MSG = 'unknown other HTTPEOOER exception' error_code = 10034 class InternalErrorException(ApiResponse): http_status_code = 800 code = 800 MSG = 'unknown other HTTPEOOER exception' error_code = 10034 class InternalErrorException(ApiResponse): Http_status_code = 500 code = 500 data = None # {} or [] # MSG = 'Internal Server Error' MSG = 'service crash exception' class RateLimitApiException(ApiResponse): Http_status_code = 429 code = 429 data = None # result can be {} or [] # MSG = 'Internal Server Error' MSG = 'limited number of requests' class CustomizeApiResponse(ApiResponse): Http_status_code = 200 code = 200 data = None # Result can be {} or [] # MSG = 'Internal Server Error' MSG = 'success' class CustomizeParameterException (ApiResponse) : http_status_code = 200 code = 200 MSG = 'parameters calibration error error_code = 10031Copy the code

9: FastAPI middleware

9.1 Overview of middleware

The so-called middleware actually plays the same role as the middleware in bottle. Some methods or operations need to be performed before all routes, such as adding an HTTP access interceptor, validating some interfaces that require authorization from the interface API, and so on.

FastAPI provides @app.Middleware (” HTTP “) to do something similar. Much like the bottle or flask hook function

The following is an example:

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

import time
from fastapi import FastAPI, HTTPException
from fastapi.exceptions import RequestValidationError
from fastapi.responses import PlainTextResponse
from starlette.exceptions import HTTPException as StarletteHTTPException

app = FastAPI()


@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request, exc) :
    return PlainTextResponse(str(exc.detail), status_code=exc.status_code)


@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc) :
    return JSONResponse({'mes':RequestValidationError :%s your sister's error! '% (str(exc))})


@app.get("/items/{item_id}")
async def read_item(item_id: int) :

    return {"item_id": item_id}


@app.middleware("http")
async def add_process_time_header(request: Request, call_next) :
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response


if __name__ == '__main__':
    import uvicorn
    uvicorn.run(app='main4:app', host="127.0.0.1", port=8000, reload=True, debug=True)

Copy the code

When the request completes, we find that we have a new request header in our response header:

9.2 HTTPSRedirectMiddleware middleware

This middleware, HTTPSRedirectMiddleware, essentially forces us to request HTTPS or WSS. It means you need secure transmission.

Example:

from fastapi import FastAPI
from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware

app = FastAPI()

app.add_middleware(HTTPSRedirectMiddleware)


@app.get("/")
async def index():
    return {"message": "OK"}
Copy the code

9.3 TrustedHostMiddleware

RustedHostMiddleware forces us to send requests with a Host option in the request Header to avoid HTTP Host Header attacks.

from fastapi import FastAPI
from fastapi.middleware.trustedhost import TrustedHostMiddleware

app = FastAPI()

app.add_middleware(
    TrustedHostMiddleware, allowed_hosts=["example.com", "*.example.com"]
)


@app.get("/")
async def main():
    return {"message": "Hello World"}
Copy the code

9.3 GZipMiddleware middleware

Mainly play the role of resource compression.

When the accept-Encoding field in the request header is “gzip, “GZipMiddleware handles the return result.

GZipMiddleware supports the minimum_size parameter: compression is disabled when the result size is smaller than the specified value. (In bytes, default is 500)

from fastapi import FastAPI
from fastapi.middleware.gzip import GZipMiddleware

app = FastAPI()

app.add_middleware(GZipMiddleware, minimum_size=1000)

Copy the code

10: FastAPI middleware cross-domain processing

Why do we need cross-domain processing? Our API is usually called to the front end, but the front end may use a different domain name than the API domain name that is not provided, which causes the same origin policy problem of the browser, so we need to do cross-domain request support.

More technical: from developer.mozilla.org/en-US/docs/…

Cross-source Resource Sharing (CORS) Cross-source resource sharing (CORS) is an HTTP-header based mechanism that allows the server to indicate any other origin S(domain, scheme, or port) than the browser itself should allow to load the resource. CORS also relies on a mechanism by which the browser makes a “pre-fly” request to a server hosting a cross-source resource to check whether the server will allow the actual request. In this prerun, the browser sends a header indicating the HTTP method and header that will be used in the actual request.

Example of cross-source request: Domain-a.com request using XMLHttpRequest https://domain-b.c… .

For security reasons, browsers restrict cross-source HTTP requests initiated from scripts. For example,XMLHttpRequest takes the API following the same origin policy. This means that Web applications using these apis can only request resources from the same source loaded by the application, unless the response from the other source contains the correct CORS header.

The CORS mechanism supports secure cross-source requests and data transfers between browsers and servers. Modern browsers use CORS in apis such as XMLHttpRequest or fetch to reduce the risk of cross-source HTTP requests.

FastAPI supports cross-domain support by adding intermediate forms, similar to bottle. He also supports only which domains are supported for cross-domain requests:

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

origins = [
    "http://localhost.tiangolo.com"."https://localhost.tiangolo.com"."http://localhost"."http://localhost:8080",
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],)@app.get("/")
async def main() :
    return {"message": "Hello World"}
Copy the code

Too lazy to access a JS, so this is temporarily not tested, later have the opportunity to test and verify, the feeling should be like this.

11: FastAPI dependency Injection Depends

Depending on the description of Depends on the website, I seem to be a bit ignorant about it, so I still need to spend some time to learn about dependency injection again.

First, dependency injection can be a function or a class. Dependency injection in the form of a function is as follows:

11.1 Simple Dependency Description

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(q: str = None, skip: int = 0, limit: int = 100) :
    return {"q": q, "skip": skip, "limit": limit}


@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)) :
    commons.update({'let':'students'})
    return commons


@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)) :
    return commons

if __name__ == '__main__':
    import uvicorn
    uvicorn.run(app='main:app', host="127.0.0.1", port=8100, reload=True, debug=True)
Copy the code

Comb through the interface request flow:

• 1: Upper Commons: Dict = Depends(common_parameters) it declares a dependency: Depends(common_parameters) : This makes a declaration of the interface’s dependencies, indicating that the interface parameter requests a function that depends on common_Parameters.

  When the interface is called, the common_parameters function is called back for request processing.

• 2: The common_parameters function is mainly responsible for receiving functions, after processing, return a dictionary,

• 3: Then pass the result returned by Depends(common_parameters) to Commons: dict, which is a dependency injection process.

So in the example above common_parameters is our dependent object

The dependent object has the following requirements for interface requests:

• The optional query parameter q is a STR.
• The optional query parameter skip that is int, 0 by default.
• The optional query parameter limit is int, which is 100 by default.
• Return a dictionary

Example request:

This dependency injection approach is also quite convenient, similar to the interface decorator approach. For example, in common_parameters, we can check and intercept the relevant parameters first, and then pass them. The scene could be similar to our previous bottle decorator:

• Same logical judgment processing
• User identity authentication

PS: Dependency injection means that the injected function depends on my function and can’t run without my dependency.

11.2 Treating a class as a dependent object

While our dependent objects are represented as functions, FastAPI also supports class representation. A dependent object must be an object that can be called, such as a class or function

Let’s look at it in class form:

from fastapi import Depends, FastAPI

app = FastAPI()


fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]


class CommonQueryParams:
    def __init__(self, q: str = None, skip: int = 0, limit: int = 100) :
        self.q = q
        self.skip = skip
        self.limit = limit


@app.get("/items/")
async def read_items(commons: CommonQueryParams = Depends(CommonQueryParams)) :
    response = {}
    # if q exists
    if commons.q:
        Let's just add q to a new dictionary
        response.update({"q": commons.q})
        response.update({"Small bell.": 'students'})
    # then intercept in our fake_ITEMs_db
    items = fake_items_db[commons.skip : commons.skip + commons.limit]
    response.update({"items": items})
    return response
Copy the code

CommonQueryParams is a class that is similar to my function in that the class object is initialized when our interface is called. CommonQueryParams = Depends(CommonQueryParams) and Commons = Depends(CommonQueryParams) are equivalent. Commons: CommonQueryParams = Depends()

PS: Since the initialization operation, we can define our own verification methods in our dependency injection to carry out relevant verification. All of our dependency injection can be used as a kind of validation premise for similar permissions.

Example run demo

There are Q parameters:

No Q parameter:

11.3 Multi-level Nesting of Dependencies

Multiple layers of nesting means you can class and you can class. Functions can depend on functions. In fact, it’s the same as our previous parameter verification.

For example:

from fastapi import Cookie, Depends, FastAPI

app = FastAPI()


def query_extractor(q: str = None) :
   return q


def query_or_cookie_extractor(
   q: str = Depends(query_extractor), last_query: str = Cookie(None)
) :
   if not q:
       return last_query
   return q


@app.get("/items/")
async def read_query(query_or_default: str = Depends(query_or_cookie_extractor)) :
   return {"q_or_cookie": query_or_default}
Copy the code

Query_or_cookie_extractor depends on query_extractor, and query_OR_cookie_extractor is injected into objects that are also dependent on the interface.

The interface execution process is as follows:

For the same dependency, if the result of processing is the same, that is, the return value is the same, we can call the dependency multiple times, in this case, we can set whether to use caching mechanism on the dependent object:

async def needy_dependency(fresh_value: str = Depends(get_value, use_cache=False)):
  return {"fresh_value": fresh_value}
Copy the code

List dependencies in the Header request,

Let’s take a look at the official sample code:

from fastapi import Depends, FastAPI, Header, HTTPException

app = FastAPI()


async def verify_token(x_token: str = Header(.)) :
 ifx_token ! ="fake-super-secret-token":
     raise HTTPException(status_code=400, detail="X-Token header invalid")


async def verify_key(x_key: str = Header(.)) :
 ifx_key ! ="fake-super-secret-key":
     raise HTTPException(status_code=400, detail="X-Key header invalid")
 return x_key


@app.get("/items/", dependencies=[Depends(verify_token), Depends(verify_key)])
async def read_items() :
 return [{"item": "Foo"}, {"item": "Bar"}]
Copy the code

The above code means to validate our request Header information, because the example is… Three dots, indicating mandatory fields:

After analyzing the above code, run it and see what happens:

1: What header parameters are not passed indicating that our header parameters are abnormal

2: filling in the header parameters:

Note: the format of the parameter is submitted, because it is the header parameter, so our code should write: X-token

Examples of errors:

So the dependency of the list above means that it must be true for two days before it passes. This feeling later still use more yo!

11.5 Multi-dependency injection is the same as lists:

from fastapi import Depends, FastAPI


from fastapi import Depends, FastAPI, Header, HTTPException
from fastapi import Depends, FastAPI

app = FastAPI()




async def verify_token(x_token: str = Header(.)) :
  ifx_token ! ="fake-super-secret-token":
      raise HTTPException(status_code=400, detail="X-Token header invalid")
  return x_token

async def verify_key(x_key: str = Header(.)) :
  ifx_key ! ="fake-super-secret-key":
      raise HTTPException(status_code=400, detail="X-Key header invalid")
  return x_key


@app.get("/items/", dependencies=[Depends(verify_token), Depends(verify_key)])
async def read_items() :
  return [{"item": "Foo"}, {"item": "Bar"}]


@app.get("/items2/")
async def items2(xt: str = Depends(verify_token),xk: str = Depends(verify_key)) :
 return {"xt": xt,'xk':xk}



if __name__ == '__main__':
  import uvicorn
  uvicorn.run(app='main:app', host="127.0.0.1", port=8100, reload=True, debug=True)
Copy the code

Xt: STR = Depends(verify_token),xk: STR = Depends(verify_key).

Normal condition:Abnormal conditions:

12: FastAPI file upload and reception processing (Supplementary)

For the file upload receive processing, is also a routine operation, Fastapi actually for these aspects of consideration is actually very thoughtful, but also for us to encapsulate the processing mechanism.

Note that file uploads also depend on:

pip install python-multipart
Copy the code

• File receives a File. The File object receives bytes, which are read into memory as bytes and are only suitable for handling small files.

• UploadFile is suitable for uploading pictures and other large files. Its internal mechanism is also written to memory, but when the file reaches a certain threshold, it is automatically saved to disk without memory overflow. In addition, it can also obtain a lot of metadata information of uploaded files, and have asynchronous operation processing of file objects, and can read and write files asynchronously. Of course, if you can, for the use of AIofile to process can also!

import uvicorn from fastapi import FastAPI, File, UploadFile from pydantic import BaseModel from fastapi.encoders import jsonable_encoder from typing import List app = @app.post("/files") async def files(file_obj: bytes = File(...) Post ("/files2") async def files2(file_objs:List[bytes] = File(...) : return {' File size ': len(file_obj)} # Post ("/ uploadFiles1 ") async def files3(file_obj: UploadFile = File(...)): return {' File size ': len(file_objs)} @app.post("/ uploadFiles1 ") async def Files3 (file_obj: UploadFile = File(...) Async def files4(file_objs: List[UploadFile] = File(...)): return {' File size ': len(file_obj)} @app.post("/ uploadFiles2 ") async def files4(file_objs: List[UploadFile] = File(... ) : return {' file size: len (file_objs)} 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

View documentation:

13: FastAPI BackgroundTasks -BackgroundTasks (added)

Background asynchronous tasks as the name implies, I do not need to wait for a query to complete the execution of the love task, I will directly return, such as some of my background synchronization tasks, just need to initiate a synchronization request, return to tell me that I have submitted the execution of the task!

13.1 Simple Examples of Background Tasks

Example:

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

Output results:

The time-consuming task starts. The interval is 5 seconds. The time-consuming task startsCopy the code

13.2 Implementing Background Tasks in Dependency Injection Mode

def tasks(msg): Def tasksaction(background_tasks) def tasksAction (background_tasks) def tasksAction (background_tasks: BackgroundTasks, MSG: STR): message = f" {msg}\n" background_tasks.add_task(tasks, message) @app.post("/backgroundTasks") async def runBackgroundTasks(msg: STR = Depends(tasksaction)): print(MSG) return" 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

PS: this kind of background task, his bottom is the way to open the thread processing!

14: FastAPI common and asynchronous functions coexist runtime analysis (Supplement)

Example analysis:

import uvicorn from fastapi import FastAPI, File, UploadFile from fastapi import BackgroundTasks, FastAPI, Depends app = FastAPI() import threading @app.get("/sync") def sync(): Print ('sync- current thread name :',threading.current_thread().getName()) return "Sync task completed!" @app.get("/async") async def asyncaction(): Print (' asyncAction - current thread name :', 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

Output test results:

Sync - Current thread name: ThreadPoolExecutor-0_0 INFO: 127.0.0.1:25378 - "GET /sync HTTP/1.1" 200 OK sync- Current thread name: Threadpoolexecutor-0_1 INFO: 127.0.0.1:25378 - "GET /sync HTTP/1.1" 200 OK sync- Current thread name: threadpoolexecutor-0_2 INFO: 127.0.0.1:25378 - "GET /sync HTTP/1.1" 200 OK AsyncAction - Current thread name: MainThread INFO: 127.0.0.1:25401 - "GET /async HTTP/1.1" 200 OK AsyncAction - Current thread name: MainThread INFO: 127.0.0.1:25401 - "GET /async HTTP/1.1" 200 OK AsyncAction - Current thread name: MainThread INFO: 127.0.0.1:25401 - "GET /async HTTP/1.1" 200 OKCopy the code

From the output, it can be seen that the asynchronous function uses the coroutine mode, synchronous to frontal mode is implemented using thread pool mode. If you don’t know whether to use asynchrony or asynchrony, the best thing to do is to use asynchrony altogether. Avoid using asynchrony in order to prevent the single thread of the coroutine from getting stuck. Because coroutines are single-threaded running drops!

conclusion

I hope the cold rice is fried again without any flavor! The above is purely personal study notes, if there is a mistake also hope to give more advice.

This time of really hope will insist to continue to give the next set, mainly is to sort out their own use of the process of refining some of the use of small experience.

END

Let students | article [original] [reproduced please contact my] | QQ: 308711822