目录

创建项目

创建目录

mkdir project
cd project

创建虚拟环境

python -m venv env
source env/bin/activate
# 退出命令 deactivate

创建 requirements.txt 文件

fastapi
python-multipart
aiofiles
uvicorn
gunicorn

安装需要的库

pip install -r requirements.txt -i https://mirrors.aliyun.com/pypi/simple/

项目目录结构

project
├── app
│   ├── __init__.py
│   ├── dependencies.py
│   ├── main.py
│   └── routers
│       ├── __init__.py
│       ├── files.py
│       └── users.py
└── requirements.txt

main.py

import uvicorn

from fastapi import FastAPI

from .routers import users
from .routers import files


app = FastAPI(title='REST API Interface', version='1.0', 
              description="基于 FastAPI 的 REST API 接口。")


app.include_router(users.router)
app.include_router(files.router)


@app.get('/')
async def index():
    return {'Hello': 'World!'}


if __name__ == '__main__':
    uvicorn.run(app, host='0.0.0.0', port=8000)

dependencies.py

import inspect

from typing import Type
from pydantic import BaseModel
from fastapi import Form


def as_form(cls: Type[BaseModel]):
    new_parameters = []

    for field_name, model_field in cls.__fields__.items():
        model_field: ModelField  # type: ignore

        new_parameters.append(
             inspect.Parameter(
                 model_field.alias,
                 inspect.Parameter.POSITIONAL_ONLY,
                 default=Form(...) if model_field.required else Form(model_field.default),
                 annotation=model_field.outer_type_,
             )
         )

    async def as_form_func(**data):
        return cls(**data)

    sig = inspect.signature(as_form_func)
    sig = sig.replace(parameters=new_parameters)
    as_form_func.__signature__ = sig  # type: ignore
    setattr(cls, 'as_form', as_form_func)
    return cls

routers/users.py

from typing import Optional
from fastapi import APIRouter, Form
from pydantic import BaseModel

from ..dependencies import as_form


router = APIRouter(tags=['Users'])


@as_form
class User(BaseModel):
    name: str
    age: Optional[int] = None


@router.post("/users", tags=['User'])
async def create_user(name: str = Form(...), age: Optional[int] = Form(None)):
    return User(name=name, age=age)


@router.put("/users/{name}", tags=['User'])
async def update_user(name: str, age: Optional[int] = Form(None)):
    return User(name=name, age=age)


@router.post("/users_by_json", tags=['User'])
async def create_user(user: User):
    return user


@router.put("/users_by_json/{name}", tags=['User'])
async def update_user(name: str, user: Optional[User] = None):
    return user


@router.get("/users/{name}", tags=['User'])
async def read_user(name: str):
    return User(name=name)


@router.get("/users", tags=['User'])
async def read_user(name: str, age: Optional[int] = None):
    return User(name=name, age=age)

routers/files.py

import json
import tempfile
import aiofiles

from enum import Enum

from fastapi import APIRouter
from fastapi import Depends, Form, UploadFile
from fastapi.responses import StreamingResponse

from starlette.requests import Request

from .users import User


CHUNK_SIZE = 262144 #256*1024


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


@router.post('/upload/file', summary='上传速度快 binary')
async def create_file(request: Request):
    file_path = f'test/image.jpg'
    with open(file_path, "wb") as file:
        async for chunk in request.stream():
            file.write(chunk)
    return {'file_path': file_path}


@router.post('/upload/tempfile', summary='上传文件，存储为临时文件 binary')
async def create_tempfile(request: Request):
    with tempfile.NamedTemporaryFile() as file:
        file_path = file.name
        async for chunk in request.stream():
            file.write(chunk)
        file.flush()
        
    return {'file_path': file_path}


@router.post("/upload/files", summary='上传多文件 form-data')
async def create_files(files: list[UploadFile]):
    """
    Form Data: https://fastapi.tiangolo.com/tutorial/request-forms/
    """

    file_paths = []
    for file in files:
        file_path = f'test/{file.filename}'
        file_paths.append(file_path)

        async with aiofiles.open(file_path, 'wb') as f:
            content = await file.read()
            await f.write(content)

    return {'file_path': file_paths}


@router.post('/upload/file_and_json', summary='上传文件和JSON数据 form-data')
async def create_uploadfile_and_json(file: UploadFile, data: str = Form(...)):
    file_path = f'test/{file.filename}'
    async with aiofiles.open(file_path, 'wb') as f:
        content = await file.read()
        await f.write(content)

    d = json.loads(data)
    d['file_path'] = file_path

    return d


@router.post('/upload/file_and_model', summary='上传文件和模型对象 form-data')
async def create_uploadfile_and_model(file: UploadFile, user: User = Depends(User.as_form)):
    """
    下面的参考链接还有点问题，User.age 是可选参数，但是请求需要设置，不然出错: 422 Unprocessable Entity。这里已经修复了。
    https://stackoverflow.com/questions/60127234/how-to-use-a-pydantic-model-with-form-data-in-fastapi
    """
    file_path = f'test/{file.filename}'
    async with aiofiles.open(file_path, 'wb') as f:
        content = await file.read()
        await f.write(content)

    return user


@router.get('/download/file', response_class=StreamingResponse, summary='下载速度快')
async def read_file():
    file_path = 'app/assets/image.jpg'

    async def iterfile(file_path, chunk_size):
        async with aiofiles.open(file_path, 'rb') as f:
            while chunk := await f.read(chunk_size):
                yield chunk

    return StreamingResponse(iterfile(file_path, CHUNK_SIZE), media_type="application/octet-stream")


class MediaType(str, Enum):
    jpg = 'jpg'
    png = 'png'
    mp4 = 'mp4'

    def get_type(self):
        types = {self.jpg: 'image/jpeg', self.png: 'image/png', self.mp4: 'video/mp4'}
        return types[self]

    def get_file(self):
        files = {self.jpg: 'image.jpg', self.png: 'image.png', self.mp4: 'video.mp4'}
        return files[self]


@router.get('/{media_type}', response_class=StreamingResponse)
async def read_media_file(media_type: MediaType):
    """
    MIME 类型: https://developer.mozilla.org/zh-CN/docs/Web/HTTP/Basics_of_HTTP/MIME_types
    HTTP 响应代码: https://developer.mozilla.org/zh-CN/docs/Web/HTTP/Status
    """

    file_path = f'app/assets/{media_type.get_file()}'

    async def iterfile(file_path, chunk_size):
        async with aiofiles.open(file_path, 'rb') as f:
            while chunk := await f.read(chunk_size):
                yield chunk

    return StreamingResponse(iterfile(file_path, CHUNK_SIZE), media_type=media_type.get_type())

运行

调试

uvicorn app.main:app --reload

uvicorn

uvicorn app.main:app --host 0.0.0.0 --workers 4

gunicorn + uvicorn

gunicorn app.main:app --bind 0.0.0.0 --workers 4 --worker-class uvicorn.workers.UvicornWorker

参考资料