preface

Previously, the online business has been using Bottle and Flask frameworks to write API interfaces. Bottle and Flask are both synchronous frameworks, and the latest iteration of support for asynchronous frameworks is a nice corner: I have also learned about its singularity in the article of Qingnan University. Out of my interest in learning asynchronous framework, I decided to try Fastapi as well. Although I have used Tornado before, I only use synchronous framework.

I wanted to give it a try and see if SANic was in the mood for novelty or tinkering with the following Fastapi!

The best way to learn a new framework is of course the official documentation provided by the framework.

Literature: fastapi.tiangolo.com

Source code: github.com/tiangolo/fa…

Reference data: www.jianshu.com/p/94710ed35…

Began to lu code

Note that FastAPI only supports Python3.6+ apis, so you need to learn from Python3.6+!

Because I am under the Windows environment, only for open debugging, really to the line, must be running in the Linux environment to reach the ultimate performance of ASGI!

1: installation of dependent libraries

Or:

FastAPI – is a modern, fast (high-performance) Python Web framework

pip install fastapi

Uvicorn – a server primarily used to load and serve applications.

pip install uvicorn

2: The first Hello World

import uvicorn
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def root(a):
    return {"message": "Hello World"}

if __name__ == '__main__':
    uvicorn.run(app=app)
Copy the code

If you dig into the uvicorn.run(app=app) method, you will see a:

def run(app, **kwargs):
    config = Config(app, **kwargs)
    server = Server(config=config)

    if (config.reload or config.workers > 1) and not isinstance(app, str):
        logger = logging.getLogger("uvicorn.error")
        logger.warn(
            "You must pass the application as an import string to enable 'reload' or 'workers'."
        )
        sys.exit(1)

    if config.should_reload:
        sock = config.bind_socket()
        supervisor = StatReload(config, target=server.run, sockets=[sock])
        supervisor.run()
    elif config.workers > 1:
        sock = config.bind_socket()
        supervisor = Multiprocess(config, target=server.run, sockets=[sock])
        supervisor.run()
    else:
        server.run()
Copy the code

Further into config = config (app, **kwargs), there are a number of related configuration items:

class Config:
    def __init__(
        self,
        app,
        host="127.0.0.1",
        port=8000,
        uds=None,
        fd=None,
        loop="auto",
        http="auto",
        ws="auto",
        lifespan="auto",
        env_file=None,
        log_config=LOGGING_CONFIG,
        log_level=None,
        access_log=True,
        use_colors=None,
        interface="auto",
        debug=False,
        reload=False,
        reload_dirs=None,
        workers=None,
        proxy_headers=True,
        forwarded_allow_ips=None,
        root_path="",
        limit_concurrency=None,
        limit_max_requests=None,
        backlog=2048,
        timeout_keep_alive=5,
        timeout_notify=30,
        callback_notify=None,
        ssl_keyfile=None,
        ssl_certfile=None,
        ssl_version=SSL_PROTOCOL_VERSION,
        ssl_cert_reqs=ssl.CERT_NONE,
        ssl_ca_certs=None,
        ssl_ciphers="TLSv1",
        headers=None,
    ):
Copy the code

So the parameters that can be added can be filled in by looking at the information of several configuration options:

As a result, it can also be modified as:

import uvicorn
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def root(a):
    return {"message": "Hello 454533343433World"}

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

What happens when you try to hot update your code? An alarm message is displayed:

WARNING:  You must pass the application as an import string to enable 'reload' or 'workers'.
Copy the code

Warning: The application must be passed as an import string in order to enable reload.

  uvicorn.run(app='app', host="127.0.0.1", port=8000, reload=True, debug=True)
Copy the code

Tip:

ERROR:    Error loading ASGI app. Import string "app" must be in format "<module>:<attribute>".
Copy the code

Well, let me check the official document again and it says:

The command line is required: module + app name:

Well, know:

    uvicorn.run(app='main:app', host="127.0.0.1", port=8000, reload=True, debug=True)
Copy the code

Then you can start the hot update restart service!

Then access the address and get the body of the message returned by the interface normally:

Then click on the official website to check the API document interaction address:

http://127.0.0.1:8000/docs

http://127.0.0.1:8000/redoc

The routing methods include GET, POST, PUT, PATCH, DELETE, and OPTIONS.

@app.post("/")
@app.put("/")
@app.delete("/")
@app.get("/")
@app.options("/")
@app.head("/")
@app.patch("/")
@app.trace("/")
async def root(a):
    return {"message": "Hello 454533333343433World"}
Copy the code

3: Route Obtain and verify parameters on a Route

Generally, our routing has static and dynamic components. Static routing means that the parameter is fixed and written dead, that is, the access address is written dead, while dynamic address needs to be generated dynamically. 94710ED35b92, the address of a blog similar to Simple book, is dynamic, in fact, the same as Bottle and Flask.

www.jianshu.com/p/94710ed35…

from fastapi import FastAPI

app = FastAPI()


@app.get("/items/{item_id}")
async def read_item(item_id):
    return {"item_id": item_id}
Copy the code

In the example above, item_id is a dynamic parameter that you can pass in at will.

Then there is a definition like bottle that can also verify the data of the parameters passed in:

from fastapi import FastAPI

app = FastAPI()



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

Item_id: int if item_id is converted to an int, an error will be reported.

About routing coverage: The following two routing addresses:

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


@app.get("/users/{user_id}")
async def read_user(user_id: str):
    return {"Preferentially matched to:": user_id}
Copy the code

/users/{user_id} overwrites /users/me!

3.1 Querying Path Parameters and Verifying Parameters

Query parameters are the parameters of the POSTMAN submission:

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

Skip =0&limit is the so-called query parameter.

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]
Copy the code

The first case of access:

The second type of visit:

Third type of access:

3.2 Ultrapath and Query Parameters

Multipath and query parameters are urls that contain dynamic parameters, and parameters that need to be submitted through the ampersand delimitor.

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
Copy the code

Request:

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

Request:

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

3.3 Mandatory and Optional Path and query parameters

The optional and mandatory parameters are determined primarily by whether to give default values:

Such as:

@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:

You can also define the submission types of Optional and mandatory parameters: you can also 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

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

3.4 Enumeration of path parameters

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"}


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

By visiting the address:

http://127.0.0.1:8000/model/alexnet

3.5 Query Parameters Other verification of Query parameters

Traditionally, wTForm was used to define similar or optional or length types for submitted field information. In Fastapi, we define the Fastapi as Query in: from Fastapi import Fastapi, Query, for example:

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
Copy the code

q: q: STR = Query(None, min_length=3,max_length=50),regex=”^ fixedQuery $”) the q parameter is optional, but if specified, the maximum length must be less than 50, and the minimum length must be greater than 3: And it needs to 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$”)

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

3.6 Querying Parameters Query Specifies the multi-value list of parameters

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:

3.7 Other Verification methods for 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.

3.8 Parameter Submitted Request Body

Generally, the Request Body is not submitted through GET. Parameters submitted by GET are called query parameters. So, if the parameters are submitted by POTS,PUT, etc., we usually PUT them in the Request Body to submit them to our back end.

FastApi provides a form for how to receive and validate the request body using: From Pydantic import BaseModel

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 above model, I define the format that the submitted Item must have, such as name as a mandatory field, description as optional and default to None, price as mandatory and need to be of type float, and tax as optional and default 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:

3.8 Request Body and a mixture of Query and Path

In the design of some apis, it is inevitable to encounter some of the above mash-up 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:

3.9 Submission of Multiple Request Bodies

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.

If another assumption is made:

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"
    },
    "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:

If, on the other hand, the client submits a single object that is embedded, what do we do? :

{
    "item": {
        "name": "Foo"."description": "The pretender"."price": 42.0."tax": 3.2}}Copy the code

FastAPI provides one:

item: Item = Body(… , Embed =True)

from fastapi import Body, FastAPI
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, item: Item = Body(... , embed=True)):
    results = {"item_id": item_id, "item": item}
    return results
Copy the code

An example request is as follows:

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.

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

3.10 Request Body的Field

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:

3.11 Verification of Other Data Types

For data format validation, generally, we do more than just

• int
• float
• str
• bool

However, the submission parameters can be in more than one of the above formats, sometimes such as mobile phone number verification, sometimes time type verification, etc

Other types:

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

4: indicates response packets

4.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:

When redefining our API to return a response, it usually returns a fixed JSON format, so we can directly define 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(a):
    return {"foo": 2.3."bar": 3.4}
Copy the code

4.2 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

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:

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

5: error handling

5.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.

5.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

5.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.

5.4 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:

Note that only HTTPException in the FastAPI package can define the request header
from fastapi import Depends, status, HTTPException
# from starlette.exceptions import HTTPException
Copy the code

6: FastAPI 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:

7: FastAPI 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.

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(a):
    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.

8: 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:

8.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

8.2 Treating classes as dependents

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()

Example run demo

There are Q parameters:

No Q parameter:

8.3 Multi-level nested 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

8.4 List List dependency

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(a):
 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!

8.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(a):
  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:

conclusion

In general, the article follows the ideas of the official documents, briefly combs them, and then we use some knowledge points of API. The next step is to try a simple scaffold based on the above information.

END

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