Python框架篇(8):FastApi-文件处理_python annotated

1. 参数接收补充

在之前的文章Python框架篇(2):FastApi-参数接收和验证学习中，忘了以下几种参数的接收，这里补充下：

1.1 注解Annotated

typing.Annotated 是 Python 标准库中 typing 模块提供的一个工具，用于在类型提示中添加元数据（metadata）。它可以帮助开发者更清晰地描述变量或函数的含义和用途,使用基本语法如下：

Annotated[type, metadata1, metadata2, ...]
1

其中，type 是变量或函数的类型，而 metadata1, metadata2, ... 是元数据，可以是任何类型的对象，如字符串、字典等。

更多文档可查看: https://docs.python.org/zh-tw/3/library/typing.html#typing.Annotated

1.2 cookie参数

1.代码清单

文件: app/router/param_router.py

from typing import Annotated
from fastapi import APIRouter, Request
from fastapi import Cookie # 导入cookie

router = APIRouter(prefix="/param", tags=["更多参数接收示例"])

@router.get("/cookie/key", summary="接收cookie中指定的key")
async def cookieKey(user_name: Annotated[str | None, Cookie()] = None):
    """接收cookie中指定的key"""
    return {"user_name": user_name}


@router.get("/cookie/all", summary="所有cookie值")
async def cookieParams(request: Request):
    """接收cookie值"""
    return {"cookies": request.cookies}
1

2.请求验证

# 命令行请求
$ curl -X 'GET' \
  'http://0.0.0.0:8088/param/cookie/all' \
  -H 'accept: application/json' \
  -H 'Cookie: user_name=liuqinghui;user_id=110;user_email=test@163.com'
 
# 响应
{"cookies":{"user_name":"liuqinghui","user_id":"110","user_email":"test@163.com"}}
1

@注意: 发现框架生成的SwaggerUI演示cookie传参会报错，这里以命令行的形式展示~

1.3 header参数

默认情况下, Header会把参数中的下划线 (_) 转换为连字符 (-) ,而且大小写不敏感，因此在代码种可以使用x_platform接收Header中的X-PlatForm值。

1.代码清单

文件: app/router/param_router.py

from typing import Annotated
from fastapi import APIRouter, Request
from fastapi import Header # Header

router = APIRouter(prefix="/param", tags=["更多参数接收示例"])
...

@router.get("/header/key")
async def headerKey(x_platform: Annotated[str | None, Header()] = None):
    """ 从header中获取指定key"""
    return {"x_platform": x_platform}
  
@router.get("/header/keys")
async def headerKey(x_ip: Annotated[list[str] | None, Header()] = None):
    """ 从header中获取重复key的值"""
    return {"x_ip": x_ip}
1

2.请求验证

# 请求验证
$ curl -X 'GET' \
  'http://0.0.0.0:8088/param/header/key' \
  -H 'accept: application/json' \
  -H 'X-PlatForm: wechat'

# 返回值
{"x_platform":"wechat"}

# ----- headers中有重复key时 ----- 
$ curl -X 'GET' \
  'http://0.0.0.0:8088/param/header/keys' \
  -H 'accept: application/json' \
  -H 'X-IP: 127.0.0.1' \
  -H 'X-IP: 172.30.10.21' 
  
# 返回
{"x_ip":["127.0.0.1","172.30.10.21"]}
1

1.4 表单参数

在使用表单参数前，需要先安装对应的包:pip install python-multipart。

1.代码清单

文件: app/router/param_router.py

from fastapi import FastAPI, Form #导入包
from app.types import response #自定义包

router = APIRouter(prefix="/param", tags=["更多参数接收示例"])

...
@router.post("/form/key")
async def formKey(username: str = Form(), password: str = Form()) -> response.HttpResponse:
    """ 接收表单中的参数"""
    body = {
        "username": username,
        "password": password
    }
    return response.ResponseSuccess(body)
1

2.请求验证

2.上传文件

2.1 代码清单

import os
from fastapi import APIRouter, Cookie, Request, Header, Form, UploadFile
from app.types import response
...

@router.post("/upload/file")
async def uploadFile(file: UploadFile | None = None, fileType: str = Form()) -> response.HttpResponse:
    """ 文件上传"""
    if not file:
        return response.ResponseFail("文件信息不能为空~")

    try:
        # 构造保存目录
        save_path = os.path.join(os.getcwd(), "tmp", fileType)
        # 不存在则创建目录
        os.makedirs(save_path, exist_ok=True)
        # 拼接文件全路径
        file_path = os.path.join(save_path, file.filename)
        # 读取文件内容并写入目标文件
        contents = await file.read()
        with open(file_path, "wb") as f:
            f.write(contents)
        body = {
            "fileName": file.filename,
            "fileType": fileType,
            "size": file.size,
        }
        return response.ResponseSuccess(body)
    except Exception as e:
        return response.ResponseFail("文件上传失败:" + str(e))
1

2.2 请求验证

$ curl -X 'POST' \
  'http://0.0.0.0:8088/param/upload/file' \
  -H 'accept: application/json' \
  -H 'Content-Type: multipart/form-data' \
  -F 'file=@/Users/hui/Downloads/computer.png' \ #这里是本地绝对路径，@
  -F 'fileType=img'
1

2.3 保存结果

3.访问文件

在某些场景下，我们需要提供一个地址，可以让前端工程师或者第三方来访问静态资源，比如返回一张图片或者一个文件。在FastAPI中，静态资源的访问实现，叫:挂载。

3.1 静态目录

...
├── app
├── main.py
...
├── static
│   ├── img
│   │   └── test.jpg
│   └── root.txt
...
1
2
3
4
5
6
7
8
9
10

3.2 挂载

在服务启动入口，加上下面server.mount(...)这行代码，即可进行静态资源进行访问。

...
# 实例化
server = FastAPI(redoc_url=None, docs_url="/apidoc", title="FastAPI学习")
# 挂载静态资源目录
server.mount("/static", StaticFiles(directory="static"), name="static")
...
1

3.3 访问

比如下面访问静态资源图片: static/img/test.jpg

4.下载文件

在工作中下载文件的场景，大部分都是异步导出文件，如发个异步任务，查出数据并把结果上传到oss，然后在下载列表中查看状态并下载结果。但是也有些特殊场景，需要直接在浏览器中，实现下载文件，下面是实现效果：

4.1 代码清单

文件: app/router/param_router.py

from fastapi.responses import FileResponse # 导入包
...


@router.get("/file/download")
async def downloadFile() -> FileResponse:
    """下载文件"""
    fileName = "test.pdf"
    file_path = os.path.join(os.getcwd(), "tmp", fileName)
    return FileResponse(file_path, filename=fileName)
1

4.2 请求验证