【Fastapi框架】Fastapi的使用和进阶_fastapi 启动

原文作者：我辈李想
版权声明：文章原创，转载时请务必加上原文超链接、作者信息和本声明。

---

---

前言

---

一、项目启动

1.安装

pip install fastapi  uvicorn
1

2.示例

新建main.py文件，

from typing import Union

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
def read_root():
    return {"Hello": "World"}


@app.get("/items/{item_id}")
def read_item(item_id: int, q: Union[str, None] = None):
    return {"item_id": item_id, "q": q}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

3.启动

方式一：命令启动

uvicorn main:app --reload
1

方式二:在mian.py中加启动代码

# mian.py

if __name__ == '__main__':
    uvicorn.run(
        app=app,
        host="127.0.0.1",
        port=5678
    )
1
2
3
4
5
6
7
8

4.路由

http://127.0.0.1:8000/
http://127.0.0.1:8000/items/1
http://127.0.0.1:8000/items/1?q=2
http://127.0.0.1:5678/docs#/
http://127.0.0.1:5678/redoc

二、进阶

1.请求数据

1.get请求:路径参数+查询参数

路由：http://127.0.0.1:8000/items/1?q=2，其中1是路径参数，q=2是查询参数。

@app.get("/")
def read_root():
    return {"Hello": "World"}


@app.get("/items/{item_id}")
def read_item(item_id: int, q: Union[str, None] = None):
    return {"item_id": item_id, "q": q}
1
2
3
4
5
6
7
8

2.post/put/delete请求：路径参数+请求体

路由：http://127.0.0.1:8000/items/1，其中1是路径参数，item是请求体。pydantic 可以验证输入和输出。

from typing import Union

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float
    fromPileNo: Optional[str] = None  # 允许为空，无参数是默认None  
    slabNoList: List[str] = [] # 默认[]
    is_offer: Union[bool, None] = None

@app.put("/items/{item_id}")
def update_item(item_id: int, item: Item):
    return {"item_name": item.name, "item_id": item_id}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

3.嵌套数据

# 1.定义基础模型
class Image(BaseModel):
    url: str

# 2.赋给其他类属性
class Item(BaseModel):
    image: Union[Image, None] = None  # 单个
    images: Union[List[Image], None] = None  # 多个
1
2
3
4
5
6
7
8

4.文件上传

from fastapi import FastAPI, status, File, Form, UploadFile

app = FastAPI()

# 方法一：上传的文件会存在内存中，适合小型文件 
@app.post("/api/uploadFile/action")
async def create_file(file: Annotated[bytes, File()]):  
	writeBytes('./media',file)
	return {
        'code':200,
        "msg":'success'
    }

# 方法二：UploadFile
@app.post("/api/uploadFile/action")
async def create_upload_file(file: UploadFile):
	writeBytes('./media',file)
	return {
        'code':200,
        "msg":'success'
    }

# 上传多个文件
@app.post("/api/uploadFile/action")
async def create_upload_files(files: list[UploadFile]):
	writeBytes('./media',file)
	return {
        'code':200,
        "msg":'success'
    }
    
# 用 File() 添加额外设定
@app.post("/api/uploadFile/action")
async def create_upload_files(
    files: Annotated[
        list[UploadFile], File(description="Multiple files as UploadFile")
    ],
):
	writeBytes('./media',file)
	return {
        'code':200,
        "msg":'success'
    }
    
# 将file写入dirs目录文件
def writeBytes(dirs,file):
    bytesFile=file.file.read()
    filename=file.filename
    if not os.path.exists(dirs):
        os.makedirs(dirs)
    with open(dirs+'/'+ filename, "wb") as f:
        f.write(bytesFile)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

5.数据校验

使用fastapi.params库中的函数，FastAPI会自动验证请求参数。Query，Bath 等。

from fastapi import FastAPI, Query

app = FastAPI()

@app.get("/items/")
async def read_items(q: str = Query(None, min_length=3, max_length=50)):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results

1
2
3
4
5
6
7
8
9
10
11

6.request参数

@router.api_route("/test/{apiname}", methods=['GET', 'POST'])
async def test1(apiname, request: Request):
    args = await request.form()  # form数据
    args2 = await request.json()   # json数据
    print(apiname)
    print(request.method)
    return {'result': '这是一个GET或者POST'}
1
2
3
4
5
6
7

7.文档注释

@router.post('/roll-plan/kpi/', summary='获取输入数据', description='', tags=['算法'])
async def get_roll_plan(times: Annotated[RollPlanTimeItem, Body(examples=[{"start_time": "2023-12-12 00:00:00",
                                                                           "end_time": "2023-12-13 00:00:00"}],
                                                                )]
                        ):
1
2
3
4
5

8.请求体对象转json

from datetime import datetime
from typing import Union

from fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    title: str
    timestamp: datetime
    description: Union[str, None] = None

@app.put("/items/{id}")
def update_item(id: str, item: Item):
    json_compatible_item_data = jsonable_encoder(item)
    # item.json()
    # item.dict()
    fake_db[id] = json_compatible_item_data
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

9.sqlalchemy查询数据转json

roll_plans = session.query(rollPlan.AppRollPlanEntity).filter(
	rollPlan.AppRollPlanEntity.StartTime >= time_range[0],
	rollPlan.AppRollPlanEntity.StartTime <= time_range[1]).all()
	
# 方式一
from sqlalchemy.ext.serializer import loads, dumps
def to_dict(row):
    row_dict = row.__dict__
    if '_sa_instance_state' in row_dict:
        del row_dict['_sa_instance_state']
    return row_dict

serialized_data = [to_dict(result) for result in roll_plans]  # 将每个结果转换成字典形式
json_data = dumps(serialized_data)  # 将字典序列化为JSON字符串
# 反序列化
deserialized_data = loads(serialized_data)
user = User(**deserialized_data)

# 方式二
import json
import decimal

class MyEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, datetime.datetime):
            print("MyEncoder-datetime.datetime")
            return obj.strftime("%Y-%m-%d %H:%M:%S")
        if isinstance(obj, bytes):
            return str(obj, encoding='utf-8')
        if isinstance(obj, int):
            return int(obj)
        elif isinstance(obj, float):
            return float(obj)
        elif isinstance(obj, decimal.Decimal):
            return float(obj)
        # elif isinstance(obj, decimal):
        #    return decimal(obj)
        else:
            return super(MyEncoder, self).default(obj)
            
def to_dict(row):
    row_dict = row.__dict__
    if '_sa_instance_state' in row_dict:
        del row_dict['_sa_instance_state']
    return row_dict

serialized_data = [to_dict(result) for result in roll_plans]  # 将每个结果转换成字典形式
json_data = json.dumps(serialized_data,cls=MyEncoder, ensure_ascii=False)  # 将字典序列化为JSON字符串
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

2.静态文件加载

from fastapi import FastAPI
from starlette.staticfiles import StaticFiles

app = FastAPI()

# 静态文件配置使用mount挂载 第一个参数为应用的前缀，第二个为路径，第三个为别名
app.mount('/static', StaticFiles(directory='static'), name='static')
1
2
3
4
5
6
7

这里需要再main.py的同级目录创建static文件，在html文件中需要使用url_for引入静态文件。

<link rel="stylesheet" href="{{ url_for('static',path='demo.css') }}">
1

3.路由管理

1.在maipn.py的同级创建user文件夹,新建userRouter.py文件

from typing import Union

from fastapi import FastAPI, APIRouter, Request

router = APIRouter(prefix='/bp', tags=['bp'])

@router.get('/')
def read_root():
    return {"Hello": "World"}

@router.get('/items/{item_id}')
async def read_item(item_id: int, q: Union[str, None] = None):
    return {"item_id": item_id, "q": q}


# @router.get('/test/{apiname}')
# @router.post("/test/{apiname}")
@router.api_route("/test/{apiname}", methods=['GET', 'POST'])
async def test1(apiname, request: Request):
	args = await request.form()
    print(apiname)
    print(request.method)
    return {'result': '这是一个GET或者POST'}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

2.修改mapin.py内容

from fastapi import FastAPI
from user import userRouter 

app = FastAPI()
app.include_router(userRouter.router)

if __name__ == '__main__':
    uvicorn.run(
        app=app,
        host="127.0.0.1",
        port=5678
    )
1
2
3
4
5
6
7
8
9
10
11
12

4.跨域配置

# 跨域配置
origins = [
    "http://localhost:3000",
]
app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
1
2
3
4
5
6
7
8
9
10
11

5.自定义中间件

1.已有中间件

HTTPSRedirectMiddleware是fasapi自带的中间件

app.add_middleware(HTTPSRedirectMiddleware)
app.add_middleware(
    TrustedHostMiddleware, allowed_hosts=["tigeriaf.com", "*.tigeriaf.com"]
)
1
2
3
4

2.自定义中间件

@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
1
2
3
4
5
6
7

6.使用jwt中间件

在 FastAPI 中使用 JWT 中间件，可以通过自定义 FastAPI 的中间件来实现。以下是在 FastAPI 中使用 JWT 中间件的步骤。

1. 安装 PyJWT。

pip install PyJWT
1

2. 导入 PyJWT 库和 FastAPI 组件。

import jwt
from fastapi import FastAPI, HTTPException
from fastapi.middleware import Middleware
from fastapi.middleware.cors import CORSMiddleware
from starlette.requests import Request
from starlette.status import HTTP_401_UNAUTHORIZED
1
2
3
4
5
6

3. 定义一些全局变量，例如：

SECRET_KEY = "mysecretkey"  # 生产环境中应使用更复杂的密钥
ALGORITHM = "HS256"  # 哈希算法
ACCESS_TOKEN_EXPIRE_MINUTES = 30  # 访问令牌的过期时间，单位为分钟
1
2
3

4. 定义 JWT 认证的中间件函数。

async def auth_middleware(request: Request, call_next):
    try:
        token: str = request.headers["Authorization"].split()[1]
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise HTTPException(status_code=HTTP_401_UNAUTHORIZED, detail="Invalid authentication credentials")
        request.state.user = username
    except (KeyError, IndexError, jwt.JWTError):
        raise HTTPException(status_code=HTTP_401_UNAUTHORIZED, detail="Invalid authentication credentials")
    response = await call_next(request)
    return response
1
2
3
4
5
6
7
8
9
10
11
12

auth_middleware 函数获取请求头中的访问令牌，并解码 JWT。如果 JWT 解码成功，则将用户名添加到请求对象的 state 属性中。如果 JWT 解码失败，则抛出 HTTPException 异常。

5. 在 FastAPI 中添加中间件。

middleware = [
    Middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"]),
    Middleware(auth_middleware)
]

app = FastAPI(
    title="My API",
    openapi_url="/api/v1/openapi.json",
    docs_url="/api/v1/docs",
    redoc_url=None,
    middleware=middleware
)
1
2
3
4
5
6
7
8
9
10
11
12

在 FastAPI 的构造函数中，使用 Middleware 类添加 CORSMiddleware 和 auth_middleware 中间件。auth_middleware 中间件将在每个路由处理函数执行前进行验证。因此，只需将路由的处理函数包装在 async def 函数中即可。

@app.get("/protected")
async def protected_route(request: Request):
    current_user = request.state.user
    return {"message": "Hello, {}".format(current_user)}
1
2
3
4

在以上示例中，使用 Request 类作为处理函数的一个参数，然后从请求的 state 属性中获取用户。

以上是使用 JWT 中间件的步骤，你可以根据项目的需求对上述代码进行进一步修改和优化。

7.数据库迁移

fastapi使用sqlalchemy来连接数据库，对于数据库的迁移推荐使用alembic，alembic是sqlalchemy推荐的迁移工具。参考链接