Fastapi项目初体验20230919

Fastapi实现get,post,文件上传

requirements.txt

fastapi[all]
uvicorn[standard]
aioftp
gunicorn

1
2
3
4
5

创建fastapi项目后main.py:

from fastapi import FastAPI, Query,Body,Form,UploadFile
from typing import List



app = FastAPI(title="FastAPI学习",description="学习项目的API文档",docs=None,redoc_url=None)


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


@app.get("/hello/{name}")
async def say_hello(name: str):
    return {"message": f"Hello {name}"}

# 查询单个参数 参数的传递:
# http://101.43.226.190:6602/chello?name=lxy_chello
@app.get("/chello/")
async def check_hello(name: str):
    return {"message": f"cHello {name}"}

# 查询多个参数 参数的传递:
# 查询字符串是键值对的集合，这些键值对位于url的?之后,并且用&字符进行分割
# 这种以键值对的方式去传递参数，是没有顺序要求的
# http://101.43.226.190:6602/mhello?name=lxy_chello&city=beijing&state=china
@app.get("/mhello/")
async def check_hello(name: str,city: str,state: str):
    return {"message": f"mHello {name} {city} {state}"}

# 查询多个参数 参数的传递默认值:
# http://101.43.226.190:6602/dhello?name=lxy_chello&city=beijing
@app.get("/dhello/")
async def check_hello(name: str,city: str,state: str = "CHINA"):
    return {"message": f"dHello {name} {city} {state}"}

# 查询多个参数 参数的传递可选值:
# http://101.43.226.190:6602/shello?name=lxy_chello
@app.get("/shello/")
async def check_hello(name: str,city: str = None,state: str = "CHINA"):
    return {"message": f"dHello {name} {city} {state}"}



# 对查询参数进行校验 也可以增加正则表达式
# http://101.43.226.190:6602/values?item_name=11
# default=...表示必须传递，如果是default=None就是可选择参数；也可加个默认值default=lxy
# 其他约束: regex正则,参数信息title,description有关校验
# 了解接口,比一般校验在前端进行
@app.get("/values/")
async def query_values(item_id:int = None,item_name:str=Query(default=...,min_length=3,max_length=6)):
    return {"message": f"values item_name: {item_name}"}


# 发送字典内容格式
# 接口文档(谷歌打开)
# http://127.0.0.1:6602/docs
@app.post("/post")
async def post_deal(data:dict = Body(default=...,example={"tranType":"支付","tranAmt":"1256.98","orderId":"1234454533232"})):
    """

    :param data:
    :return:
    """
    tranType = data["tranType"]
    tranAmt = data["tranAmt"]
    orderId = data["orderId"]
    print(data)
    print(type(data))
    print(tranType,tranAmt,orderId)
    return {"message_post":"success","data":data}

# 表单数据
# 有了json数据为什么还要表单发送呢,因为OAuth2规范"密码流",模式规定要通过表单字段发送username和password
# 该规范要求字段必须用username和password,并通过表单字段发送,不能用json
# 媒体类型」编码一般为 application/x-www-form-urlencoded表单数据的
@app.post("/login")
def post_form_login(username:str = Form(...),password:str = Form(...)):

    return {"username":username,"password":password}

# 上传文件路由函数
# 不包含文件时，表单数据一般用 application/x-www-form-urlencoded 「媒体类型」编码。但表单包含文件时，编码为multipart/form-data。
# UploadFile 的属性如下：
# ▪filename：上传文件名字符串（str），例如，myimage.jpg；
# ▪content_type：内容类型（MIME类型/媒体类型）字符串（str），例如，image/jpeg;
# ·file：SpooledTemporaryFile（file-like对象)。
# 其实就是Python文件，可直接传递给其他预期file-like 对象的函数或支持库。

@app.post("/file")
def upload_file(file: UploadFile):
    # 读取文件内容
    contents = file.file.read()
    # 保存文件 二进制的方式
    with open(f"./{file.filename}",mode="wb") as f:
        f.write(contents)

    return {"msg":"upload_file success"}

# 限制文件类型
@app.post("/file_PIC")
def upload_file(file: UploadFile):
    if file.content_type not in ["image/png","image/jpeg","image/jpg",]:
        return {"msg":"failed, content type error,must be png/jpeg/jpg"}
    # 读取文件内容
    contents = file.file.read()
    # 保存文件 二进制的方式
    with open(f"./{file.filename}",mode="wb") as f:
        f.write(contents)

    return {"msg":"upload_file success"}

# 协程函数上传图片
@app.post("/file_async")
async def upload_file(file: UploadFile):
    if file.content_type not in ["image/png","image/jpeg","image/jpg",]:
        return {"msg":"failed, content type error,must be png/jpeg/jpg"}
    # 读取文件内容
    contents = await file.read()
    # 保存文件 二进制的方式
    with open(f"./{file.filename}",mode="wb") as f:
        f.write(contents)

    return {"msg":"upload_file success"}

#在 async 路径操作函数内，要用以下方式读取文件内容：
#contents = await file.read
#在普通 def 路径操作函数内，则可以直接访问 UploadFile.file，例如：
#contents = file.file.read()

# 上传多个文件
@app.post("/files_async")
async def upload_files(files: List[UploadFile]):
    for file in files:
        if file.content_type not in ["image/png","image/jpeg","image/jpg",]:
            continue
        # 读取文件内容
        contents = await file.read()
        # 保存文件 二进制的方式
        with open(f"./{file.filename}",mode="wb") as f:
            f.write(contents)

    return {"msg":"upload_file success"}


if __name__ == '__main__':
    import uvicorn

    uvicorn.run(app='main:app',host='0.0.0.0',port=6602,reload=True)


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
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152