forked from liaogx/fastapi-tutorial
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathchapter04.py
More file actions
149 lines (107 loc) · 4.58 KB
/
chapter04.py
File metadata and controls
149 lines (107 loc) · 4.58 KB
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
#!/usr/bin/python3
# -*- coding:utf-8 -*-
# __author__ = '__Jack__'
from typing import Optional, List
from fastapi import APIRouter, status, Form, File, UploadFile, HTTPException
from pydantic import BaseModel, EmailStr
app04 = APIRouter()
"""Response Model 响应模型"""
class UserIn(BaseModel):
username: str
password: str
email: EmailStr
mobile: str = "10086"
address: str = None
full_name: Optional[str] = None
class UserOut(BaseModel):
username: str
email: EmailStr # 用 EmailStr 需要 pip install pydantic[email]
mobile: str = "10086"
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表示默认值不包含在响应中,仅包含实际给的值,如果实际给的值与默认值相同也会包含在响应中"""
print(user.password) # password不会被返回
# return user
return users["user01"]
@app04.post(
"/response_model/attributes",
response_model=UserOut,
# response_model=Union[UserIn, UserOut],
# response_model=List[UserOut],
response_model_include=["username", "email", "mobile"],
response_model_exclude=["mobile"]
)
async def response_model_attributes(user: UserIn):
"""response_model_include列出需要在返回结果中包含的字段;response_model_exclude列出需要在返回结果中排除的字段"""
# del user.password # Union[UserIn, UserOut]后,删除password属性也能返回成功
return user
# return [user, user]
"""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}
"""Form Data 表单数据处理"""
@app04.post("/login/")
async def login(username: str = Form(...), password: str = Form(...)): # 定义表单参数
"""用Form类需要pip install python-multipart; Form类的元数据和校验方法类似Body/Query/Path/Cookie"""
return {"username": username}
"""Request Files 单文件、多文件上传及参数详解"""
@app04.post("/file")
async def file_(file: bytes = File(...)): # 如果要上传多个文件 files: List[bytes] = File(...)
"""使用File类 文件内容会以bytes的形式读入内存 适合于上传小文件"""
return {"file_size": len(file)}
@app04.post("/upload_files")
async def upload_files(files: List[UploadFile] = File(...)): # 如果要上传单个文件 file: UploadFile = File(...)
"""
使用UploadFile类的优势:
1.文件存储在内存中,使用的内存达到阈值后,将被保存在磁盘中
2.适合于图片、视频大文件
3.可以获取上传的文件的元数据,如文件名,创建时间等
4.有文件对象的异步接口
5.上传的文件是Python文件对象,可以使用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}
"""【见run.py】FastAPI项目的静态文件配置"""
"""Path Operation Configuration 路径操作配置"""
@app04.post(
"/path_operation_configuration",
response_model=UserOut,
# tags=["Path", "Operation", "Configuration"],
summary="This is summary",
description="This is description",
response_description="This is response description",
deprecated=True,
status_code=status.HTTP_200_OK
)
async def path_operation_configuration(user: UserIn):
"""
Path Operation Configuration 路径操作配置
:param user: 用户信息
:return: 返回结果
"""
return user.dict()
"""【见run.py】FastAPI 应用的常见配置项"""
"""Handling Errors 错误处理"""
@app04.get("/http_exception")
async def http_exception(city: str):
if city != "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}