后端（FastAPI)学习笔记（CLASS 2）：FastAPI框架

FastAPI是一个用于构建API的现代、快速（高性能）的web框架，使用python并基于标准的python类型提示。

一、第一个web应用

定义一个返回hello world的接口

from fastapi import FastAPI
import uvicorn

# 创建一个FastAPI应用
app = FastAPI()

# 开发第一个接口
@app.get("/")
async def root():
    return { "message": "hello world", "code":1001, "data": { "name": "张三", "age": 18 } }

if __name__ == "__main__":
    uvicorn.run(app, host="localhost", port=3000)

在路径后面添加/docs是该框架自带的接口文档

二、请求解析

1、查询参数

如果要给接口定义参数，直接在处理函数中定义参数即可

from fastapi import FastAPI
import uvicorn

app = FastAPI()

@app.get("/items/")
async def read_item(page: int = 0, size: int = 0):
    return { "page": page, "size": size, "data": [11, 22, 33, 44] }

if __name__ == '__main__':
    uvicorn.run(app, host="localhost", port=3000)

结合查询参数设置默认值

def demo_page(page: str | None = None, size: Union[int, None] = None):
    """
    结合查询参数设置默认值
    page: str | None = None
    size: Union[int, None]
    """

* 查询参数的校验

fastapi中提供一个Query对象，可以使用Query显示地将其声明为查询参数，并可以给参数设置类型和相关的限制

# ======== 查询参数的校验 ==========
from typing import Union
from fastapi import FastAPI, Query
import uvicorn

app = FastAPI()

@app.get("/demo/page/check")
def demo_page_check(token: Union[str, None] = Query(
    default=None, # 默认值
    min_length=5,
    max_length=10,
    regex="^[a-zA-Z0-9_-}{5, 10}$]", # 正则表达式
    title="这是一个token",
    description="参数的描述说明信息",
    deprecated=True,
    alias="token" # 参数的别名
)):
    return { "name": token }

if __name__ == '__main__':
    uvicorn.run(app, host="localhost", port=3000)

2、路径参数

如果要设置接口的路径参数，可以直接在请求路径在通过{参数}来定义参数，然后再对应的处理函数中定义同名的参数来进行接收

from fastapi import FastAPI
import uvicorn

app = FastAPI()

@app.get('/item/{item_id}')
async def read_item(item_id: int):
    return {"item_id": item_id}

if __name__ == '__main__':
    uvicorn.run(app, host='localhost', port=3000)

* 路径参数的校验

与使用Query为查询参数声明更多的校验和元数据的方式相同，你也可以使用path为路径参数声明相同类型的校验和元数据

from fastapi import FastAPI, Path

app = FastAPI()

@app.get('/item/{item_id}')
async def read_item(
    item_id: int = Path(title='The ID of the item to get', gt=0, le=1000),
    q: str
):
    results = {"item_id": item_id}
    if q: 
        results.update({"q":q})
        return results

路径参数的数值校验方式

gt： 大于（gather than）

ge： 大于等于（gather than or equal）

lt： 小于（less than）

le：小于等于（less than or equal）

3、请求体参数

1、表单参数

from fastapi import FastAPI, Form
import uvicorn

app = FastAPI()

@app.post('/login/')
async def login(username: str = Form(
    min_length=5,
    max_length=10,
    title='用户名',
    description='用户名',
    deprecated=True,
    alias='username'
    ), password: str = Form()):
    return { "username": username, "password": password }

if __name__ == '__main__':
    uvicorn.run(app, host='localhost', port=3000)

注意：如果不使用Form对象，那么就是查询参数，且要使用post请求方式

2、JSON参数

如果要定义请求体参数，则需要使用pydantic定义参数的模型，然后再处理参数中指定参数的模型类型

from fastapi import FastAPI, Form
from pydantic import BaseModel
import uvicorn

app = FastAPI()

class RegisterInfo(BaseModel):
    username: str
    password: str
    password_confirm: str
    email: str
    mobile: str

@app.post('/api/user/register/')
async def register(item: RegisterInfo):
    return item.dict()

if __name__ == '__main__':
    uvicorn.run(app, host='localhost', port=3000)

4、cookie参数

定义在cookie中传递的参数

from typing import Annotated
import uvicorn
from fastapi import Cookie, FastAPI

app = FastAPI()

@app.get("/items/")
async def read_items(token: Annotated[str | None, Cookie()] = None):
    return {"token": token}

if __name__ == '__main__':
    uvicorn.run(app, host='localhost', port=3000)

注意，请不要使用接口文档进行传值测试，可以写一个脚本来测试

5、header参数

定义在请求头的传递的字段参数

from typing import Annotated
from fastapi import FastAPI, Header
import uvicorn

app = FastAPI()

@app.get("/items/")
async def read_items(user_agent: Annotated[str | None, Header()] = None):
    return { "User_Agent": user_agent }

if __name__ == '__main__':
    uvicorn.run(app, host='localhost', port=3000)

*  请求解析：

路径参数，查询参数，

请求头参数：表单，json

cookie

请求头

6、文件上传

UploadFile用于定义客户端的上传文件。因为上传文件以【表单数据】的形式发送，所以接收上传文件，要预先安装python-multipart

安装命令： pip install python-multipart

案例代码：

from fastapi import FastAPI, File, UploadFile
import uvicorn

app = FastAPI()

# 单文件上传
@app.post('/upload/file')
def upload_file(file: UploadFile):
    name = file.filename
    content_type = file.content_type
    
    return { 'name': name, 'content_type': content_type }

if __name__ == '__main__':
    uvicorn.run(app, host='localhost', port=3000)

当我上传该文件后：

如果需要上传多个文件，只需将参数的类型声明改为List[UploadFile]，并且使用数组进行返回就行

* 文件的校验与存储

import os
from fastapi import FastAPI, File, UploadFile
import uvicorn

app = FastAPI()

FILE_TYPES = ['image/jpeg', 'image/png', 'image/bmp', 'image/jpg', 'image/webp']
FILE_UPLOAD_URL = './uploads'

# 单文件上传
@app.post('/upload/file')
def upload_file(file: UploadFile):
    name = file.filename
    content_type = file.content_type
    # 对文件进行校验
    if content_type not in FILE_TYPES:
        return {'error': '文件类型错误'}
    if file.size > 1024 * 50:
        return {'error': '文件大小超过50kb'}
    # 判断文件是否存在
    if name in os.listdir(FILE_UPLOAD_URL):
        return {'error': '文件已存在'}
    
    
    with open(os.path.join(FILE_UPLOAD_URL, name), 'wb') as f:
        f.write(file.file.read())
    
    return { 'name': name, 'content_type': content_type, "url": 'http://localhost:3000/uploads/' + name, "file": file }

if __name__ == '__main__':
    uvicorn.run(app, host='localhost', port=3000)

其中使用文件操作的函数，可以将图片存进相对路径的对应位置，实际中可以使用创建写的方式，按照图片上传的时间进行分类

UploadFile的属性如下：

1、filename：上传文件名字符串（str），例如，myImage.jpg

2、content_type：内容类型（MIME类型/媒体类型）字符串（str），例如，image/jpeg

3、file：SpooledTemporaryFile（file-like对象），其实就是python文件，可直接传递给其他预期file-like对象的函数或支持库

UploadFile支持以下async方法，（使用内部 SpooledTemporaryFile）可调用相应的文件方法

4、write（data）：把data（str或bytes）写入文件

5、read（size）：按指定数量的字节或字符（size（int））读取文件内容

6、close（）：关闭文件

7、错误处理

使用HTTPException的类即可，以下是语法示例：

from fastapi import FastAPI, HTTPException
import uvicorn

app = FastAPI()
userId = [1, 2, 3, 4]

@app.get('/api/user/{user_id}')
async def find_user(user_id: int):
    if user_id not in userId:
        raise HTTPException(status_code=404, detail='不存在该用户')
    
    return { "data": { "user_id": user_id } }

if __name__ == '__main__':
    uvicorn.run(app, host='localhost', port=3000)

后续也一样，查询数据库也是类似的

三、请求响应

1、响应状态码

指定响应模型的方式相同，在以下任意路径操作中，可以使用status_code参数声明用于响应的HTTP状态码：

1、@app.get()：定义get请求的API

2、@app.post()：定义post请求的API

3、@app.put()：定义put请求的API

4、@app.delete()：定义delete请求的API

等等......

以下为模拟增删改查的代码：

from typing import Optional
from fastapi import FastAPI, Path, HTTPException
from pydantic import BaseModel
import uvicorn

app = FastAPI()

# 定义一个模拟数据库的数组
users = [
    {
        "user_id": 1,
        "name": '小明',
        "age": 18,
        "sex": '男'
    },
    {
        "user_id": 2,
        "name": '小红',
        "age": 20,
        "sex": '女'
    }
]

class User(BaseModel):
    user_id: int
    name: str
    age: int
    sex: str
    
class UserUpdate(BaseModel):
    user_id: Optional[int] = None
    name: Optional[str] = None
    age: Optional[int] = None
    sex: Optional[str] = None
    
# 查询所有的用户
@app.get('/api/user', status_code=200)
async def find_all_user():
    return { "msg": '查询成功', "data": users}

# 根据id查询用户信息
@app.get('/api/user/{user_id}', status_code=200)
async def find_user_by_id(user_id: int = Path(title='用户的id')):
    for user in users:
        if user.get('user_id') == user_id:
            return { "msg": f'找到id为{user_id}的用户', "data": user }
    
    raise HTTPException(status_code=404, detail=f'不存在id为{user_id}的用户')

# 增加用户（注册）
@app.post('/api/user/add', status_code=201)
async def add_user(user: User):
    user.user_id = len(users) + 1
    users.append(user.dict())   

    return { 'msg': '添加成功', "data": users}

# 更新用户信息
@app.put('/api/user/update/{user_id}', status_code=200)
async def update_user(user_id: int, user_update: UserUpdate):
    for user in users:
        if user.get('user_id') == user_id:
            update_data = user_update.dict(exclude_unset=True)
            if user_id in update_data:
                del update_data['user_id']
            users[users.index(user)].update(update_data)
            return { 'msg': '更新成功', 'data': users }
    
    raise HTTPException(status_code=404, detail=f'未找到id为{user_id}的用户')

# 删除用户信息
@app.delete('/api/user/delete', status_code=204)
async def delete_user(user_id: int):
    for user in users:
        if user.get('user_id') == user_id:
            users.remove(user)
            return { 'msg': f'删除id为{user_id}的数据成功', "data": users }
        
    raise HTTPException(status_code=404, detail=f'id为{user_id}的用户不存在')

    """
    新增类型：201
    删除类型：204
    """

if __name__ == '__main__':
    uvicorn.run(app, host='localhost', port=3000)
    

2、响应模型

即定义一个类，该类和返回的数据的数据结构相同，响应模型可在接口文档中提现，例如：

class User(BaseModel):
    user_id: int
    name: str
    age: int
    sex: str

# 查询所有的用户
@app.get('/api/user', status_code=200, response_model=list[User])
async def find_all_user():
    return { "msg": '查询成功', "data": users}

3、自定义响应内容

fastAPI默认返回的是json数据，如果我们有需求返回其他类型的内容，应该怎么实现呢？fastAPI中有内置的响应对象可以帮我们实现返回自定义的响应内容

1、JsonResponse

FastAPI中默认返回的就是JsonResponse格式数据，我们也可以使用JsonResponse来自定义返回的数据

@app.get('/item/{id}')
def update_item(id: str):
    if id == 'foo':
        return { "id": 'foo', "value": 'my hero' }
    return JSONResponse(status_code=404, content={"message": '未找到'})

2、HTMLResponse

@app.get('/item/', response_class=HTMLResponse)
async def read_item():
    return """
    <html>
        <head>
            <title>Some HTML in here</title>
        </head>
        <body>
            <h1>Look ma!HTML!</h1>
        </body>
    </html>
    """

3、更多响应模型

除了返回json和html之外，fastapi对应常用资源的返回都做了封装，具体参考官方文档

4、响应Cookies

可以通过响应对象的set_cookie方法来设置返回的响应头信息

from fastapi import FastAPI, Response
from fastapi.responses import JSONResponse

app = FastAPI()

@app.post('/cookie-and-object')
def create_cookie(response: Response):
    response.set_cookie(key="fakesession", value="fake-cookie-session-value")
    
    return {"msg": "come to the dark side"}

@app.post('/cookie/')
def create_cookie():
    content = { "msg": "come to the dark side" }
    res = JSONResponse(content=content)
    res.set_cookie(key="fakesession", value="fake-cookie-session-value")
    return res

5、响应头

在返回响应之前同样也支持在response headers中去添加响应头进行返回

from fastapi import FastAPI, Response
from fastapi.responses import JSONResponse

app = FastAPI()

@app.get("/header/")
def get_headers(response: Response):
    response.headers["X-Cat-Dog"] = "alone in the world"
    return {"msg": "Hello World"}

@app.get("/headers/")
def get_headers():
    content = {"msg": "Hello World"}
    headers = {"X-Cat-Dog": 'alone', "Content-Language": "zh-CN"}
    return JSONResponse(content=content, headers=headers)

6、CORS（跨域资源共享）

CORS 是 Web 浏览器的一种安全机制，用于限制跨域请求，以防止恶意网站窃取数据。当浏览器发起跨域请求时，会先发送一个预检请求（Preflight Request），即 HTTP OPTIONS 请求，询问目标服务器是否允许该请求。只有当服务器响应的 CORS 头部信息允许跨域请求时，浏览器才会继续发送实际的请求。

• 相关头部信息：
  • Access - Control - Allow - Origin：指定哪些域可以访问资源，可设置为具体域名，如https://frontend.com，也可以是*表示允许所有域。
  • Access - Control - Allow - Methods：指定允许的 HTTP 方法，如GET、POST、PUT等。
  • Access - Control - Allow - Headers：指定允许的请求头部字段。
  • Access - Control - Allow - Credentials：是否允许携带用户凭证，如 Cookies。若值为true，则需要明确指定Access - Control - Allow - Origin，且不能为*。
  • Access - Control - Expose - Headers：指定哪些响应头可以公开给前端。

在 FastAPI 中，需要从fastapi.middleware.cors导入CORSMiddleware来处理 CORS 问题，即from fastapi.middleware.cors import CORSMiddleware。

app.add_middleware(
    CORSMiddleware,
    allow_origins=("*"),
    allow_credentials=True,
    allow_methods=("*"),
    allow_headers=("*")
)

origins = ["https://localhost.tiangolo.com", "http://localhost", "http://localhost:8080"]
app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=("GET", "POST", "PUT", "DELETE"),
    allow_headers=("*")
)