FastAPI– Python high-performance Web framework

Chapter 2 FastAPI introduction and project preparation

2.1 What is the relationship between Starlette, Pydantic and FastAPI framework?

Python type hints, type hints Type hints

from typing import List
def func(name:str,age:int,l:List) :
    
    print(name,age)
    print(l)    
# Python type hints, used well to improve code robustness

Copy the code

Pydantic is a library that defines data validation, serialization, and documentation (using JSON schema) based on Python type hints

Starlette is a lightweight ASGI framework/toolkit that is ideal for building high-performance Asyncio services

Installation environment, Python environment must be 3.6 or higher, then go to Github to github.com/liaogx/fast… Copy and install the package in reqiurements.txt

Be sure to pay attention to version compatibility

2.2 Pydantic Basics Tutorial

Data validation and settings management using python type annotations. Setting Management pydantic Enforces Type Hints at Runtime And provides User friendly Errors when data is invalid. Pydantic can provide type hints at runtime. Define How data should be in pure, Canonical Python; Validate it with Pydantic. Define how data should be stored in pure canonical Python code, and validate it with PydanticCopy the code

from typing import List
from datetime import datetime
from pydantic import BaseModel

# if a class attribute has a default value that is optional, no default value is required
class User(BaseModel) :
    id = int
    name: str = "andy"
    signup_list: datetime
    friends: List[int] = []

external_data = {
    "id": 1."signup_list": "The 2021-02-02 10:10"."friends": [1.2."3"]
}
user = User(**external_data)
print(user.name)
print(user.friends)
print(repr(user.signup_list))
print(user.dict())  Output in dictionary format
print(user.json())  Output in JSON format


# class People(User):
# def start(self):
# print(f" my name {self.name}")
# def friend(self):
#         print(f"我的朋友们{str(self.friends)}")
#
# people = People(**external_data)
#
# people.start()
# people.friend()

Copy the code

Pycharm installs the Pydantic plugin

Verification failure processing

Verification failure processing
try:
    User(id=1,signup_list=datetime.today(),friends=[1.2.'hello world'])
except ValidationError as e:
    print(e.json())
Copy the code

Printed result

[{"loc": [
      "friends".2]."msg": "value is not a valid integer"."type": "type_error.integer"}]Copy the code

Properties and methods of the model class

print(user.dict())
print(user.json())
print(user.copy())  This is a shallow copy
print(User.parse_obj(external_data))
print(User.parse_raw('{"id": "123", "signup_ts": "2020-12-22 12:22", "friends": [1, 2, "3"]}'))

path = Path('pydantic_tutorial.json')
path.write_text('{"id": "123", "signup_ts": "2020-12-22 12:22", "friends": [1, 2, "3"]}')
print(User.parse_file(path))

print(user.schema())
print(user.schema_json())

user_data = {"id": "error"."signup_ts": "The 2020-12-22 12 22"."friends": [1.2.3]}  # id is a string error
print(User.construct(**user_data))  Create a model class without checking the data. It is not recommended to pass unvalidated data into the Construct method

print(User.__fields__.keys())  When defining a model class, specify the type of all fields, so that the order of the fields will not be out of order
Copy the code

Recursive model

class Sound(BaseModel) :
    sound: str

class Dog(BaseModel) :
    birthday: date
    weight: float = Optional[None]
    sound: List[Sound]

dogs = Dog(birthday=date.today(),weight=9.99,sound=[{"sound":"wang wang~"}, {"sound": "ying ying ~"}])
print(dogs.json())
Copy the code

The execution result

{"birthday": "2021-02-02"."sound": [{"sound": "wang wang~"}, {"sound": "ying ying ~"}}]Copy the code

ORM models, which are created from class instances to conform to ORM objects

from sqlalchemy.dialects.postgresql import ARRAY
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Integer, String, Column

Base = declarative_base()

class CompanyOrm(Base) :
    __tablename__ = 'companies'
    id =Column(Integer, primary_key=True, nullable=True)
    public_key = Column(String(20), index=True, nullable=False, unique=True)
    name = Column(String(100), unique=True)
    domains = Column(ARRAY(String(255)))


class CompanyModel(BaseModel) :
    id: int
    public_key: constr(max_length=20)
    name: constr(max_length=100)
    domains: List[constr(max_length=255)]

    class Config:
        orm_mode = True


co_orm = CompanyOrm(
    id=1,
    public_key="akey",
    name="andy",
    domains=['123.com'.'456.com'])print(CompanyModel.from_orm(co_orm))
Copy the code

Chapter 3: Request parameters and validation

3.1 Simple use of FastAPI

from typing import Optional
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class People(BaseModel) :
    id: int
    name: str
    addr: Optional[str] = None


@app.get('/')
async def person() :
    return {"name": "hello world"}


@app.get('/city/{city}')
async def result(city: str, query_str: Optional[str] = None) :
    return {'city': city, 'query_str': query_str}


@app.post('/person/{person}')
async def result(person: str, city_info: People) :
    return {"person": person, "id": city_info.id.'name': city_info.name, 'addr': city_info.addr}

Copy the code

#Project initiationUvicorn filename :app --reload
#Generation of interface documents
localhost:8000/docs

#Interfaces that display interfaces
localhost:8000/redoc
Copy the code

3.2 fastApi structure

Start by creating a file package for each app

Then implement it in chapter03.py

from enum import Enum

from fastapi import APIRouter

app03 = APIRouter()

Copy the code

Same operation as any other app

In the *inif.py* file

from .chapter03 import app03
from .chapter04 import app04
Copy the code

In the run.py file

import uvicorn
from fastapi import FastAPI
from tutorial import app03

app = FastAPI()

# Integrate subapps of the app
app.include_router(app03, prefix='/chapter03', tags=['Chapter 3, Request Parameters and Validation'])



if __name__ == '__main__':
    uvicorn.run('run:app',host='0.0.0.0',port=8000,reload=True,debug=True,workers=1)


Copy the code

from enum import Enum
from typing import Optional.List
from fastapi import APIRouter, Query, Cookie, Header
from pydantic import BaseModel, Field
from datetime import date

app03 = APIRouter()

"""Path Parameters and Number Validations


@app03.get('/path/parameters')
def path_params01() :
    return {"message": "This is a message"}


@app03.get('/path/{parameters}')
def path_parameters02(parameters: str) :
    return {"message": parameters}


# enumeration
class CityName(str, Enum) :
    Beijing = "Beijing china"
    Shanghai = "Shanghai china"


@app03.get('/enum/{city}')
async def latest(city: CityName) :
    if city == CityName.Shanghai:
        return {"city_name": city, "confirmed": 1492."death": 7}
    elif city == CityName.Beijing:
        return {'city_name': city, "confirmed": 971."death": 9}
    return {"city_name": city, 'latest': "unknown"}


@app03.get('/files/{filepath:path}')
def filepath(filepath: str) :
    return f"This file path is {filepath}"


"""Query Parameters and String Validations


@app03.get('/query')
def page_limit(page: int = 1, limit: Optional[int] = None) :
    if limit:
        return {'page': page, 'limit': limit}
    else:
        return {"page": page}


@app03.get('/query/bool/conversion')  # bool type conversion: yes on 1 True True will be converted to True, other false
def query_params_validate(query: bool = False) :
    return F "returns zero{query}"


@app03.get("/query/validations")
def query_params_validate(
        value: str = Query(. , min_length=8, max_length=16, regex="^a"),
        values: List[str] = Query(['V1'.'V2'], alias="alias_name")
) :
    return value, values


"("Request Body and Fields")


class CityInfo(BaseModel) :
    name: str= Field(... , example="Beijing")  # example is an annotation. The value is not validated
    country: str
    country_code: str = None  Give a default value
    country_population: int = Field(default=800, title="Population size", description="The number of people in the country", ge=800)

    class Config:
        schema_extra = {
            "example": {
                "name": "Shanghai"."country": "China"."country_code": "CN"."country_population": 1400000000,}}@app03.post("/request_body/city")
def city_info(city: CityInfo) :
    print(city.name, city.country)  The property pops up automatically when you type city. In the IDE
    return city.dict(a)""Request Body - Nested Request Body of Nodels data format


class Data(BaseModel) :
    city: List[CityInfo] = None
    date: date
    confirmed: int = Field(ge=0, description="Confirmed number", default=0)
    deaths: int = Field(ge=0, description="Deaths", default=0)
    recovered: int = Field(ge=0, description="Recovery number", default=0)


@app03.put('/request_body/nested')
def nested_models(data: Data) :
    return data


""" How to set Cookie and Header parameters """


@app03.get("/cookie")  This can only be tested with postman. Add Cookie = cookie_id= XXX to the Header
Use the Cookie class to define the Cookie parameter, otherwise it is the query parameter
def cookie(cookie_id: Optional[str] = Cookie(None)) :
    return cookie_id


@app03.get("/header")
def header(user_agent: Optional[str] = Header(None, convert_underscores=True), x_token: List[str] = Header(None)) :
    "" Some HTTP proxies and servers don't allow underscores in request headers, so the Header provides the convert_underscores attribute to set param user_Agent: Convert_underscores =True changes user_Agent to user-agent :param x_Token: X_Token is a list of multiple values :return: """
    return {"User-Agent": user_agent, "x-Token": x_token}

Copy the code

Query–from fastAPI import Query

Verify parameters in the request body using Field– from Pydantic import Field

The Filed class is used when defining the requested body data using Pandtic.

When you validate the data using the Path parameter using the Path class,

The Query class is used to validate Query parameters

Chapter 4 response handling and FastAPI configuration

4-1 response model example elaboration

from fastapi import BaseModel
from pydantic import EmailStr
from typing import Option,List
""4.1 Reponse Model


class UserIn(BaseModel) :
    username: str
    password: str
    email: EmailStr
    mobile: str = '110'
    address: str = None
    full_name: Optional[str] = None


class UserOut(BaseModel) :
    username: str
    email: EmailStr
    mobile: str = '110'
    address: str = None
    full_name: Optional[str] = None


users = {
    "user01": {"username": "user01"."password": "123123"."email": "[email protected]"},
    "user02": {"username": "user02"."password": "123456"."email": "[email protected]"."mobile": "110"}}@app04.post("/response_model", response_model=UserOut, response_model_exclude_unset=True)
async def response_model(user: UserIn) :
    Response_model_exclude_unset =True indicates that the default value is not included in the response, only the value actually given. If the value actually given is the same as the default value, it will be included in the response.

    print(user.password)
    return users["user02"]


@app04.post(
    "/response_model/attributes",
    response_model=UserOut,  # specify the model class
    # response_model=Union[UserIn, UserOut], # combine two model classes
    # response_model=List[UserOut], # response_model=List[UserOut]
    response_model_include=["username"."email"."mobile"].# which fields to include
    response_model_exclude=["mobile"]  # Which fields to filter out
)
async def response_model_attributes(user: UserIn) :
    """response_model_include lists the fields that need to be included in the return result; Response_model_exclude lists the fields that need to be excluded from the return result.
    # del user.password # Union[UserIn, UserOut
    return user
    # return [user, user]


Copy the code

4.2 Response status code and Shortcut Properties

from fastapi import status
""Response Status Code Response Status Code

 
@app04.post("/status_code", status_code=200)
async def status_code() :
    return {"status_code": 200}


@app04.post("/status_attribute", status_code=status.HTTP_200_OK)
async def status_attribute() :
    print(type(status.HTTP_200_OK))
    return {"status_code": status.HTTP_200_OK}

Copy the code

4.3 Form data processing

from fastapi import Form, File, UploadFile

"""Form Data processing ""


@app04.post("/login/")
async def login(username: str = Form(.), password: str = Form(.)) :  Define the form parameters
    PIP install python-multipart; The Form metadata and validation methods are similar to Body/Query/Path/Cookie""
    return {"username": username}


"""Request Files Single file, multiple file upload and parameter description


@app04.post("/file")
async def file_(file: bytes = File(.)) :  Files: List[bytes] = File(...)
    Using the File class the contents of files are read into memory as bytes suitable for uploading small files.
    return {"file_size": len(file)}


@app04.post("/upload_files")
async def upload_files(files: List[UploadFile] = File(.)) :  UploadFile = file (...)
    """ The advantages of using UploadFile class are as follows: 1. The file is stored in memory. When the memory usage reaches the threshold, it will be saved on disk. It is suitable for large files such as pictures and videos. 3. It can obtain the metadata of uploaded files, such as file name and creation time. Write (), read(), seek(), close()
    for file in files:
        contents = await file.read()
        print(contents)
    return {"filename": files[0].filename, "content_type": files[0].content_type}

Copy the code

4.4 FastAPI project static file configuration

# in the run. Py
from fastapi.staticfiles import StaticFiles
# mount means to mount a completely separate application from a directory, which is not shown in the API interaction documentation
app.mount(path='corona/static',app=StaticFiles(directory='./corona/statuic'),name="static")

Copy the code

4.4 FastAPI project static file configuration

# in the run. Py
from fastapi.staticfiles import StaticFiles
# mount means to mount a completely separate application from a directory, which is not shown in the API interaction documentation
app.mount(path='static',app=StaticFiles(directory='./corona/statuic'),name="static")

Copy the code

4.5 Configuring Path Operations

@app04.post(
    "/path_operation_configuration",
    response_model=UserOut,  # specify the response model
    # tags=["Path", "Operation", "Configuration"],
    summary="This is summary",  
    description="This is description",
    response_description="This is response description",
    deprecated=True.# indicates whether the interface is available
    status_code=status.HTTP_200_OK  # status code
)
async def path_operation_configuration(user: UserIn) :
    """ Path Operation Configuration Path Operation Configuration :param user: user information :return: return result """
    return user.dict(a)Copy the code

4.6 FastAPI Common Configuration Items

# run.py
app = FastAPI(
    title='FastAPI Tutorial and Coronavirus Tracker API Docs',
    description='FastAPI tutorials will be coronavirus outbreak tracker API interface document, project code: https://github.com/liaogx/fastapi-tutorial',
    version='1.0.0',
    docs_url='/docs',
    redoc_url='/redocs'.)Copy the code

4.7 FastAPI Exception Handling

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

@app04.get("/http_exception")
async def http_exception(city: str) :
    ifcity ! ="Beijing":
        raise HTTPException(status_code=404, detail="City not found!", headers={"X-Error": "Error"})
    return {"city": city}


@app04.get("/http_exception/{city_id}")
async def override_http_exception(city_id: int) :
    if city_id == 1:
        raise HTTPException(status_code=418, detail="Nope! I don't like 1.")
    return {"city_id": city_id}

Copy the code

Chapter 5: Dependency injection systems for FastAPI

5.1 This section describes the dependency injection system and its application scenarios

Dependency injection refers to the process of importing or declaring dependencies, such as subfunctions and database connections, in order to ensure the successful operation of the code

1 improve code reuse rate 2 share database links 3 enhance security, authentication and role managementCopy the code

FastAPI compatibility

1 All relational databases, supporting NoSQL databases 2 third-party packages and apis 3 Authentication and authorization systems 4 Response data injection systemsCopy the code

5.2 Create, import, and Declare dependencies

We use inheritance to pass parameters in Django or Flask, and declare dependencies in FastAPI

Function as dependency

Create, import, and declare Dependencies

async def common_parameters(q: Optional[str] = None, page: int = 1, limit: int = 100) :
    return {"q": 1."page": page, "limit": limit}

Def dependencies can be called in async def, and async def dependencies can be imported in def
@app05.get('/dependency01')
async def dependency01(commons: dict = Depends(common_parameters)) :
    return commons


@app05.get('/dependency02')
def dependency02(commons: dict = Depends(common_parameters)) :
    return commons

Copy the code

5.3 How to Use a class as a dependency

Class as dependency

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


class CommonQueryParams:
    def __init__(self, q: Optional[str] = None, page: int = 1, limit: int = 100) :
        self.q = q
        self.page = page
        self.limit = limit


@app05.get('/classes_as_dependencies')
async def classes_as_dependencies(commons=Depends(CommonQueryParams)) :
    response = {}
    if commons.q:
        response.update({"q": commons.q})
    itmes = fake_itmes_db[commons.page:commons.limit]
    response.update({'items': itmes})
    return response

Copy the code

5.4 Self-created and invoked

def query(q: Optional[str] = None) :
    return q


def sub_query(q: str = Depends(query), last_query: Optional[str] = None) :
    if not q:
        return last_query
    return q


@app05.get("/sub_dependency")
async def sub_dependency(final_query: str = Depends(sub_query, use_cache=True)) :
    """ USe_cache defaults to True, which means that when multiple dependencies have a child in common, the child will only be called once per request, and multiple calls will fetch """
    return {"sub_dependency": final_query}

Copy the code

5-5 Importing dependencies in the path operation decorator

"" import dependency in path operation decorator ""


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


async def verify_key(x_key: str = Header(.)) :
    """ returns a child dependency of the value, but the return value will not be called """
    ifx_key ! ="fake-supplier-secret-token":
        raise HTTPException(status_code=400, detail="x_Key header invalid")
    return x_key


@app05.get('/dependency_in_path_operation', dependencies=[Depends(verify_token), Depends(verify_key)])
async def dependency_in_path_operation() :
    return {"fuck": "fuck"}

Copy the code

5.6 Use of global dependencies in FastAPI framework

We can not only add dependencies to a single interface to verify that tokens protect sercet_key

It can also be added in a child application

app05 = APIRouter(dependencies=[Depends(verify_token),Depends(verify_key)])
Copy the code

Dependencies can also be added to the global app

app = FastAPI(dependencies=[Depends(verify_token),Depends(verify_key)])
Copy the code

In fact, this can be our interface for permissions and so on verification

5.7 Using Yield dependencies and subdependencies

"" if you want to add a new item to your list, you need to add it to your list.


Python3.6 requires PIP install async-exit-stack async-generator
The following is all pseudo-code

async def get_db() :
    db = "db_connection"
    try:
        yield db
    finally:
        db.endswith("db_close")


async def dependency_a() :
    dep_a = "generate_dep_a()"
    try:
        yield dep_a
    finally:
        dep_a.endswith("db_close")


async def dependency_b(dep_a=Depends(dependency_a)) :
    dep_b = "generate_dep_b()"
    try:
        yield dep_b
    finally:
        dep_b.endswith(dep_a)


async def dependency_c(dep_b=Depends(dependency_b)) :
    dep_c = "generate_dep_c()"
    try:
        yield dep_c
    finally:
        dep_c.endswith(dep_b)

Copy the code

Chapter 6: Security, Authentication, and Authorization

6.1 OAuth2 password mode and FastAPI OAuth2 passwordbearer

OAuth2 password mode and FastAPI OAuth2 passwordbearer

OAuth2PasswordBearer is a class that accepts urls as parameters: The client sends the USERNAME and password parameters to that URL and then gets a Token OAuth2PasswordBearer that doesn't create the corresponding URL path operations, it just specifies the URL address that the client uses to request the Token when the request comes in, FastAPI checks the Authorization header information for requests and returns an UNAUTHORIZED 401 status code if the Authorization header information is not found or if the header content is not Bearer token.
oauth2_schema = OAuth2PasswordBearer(tokenUrl="/chapter06/token")


@app06.get('/oauth2_password_bearer')
async def oauth2_password_bearer(token: str = Depends(oauth2_schema)) :
    return {"token": token}
Copy the code

6.2 OAuth2 authentication based on Password and Bearer token

OAuth2 password mode and FastAPI OAuth2 passwordbearer

OAuth2PasswordBearer is a class that accepts urls as parameters: The client sends the USERNAME and password parameters to that URL and then gets a Token OAuth2PasswordBearer that doesn't create the corresponding URL path operations, it just specifies the URL address that the client uses to request the Token when the request comes in, FastAPI checks the Authorization header information for requests and returns an UNAUTHORIZED 401 status code if the Authorization header information is not found or if the header content is not Bearer token.

oauth2_schema = OAuth2PasswordBearer(tokenUrl="/chapter06/token")  # request Token URL http://127.0.0.1:8000/chapter06/token


@app06.get("/oauth2_password_bearer")
async def oauth2_password_bearer(token: str = Depends(oauth2_schema)) :
    return {"token": token}


OAuth2 authentication based on Password and Bearer token

fake_users_db = {
    "john snow": {
        "username": "john snow"."full_name": "John Snow"."email": "[email protected]"."hashed_password": "fakehashedsecret"."disabled": False,},"alice": {
        "username": "alice"."full_name": "Alice Wonderson"."email": "[email protected]"."hashed_password": "fakehashedsecret2"."disabled": True,}}def fake_hash_password(password: str) :
    return "fakehashed" + password


class User(BaseModel) :
    username: str
    email: Optional[str] = None
    full_name: Optional[str] = None
    disabled: Optional[bool] = None


class UserInDB(User) :
    hashed_password: str


@app06.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()) :
    user_dict = fake_users_db.get(form_data.username)
    if not user_dict:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Incorrect username or password")
    user = UserInDB(**user_dict)
    hashed_password = fake_hash_password(form_data.password)
    if not hashed_password == user.hashed_password:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Incorrect username or password")
    return {"access_token": user.username, "token_type": "bearer"}


def get_user(db, username: str) :
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)


def fake_decode_token(token: str) :
    user = get_user(fake_users_db, token)
    return user


async def get_current_user(token: str = Depends(oauth2_schema)) :
    user = fake_decode_token(token)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials",
            headers={"WWW-Authenticate": "Bearer"},  If authentication fails, return "www-authenticate" in the request header
        )
    return user


async def get_current_active_user(current_user: User = Depends(get_current_user)) :
    if current_user.disabled:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Inactive user")
    return current_user


@app06.get("/users/me")
async def read_users_me(current_user: User = Depends(get_current_active_user)) :
    return current_user


fake_users_db.update({
    "john snow": {
        "username": "john snow"."full_name": "John Snow"."email": "[email protected]"."hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW"."disabled": False,}})Copy the code

6.3 Develop authentication based on JSON Web Tokens

OAuth2 with Password (and hashing), Bearer of JWT tokens developed authentication based on JSON Web tokens
SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"  Openssl rand-hex 32
ALGORITHM = "HS256"  # algorithm
ACCESS_TOKEN_EXPIRE_MINUTES = 30  Access token expires in minutes


class Token(BaseModel) :
    "" Token returned to user """
    access_token: str
    token_type: str


pwd_context = CryptContext(schemes=['bcrypt'], deprecated='auto')
oauth2_schema = OAuth2PasswordBearer(tokenUrl='/chapter06/jwt/token')


def verify_password(plain_password, hashed_password) :
    "" verify password ""
    return pwd_context.verify(plain_password, hashed_password)


def jwt_get_user(db, username: str) :
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)


def jwt_authenticate_user(db, username: str, password: str) :
    user = jwt_get_user(db, username)
    if not user:
        return False
    if not verify_password(password, user.hashed_password):
        return False
    return user


def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) :
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({'exp': expire})
    encode_jwt = jwt.encode(claims=to_encode, key=SECRET_KEY, algorithm=ALGORITHM)
    return encode_jwt


@app06.post('/jwt/token', response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()) :
    user = jwt_authenticate_user(db=fake_users_db, username=form_data.username, password=form_data.password)
    if not user:
        raise HTTPException(
            status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},)Get the expiration time of the Token
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username},
        expires_delta=access_token_expires
    )
    return {"access_token": access_token, "token_type": "bearer"}


async def jwt_get_current_user(token: str = Depends(oauth2_schema)) :
    credentials_exception = HTTPException(
        status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},)try:
        payload = jwt.decode(token=token, key=SECRET_KEY, algorithms=[ALGORITHM])
        username = payload.get("sub")
        if username is None:
            raise credentials_exception

    except JWTError:
        raise credentials_exception

    user = jwt_get_user(db=fake_users_db, username=username)
    if user is None:
        raise credentials_exception
    return user


async def jwt_get_current_active_user(current_user: User = Depends(jwt_get_current_user)) :
    if current_user.disabled:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Inactive user")
    return current_user


@app06.get("/jwt/users/me")
async def jwt_read_users_me(current_user: User = Depends(jwt_get_current_active_user)) :
    return current_user

Copy the code

Chapter 7 FastAPI database operation and multi – application directory structure design

The project structure

7.1 Configuring SQLAlchemy database operations in FlastAPI

# database.py


from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = 'sqlite:///./coronavirus.sqlite3'
# SQLALCHEMY_DATABASE_URL = "postgresql: / / username: password @ host: port/database_name" # MySQL or postgresql connection methods

engine = create_engine(
    # echo=True indicates that the engine logs all statements and their argument lists with the repr() function
    # Since SQLAlchemy is multithreaded, specify check_same_thread=False to make the created object available to any thread. This parameter is only set when using an SQLite database
    SQLALCHEMY_DATABASE_URL, encoding='utf-8', echo=True, connect_args={'check_same_thread': False})In SQLAlchemy, CRUD is done through sessions, so we have to create sessions first. Each SessionLocal instance is a database session
# flush() sends a database statement to the database, but the database does not necessarily perform a write to disk; Commit () commits the transaction, saving the changes to a database file
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, expire_on_commit=True)

Create a basic mapping class
Base = declarative_base(bind=engine, name='Base')
Copy the code

7.2 SQLAlchemy model class development

# models.py

from sqlalchemy import Column, String, Integer, BigInteger, Date, DateTime, ForeignKey, func
from sqlalchemy.orm import relationship

from .database import Base


class City(Base) :
    __tablename__ = 'city'  Alter table name

    id = Column(Integer, primary_key=True, index=True, autoincrement=True)
    province = Column(String(100), unique=True, nullable=False, comment='Province/Municipality')
    country = Column(String(100), nullable=False, comment='countries')
    country_code = Column(String(100), nullable=False, comment='Country code')
    country_population = Column(BigInteger, nullable=False, comment='National population')
    data = relationship('Data', back_populates='city')  # 'Data' is the associated class name; Back_populates to specify the property name for reverse access

    created_at = Column(DateTime, server_default=func.now(), comment='Creation time')
    updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now(), comment='Update Time')

    __mapper_args__ = {"order_by": country_code}  # desc()

    def __repr__(self) :
        return f'{self.country}_{self.province}'


class Data(Base) :
    __tablename__ = 'data'

    id = Column(Integer, primary_key=True, index=True, autoincrement=True)
    city_id = Column(Integer, ForeignKey('city.id'), comment='Province/Municipality')  The string format in # ForeignKey is not the class name. Property name, but table name. The field name
    date = Column(Date, nullable=False, comment='Data date')
    confirmed = Column(BigInteger, default=0, nullable=False, comment='Confirmed number')
    deaths = Column(BigInteger, default=0, nullable=False, comment='Number of deaths')
    recovered = Column(BigInteger, default=0, nullable=False, comment='Cure quantity')
    city = relationship('City', back_populates='data')  # 'City' is the associated class name; Back_populates to specify the property name for reverse access

    created_at = Column(DateTime, server_default=func.now(), comment='Creation time')
    updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now(), comment='Update Time')

    __mapper_args__ = {"order_by": date.desc()}  # In descending order by date

    def __repr__(self) :
        return f'{repr(self.date)}Confirmed:{self.confirmed}Patients'


"" "attach three basic operation of SQLAlchemy tutorial SQLAlchemy http://www.taodudu.cc/news/show-175725.html Python3 + SQLAlchemy + Sqlite3 ORM tutorial Basic knowledge of https://www.cnblogs.com/jiangxiaobo/p/12350561.html SQLAlchemy Autoflush and the Autocommit mode https://zhuanlan.zhihu.com/p/48994990 """

Copy the code

7.3 Pydantic establishes data format classes corresponding to model classes

"" Standardize data format ""
# schemas.py

from datetime import date as date_
from datetime import datetime

from pydantic import BaseModel


class CreateData(BaseModel) :
    date: date_
    confirmed: int = 0
    deaths: int = 0
    recovered: int = 0


class CreateCity(BaseModel) :
    province: str
    country: str
    country_code: str
    country_population: int


class ReadData(CreateData) :
    id: int
    city_id: int
    updated_at: datetime
    created_at: datetime

    class Config:
        orm_mode = True


class ReadCity(CreateCity) :
    id: int
    updated_at: datetime
    created_at: datetime

    class Config:
        orm_mode = True

Copy the code

7.4 Encapsulate functions created and queried

# crud.py


from sqlalchemy.orm import Session
from corona import models, schemas


Select * from city by id
def get_city(db: Session, city_id: int) :
    return db.query(models.City).filter(models.City.id == city_id).first()


Select * from city by city name
def get_city_by_name(db: Session, name: str) :
    return db.query(models.City).filter(models.City.province == name).first()


# select a number of cities
def get_cities(db: Session, skip: int = 0, limit: int = 10) :
    return db.query(models.City).offset(skip).limit(limit).all(a)# Create a city
def create_city(db: Session, city: schemas.CreateCity) :
    db_city = models.City(**city.dict())
    db.add(db_city)
    db.commit()
    db.refresh(db_city)
    return db_city


# Fetch data
def get_data(db: Session, city: str = None, skip: int = 0, limit: int = 100) :
    if city:
        return db.query(models.Data).filter(models.Data.has(province=city))
    return db.query(models.Data).offset(skip).limit(limit).all(a)# create data
def create_city_data(db: Session, data: schemas.CreateCity, city_id: int) :
    db_data = models.Data(**data.dict(), city_id=city_id)
    db.add(db_data)
    db.commit()
    db.refresh(db_data)
    return db_data

Copy the code

7.5 Jinjia2 Render the front-end page

main.py

#! /usr/bin/python3
# -*- coding:utf-8 -*-
# __author__ = '__Jack__'

from typing import List

import requests
from fastapi import APIRouter, Depends, HTTPException, BackgroundTasks, Request
from fastapi.templating import Jinja2Templates
from pydantic import HttpUrl
from sqlalchemy.orm import Session

from coronavirus import crud, schemas
from coronavirus.database import engine, Base, SessionLocal
from coronavirus.models import City, Data

application = APIRouter()

templates = Jinja2Templates(directory='./coronavirus/templates')

Base.metadata.create_all(bind=engine)


def get_db() :
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@application.post("/create_city", response_model=schemas.ReadCity)
def create_city(city: schemas.CreateCity, db: Session = Depends(get_db)) :
    db_city = crud.get_city_by_name(db, name=city.province)
    if db_city:
        raise HTTPException(status_code=400, detail="City already registered")
    return crud.create_city(db=db, city=city)


@application.get("/get_city/{city}", response_model=schemas.ReadCity)
def get_city(city: str, db: Session = Depends(get_db)) :
    db_city = crud.get_city_by_name(db, name=city)
    if db_city is None:
        raise HTTPException(status_code=404, detail="City not found")
    return db_city


@application.get("/get_cities", response_model=List[schemas.ReadCity])
def get_cities(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)) :
    cities = crud.get_cities(db, skip=skip, limit=limit)
    return cities


@application.post("/create_data", response_model=schemas.ReadData)
def create_data_for_city(city: str, data: schemas.CreateData, db: Session = Depends(get_db)) :
    db_city = crud.get_city_by_name(db, name=city)
    data = crud.create_city_data(db=db, data=data, city_id=db_city.id)
    return data


@application.get("/get_data")
def get_data(city: str = None, skip: int = 0, limit: int = 100, db: Session = Depends(get_db)) :
    data = crud.get_data(db, city=city, skip=skip, limit=limit)
    return data


def bg_task(url: HttpUrl, db: Session) :
    Db: Session = Session (get_db)

    city_data = requests.get(url=f"{url}? source=jhu&country_code=CN&timelines=false")

    if 200 == city_data.status_code:
        db.query(City).delete()  Delete the original data before synchronizing data
        for location in city_data.json()["locations"]:
            city = {
                "province": location["province"]."country": location["country"]."country_code": "CN"."country_population": location["country_population"]
            }
            crud.create_city(db=db, city=schemas.CreateCity(**city))

    coronavirus_data = requests.get(url=f"{url}? source=jhu&country_code=CN&timelines=true")

    if 200 == coronavirus_data.status_code:
        db.query(Data).delete()
        for city in coronavirus_data.json()["locations"]:
            db_city = crud.get_city_by_name(db=db, name=city["province"])
            for date, confirmed in city["timelines"] ["confirmed"] ["timeline"].items():
                data = {
                    "date": date.split("T") [0].# change '2020-12-3t00:00:00 Z' to '2020-12-31'
                    "confirmed": confirmed,
                    "deaths": city["timelines"] ["deaths"] ["timeline"][date],
                    "recovered": 0  # There is no data on how many people are cured per day in each city
                }
                City_id is the primary key ID in the city table, not the ID in the coronavirus_data table
                crud.create_city_data(db=db, data=schemas.CreateData(**data), city_id=db_city.id)


@application.get("/sync_coronavirus_data/jhu")
def sync_coronavirus_data(background_tasks: BackgroundTasks, db: Session = Depends(get_db)) :
    "" synchronizing COVID-19 data from Johns Hopkins University """
    background_tasks.add_task(bg_task, "https://coronavirus-tracker-api.herokuapp.com/v2/locations", db)
    return {"message": "Synchronizing data in background..."}


@application.get("/")
def coronavirus(request: Request, city: str = None, skip: int = 0, limit: int = 100, db: Session = Depends(get_db)) :
    data = crud.get_data(db, city=city, skip=skip, limit=limit)
    return templates.TemplateResponse("home.html", {
        "request": request,
        "data": data,
        "sync_data_url": "/coronavirus/sync_coronavirus_data/jhu"
    })

Copy the code

home.html

<! DOCTYPE html> <html lang="en"> <head> <title> Novel Coronavirus Pandemic tracker </title> <link rel="stylesheet" href="{{ url_for('static', path='/semantic.min.css') }}">
    <script src="{{url_for (' static ', path ="/jquery - 3.5.1 track of/jquery - 3.5.1 track of. Min. Js')}}"></script>
    <script src="{{ url_for('static', path='/semantic.min.js') }}"></script>
    <script>
        $(document).ready(function () {
            $("#filter").click(function () {
                const city = $("#city").val();
                window.location.href = "http://" + window.location.host + "/coronavirus? city=" + city;
            });
            $("#sync").click(function () {
                $.get("{{ sync_data_url }}", function (result) {
                    alert("Message: " + result.message);
                });
            });
        });
    </script>
</head>

<body>
<div class="ui container">
    <h2></h2>
    <h1 style="text-align: centeR ">    filter" style="float: left" type="submit" class="ui button alert-secondary ui input"> city"></label><input id="city" type="text" placeholder="city" value="">  sync" style="float: right" type="submit" class="ui button primaryui celled table"> < thead > < tr > < th > city < / th > < th > date < / th > < th > total number of confirmed < / th > < th > total deaths < / th > < th > total recovery number < / th > < th > update time < / th > < / tr > < thead >  {% for d in data %}  {{ d.city.province }} {{ d.date }} {{ d.confirmed }} {{ d.deaths }} {{ d.recovered }} {{ d.updated_at }}  {% endfor %}     Copy the code

7.6 Large engineering wooden road structure design – application file splitting

from fastapi import APIRouter, Depends, Request

[""] [See Coronavirus application] SQL (Relational) Databases

"""Bigger Applications - Multiple Files directory structure design """


async def get_user_agent(request: Request) :
    print(request.headers["User-Agent"])


app07 = APIRouter(
    prefix="/bigger_applications",
    tags=["Chapter 7 FastAPI database Operation and Multi-application directory structure Design"].# Has the same name as the tags in run.py
    dependencies=[Depends(get_user_agent)],
    responses={200: {"description": "Good job!"}},)@app07.get("/bigger_applications")
async def bigger_applications() :
    return {"message": "Bigger Applications - Multiple Files"}

Copy the code

Chapter 8: Middleware, CORS, Background tasks, test cases

8.1 Concepts and development examples of middleware

# run.py

from fastapi import FastAPI, Request
app = FastAPI(
    title='FastAPI Tutorial and Coronavirus Tracker API Docs',
    description='FastAPI tutorials will be coronavirus outbreak tracker API interface document, project code: https://github.com/liaogx/fastapi-tutorial',
    version='1.0.0',
    docs_url='/docs',
    redoc_url='/redocs'.)@app.middleware('http')
async def add_process_time_header(request: Request, call_next) :  # call_next takes the request as an argument
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers['X-Process-Time'] = str(process_time)  # add a custom request header starting with "X-"
    return response
Copy the code

8.2 Principles of Resource Sharing CORS

To put it bluntly, we are cross-site, js request conflict or disallow

8.3 CORSMiddleware implementation of FastAPI

# run.py
from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=[
        "http://127.0.0.1"."http://127.0.0.1:8080"
    ],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],)Copy the code

8.4 FastAPI implements background tasks of Lesie Celery

from typing import Optional

from fastapi import APIRouter, BackgroundTasks, Depends

app08 = APIRouter()

[see run.py] Middleware """

# Note: The dependency exit portion of the code with yield and background tasks will run after the middleware

[see run.py] CORS (Cross-origin Resource Sharing)

# Domain: protocol + domain + port

"(""Background Tasks """


def bg_task(framework: str) :
    with open("README.md", mode="a") as f:
        f.write(f"## {framework}"Frame elaboration")


@app08.post("/background_tasks")
async def run_bg_task(framework: str, background_tasks: BackgroundTasks) :
    """ : Param Framework: arguments to the background task function called :param background_tasks: FastAPI.BackgroundTasks :return: """
    background_tasks.add_task(bg_task, framework)
    return {"message": "Task already running in background"}


def continue_write_readme(background_tasks: BackgroundTasks, q: Optional[str] = None) :
    if q:
        background_tasks.add_task(bg_task, "\n> The overall introduction of FastAPI, quick start development, combined with API interactive documents one by one to explain the use of core modules \n")
    return q


@app08.post("/dependency/background_tasks")
async def dependency_run_bg_task(q: str = Depends(continue_write_readme)) :
    if q:
        return {"message": "Readme. md updated successfully"}

Copy the code

8.5 Background Task Updates

See 7.5 main. Py

8.6 TestClient Writing Test Cases

from fastapi.testclient import TestClient

from run import app

"(" > < p style =" max-width: 100%; clear: both;

client = TestClient(app)  PIP install pytest


def test_run_bg_task() :  # Function names beginning with "test_" are the pyTest specification. Notice not async def
    response = client.post(url="/chapter08/background_tasks? framework=FastAPI")
    assert response.status_code == 200
    assert response.json() == {"message": "Task already running in background"}


def test_dependency_run_bg_task() :
    response = client.post(url="/chapter08/dependency/background_tasks")
    assert response.status_code == 200
    assert response.json() is None


def test_dependency_run_bg_task_q() :
    response = client.post(url="/chapter08/dependency/background_tasks? q=1")
    assert response.status_code == 200
    assert response.json() == {"message": "Readme. md updated successfully"}

Copy the code