前言

在 Python Web 框架领域，Django、Flask 长期占据主流，但面对高并发、异步场景时，它们的性能瓶颈逐渐显现。FastAPI 作为新一代异步 Web 框架，凭借极致性能（比肩 Node.js/Go）、强类型校验、自动生成接口文档等特性，成为构建高性能 API 的首选。本文从基础入门到生产级实战，全面讲解 FastAPI 的核心用法，帮你快速掌握这一利器。

1. 初识FastAPI

快速	可与Nodejs和Go比肩的极高性能（归功于Pydantic和Starlette），是最快的python web框架之一
高效编码	提高功能开发速度约200%至300%
更少bug	减少约40%的人为导致错误
智能	极佳的编辑器支持，处处皆可自动补全，减少调试时间
简单	设计的易于使用和学习，阅读文档的时间更短
简短	使代码重复最小化，通过不同的参数声明实现丰富功能
健壮	生产可用级别的代码，自动生成的交互式文档

Startlette 负责web部分（Asyncio） 官方文档：https://starlette.dev/

Pydantic 负责数据部分（类型提示） 官方文档：https://pydantic.com.cn/ ，核心中的核心，异步，这是我用django最难受的地方，长查询卡的要死，因此来用fastapi。

由于FastAPI倾向于API，这里说一下API：

目前市面上大部分公司开发人员使用的接口实现规范主要有：Restful，RPC

Restful： 表述性状态转移，一种为web开发而定义的接口设计风格，尤其适用于前后端分离的应用模式中

关键：面向资源开发，因此在定义接口时，客户端访问的URL路径就表示这种要操作的资源数据，而对于数据资源分别使用不同的动作来表达对这个数据的增删改查

2.  quick start

2.1 安装 FastAPI

   Pycharm终端执行：

pip install fastapi

还需要一个ASGI服务器（异步服务器），生产环境可以使用uvicorn

pip install uvicorn

 quick start

from fastapi import FastAPI

app = FastAPI()

@app.get("/")  #路径装饰器
def home():    #路径操作函数
    return {"user_id":100}
@app.get("/shop")
def shop():
    return {"shop":'商店'}

启动命令，Pycharm终端运行

uvicorn 文件名:app --reload  #ctrl + S自重启

文件运行方式

import uvicorn
if __name__ == '__main__':
    uvicorn.run("文件名:app", port=8000, reload=True)

测试
访问： http://127.0.0.1:8000 成功
另外它有另外的文档
http://127.0.0.1:8000/docs    #非常详细

3. 路径操作

3.1 路径装饰器(增删改查)

from fastapi import FastAPI
import uvicorn
app = FastAPI()

@app.get("/get")    #(,tags = ["这是一个get请求"]) 类似于django里的verbose
def get_test():
    return {"method":"get方法"}

@app.post("/post")
def post_test():
    return {"method":"post方法"}

@app.put("/put")
def put_test():
    return {"method":"put方法"}

@app.delete("/delete")
def delete_test():
    return {"method":"delete方法"}

if __name__ == '__main__':
    uvicorn.run("01quickstart:app", port=8000, reload=True)

页面友好表示

#在路径装饰器中有三个参数以帮助api文档的解读
@app.方法("url",tags=["..."],                 #一个标记，类似于django里的verbose
                summary="这是一个测试"，        #标题，在api文档路由里面展示的，不用展开就能看到
       			description="这是一个.....",   #详情，展开后才能看到
       			response_description = "...", #它的响应回复的内容
        		deprecated = True, 			  #是否废弃这个接口
       )

3.2 inlcude_router(路由分发、解耦)

如果在企业级应用的话它会有很多应用，不可能把所有路由都放到同一个文件里面，因此我们应该弄一个路由分发，类似于django的include app(跟django结构一模一样)

首先一个main.py 文件作为主应用

from fastapi import FastAPI
import uvicorn
from app1.urls import app1
app = FastAPI()    #实例化一个对象，api应用

app.include_router(app1, prefix='/app1', tags=["app1登录"]) #prefix: 给子应用加上统一前缀，django里面的path自动实现


if __name__ == '__main__':
    uvicorn.run("01quickstart:app", port=8000, reload=True)

然后创建app包，里面来一个urls

'''app1.urls'''
from fastapi import FastAPI, APIRouter

app1 = APIRouter() #实例化一个对象，路由分发的

@app1.post("/app1/login")
async def login():
    return {'Login': 'login'}

4. 请求与响应（后面都用路由分发，方便测试）

项目结构

4.1 路径参数

'''app2.urls'''
from fastapi import FastAPI, APIRouter

app2 = APIRouter() #实例化一个对象，路由分发的

@app2.get("/app1/user_id/{id}")    
async def user(id):
    print(f'ID    {id}  类型：',type(id))   # 调试查看 
    #注意，返回的统统都是字符串形式,因此要返回指定格式，{id:int},跟正常定义变量差不多，只不过返回过来并且中间加了个:
    return {'ID': 'id}

注意：这里标明一个机制优势，和django不一样，django是约定大于配置，这里FastAPI有一个机制，就是它的路径是有一个优先级的，这里给一个例子：

@app2.get("/app1/user_id/1")
async def user():
	return {'ID':ID}

@app2.get("/app1/user_id/{id}")
async def user(id):
	return {'ID': id}

这里如果我们去调试带传参的那个接口，它是有一个物理性质的优先级的，就是相同的路由的情况下，它会去访问你写在最前面的接口，就是你访问第二个接口如果传参是1的话，它返回的是第一个接口

4.2 查询参数

路径函数中声明不属于路径的其他函数参数，它们将被自动解释为"查询字符串参数"，就是URL？之后用&凤娥的key-value键值对

'''app2.urls'''
from fastapi import FastAPI
app2 = APIRouter()

@app2.get("/jobs")
async def get_jobs(kd, xl, gj):
    #基于kd, xl, gj数据库查询岗位信息
    return {
        "kd":kd,
        "xl":xl,
        "gj":gj
    }

这里已经实现查询参数的接口，但是会有一个问题，必填项和选填项的问题，以及路径参数和查询参数共存的问题

这里如果既需要查询参数又需要路径参数的话，很简单：

@app2.get("/jobs/{kd}") #直接加路径参数就行了
async def get_jobs(kd, xl, gj):
    #基于kd, xl, gj数据库查询岗位信息
    return {
        "kd":kd,
        "xl":xl,
        "gj":gj
    }

如果说既有必填项和选填项：

@app2.get("/jobs") 
async def get_jobs(kd, xl = None, gj = None):			#加个默认值就行了
    #基于kd, xl, gj数据库查询岗位信息
    return {
        "kd":kd,
        "xl":xl,
        "gj":gj
    }

注：自Python3.5开始，PEP484为Python引入了类型注解(type hints) typing 的主要作用有：

1.	类型检查，防止运行时出现参数，返回值类型不符
2.	作为开发文档附加说明，方便使用者调用时传入和返回参数类型
3.	模块加入不会影响程序的运行不会报正式的错误，python支持typing检查错误时会出现黄色警告

type hints 主要指示函数的输入和输出的数据类型，数据类型在typing包中，基本类型有str list dict等等：

async def get_jobs(kd, xl : str, gj : dict):
	pass

Union 是当有多种可能的数据类型时使用，比如函数有可能根据不同情况返回str或返回list，那么就可以写成Union[list,str]：

from typing import Union,Optional
async def get_jobs(kd, xl : Union[str,None] , gj = None): 
#这样声明类型后就不会是可选项了，可以这样Union[str,None]=None

Optional是Union的一个简化，当数据类型中有可能是None时，比如有可能时str或None，则Optional[str]，相当于Union[str,None]：

from typing import Union,Optional
async def get_jobs(kd, xl : Optional[str]=None , gj = None): 

python被人诟病的就是它的这个类型声明，强不强弱不弱的，很迷糊

4.3 请求体

当需要将数据从客户端（例如浏览器）发送API时，将其作为[请求体]发送。请求体是客户端发送给API的数据。响应体是 API 发送给客户端的数据。API几乎总是要发送响应体，但是客户端并不总是需要发送响应体。

fastapi使用 Pydentic 模型来声明响应体，并且能够获得他们所具有的所有能力和优点

FastAPI基于Pydentic，Pydentic主要用来做类型强制检查（数据校验），不符合要求就会抛出异常。

对于API服务，支持类型检查非常有用，会让服务更加健壮，也会加快开发速度，因为开发者再也不用自己一行一行做类型检查。

在这里说一下自己的感悟，任何一个健壮的web后端框架都会是的，你像Django的ORM,Spingboot的MVC，都是这样的，为什么回去学这个东西，可以做一下测试，对于响应速度，Spingboot和Django都去用过，FastAPI去使用Pydantic速度非常快，很夸张，夸张到什么程度，和Django相比，它几乎快到十倍（六到十倍），和Spingboot相当，它是去做大模型的完美载体，也能和Langchain这种大模型工具契合完美。

安装上手

pip install pydantic

'''以用户表为例'''
from pydantic import BaseModel,Field
form fastapi import APIRouter
from datetime import date
from typing import List
app3 = APIRouter()

class User(BaseModel):
    name:str   #加默认值，name:str = 'root'
    age:int	   #加范围， age:int = Field(default = 0, gt = 0, lt = 100) 0-100  Filed - > pydantic里面
    birth:date
    frieds: List[int]
@app3.post("/data")
async def data(user:User):
    '''
    这里还可以对对象进行字典序
    User.dict()
    但是pydantic高版本dict方法已经弃用，用model_dump替代
    还可以指定返回
    user.name,user.birth    #和Django非常相似，但是比Django简单
    '''
    return User

这个模型类还有很多玩法,还有更优雅的写法

birth:Union[str,None] = None
birth:Optional[str] = None
birth:List[int] = []
birth:str|None = None
name:str = Field(regex="^a") #正则表达式，限制它必须以a开头，这里新版的用patten

校验规则的定义（*）

from pydantic import validator,field_validator
class User(BaseModel):
    name:str   #加默认值，name:str = 'root'
    age:int	   #加范围， age:int = Field(default = 0, gt = 0, lt = 100) 0-100  Filed - > pydantic里面
    birth:date
    frieds: List[int]
	@validator('name')    #validator在pydantic高版本中已经被启用，@field_validator适用高版本
    def name_must_alpha(cls,value):
        assert value.isalpha(),'名字必须使用字母'
        return value

模型类的嵌套（*）

class Addr(BadeModel):
	addr:str
	provice:str
class User(BaseModel):
	addr:Addr   #这样就是Addr类型的了
	data:List[Addr] #这是Addr类型的列表

4.4表单数据

在OAuth2 规范的一种使用方式（密码流）中，需要将用户名、密码作为表单字段发送，而不是JSON，FastAPI可以使用Form组件来接收表单数据，需要先使用python-mltipart包

pip install python-multipart

'''以用户表为例'''
from pydantic import BaseModel,Field
form fastapi import APIRouter,Form    #python-multipart包放在fastapi里面了
from datetime import date
from typing import List
app4 = APIRouter()

@app4.post("/regin")
async def data(username:str = Form(), password:str = Form()):   #Form表单格式
    print(f'username:{username}, password:{password}')
    return {
        "username":username
    }

4.5文件上传功能

（a）简单文件上传

from fastapi import File

app5 = APIRouter()

@app5.post("/file")
async def get_file(file: bytes = File()):
    #适合小文件上传
    print("file",file)
    return {
        "file":len(file)
    }
    #目前这种不太常用，很容易把服务器搞崩，大多数用的即使是小文件也是使用文件句柄的方式去获取文件，但是如果你只是上传小的文件可以用这个，因为这个很简单

（b）文件批量上传

from fastapi import File
from typing import List
app5 = APIRouter()

@app5.post("/files")
async def get_files(files: List[bytes] = File()):
    #适合小文件上传
    for file in files:
    	print("file",len(file)
    return {
        "files":len(files)
    }

（c）Upload文件句柄方式上传

from fastapi import UploadFile
 

app1 = APIRouter()
@app1.post("/file")
async def get_file(file: UploadFile):
    print("file", file)
    path = os.path.join("imgs", file.filename)
    with open(path, "wb") as f:      #按句柄写
        for line in file.file:
            f.write(line)
    return {
        "file": file.filename
    }

参数	作用	示例
...（三个点）	必填字段	Form(...) 表示必须提供
default	默认值	Form("guest") 或 Form(default=None)
min_length	字符串最小长度	Form(..., min_length=3)
max_length	字符串最大长度	Form(..., max_length=100)
regex	正则校验	Form(..., regex="^\d+$")
gt / ge	数字大于/大于等于	Form(..., gt=0)
lt / le	数字小于/小于等于	Form(..., lt=100)
alias	别名（用于兼容）	Form(..., alias="user_name")
description	文档描述	Form(..., description="用户名")

文件上传方式使用推荐

文件大小	推荐方案	核心 API	内存占用	优点	缺点
< 100MB	shutil	shutil.copyfileobj	1MB	极简、安全、事务性	同步写入可能轻微阻塞
100MB-2GB	异步分块	aiofiles + 分块	1MB	真异步、高并发	代码稍复杂
> 2GB	分片 + 断点续传	分片上传 + 合并	5MB/片	支持断点续传	需前端配合、实现成本高

（1）Shutil

import shutil
from pathlib import path
from tempfile import NamedTemporaryFile
from fastapi import UploadFile,HTTPException

UPLOAD_DIR = Path("imgs")
UPLOAD_DIR.mkdir(exist_ok=True)

async def save_uplaod_file(upload_file: UploadFile, destination: Path) -> Path:
    '''生产级别的文件保存（基本上用这个没问题了）：自动处理临时文件和错误清理'''
    safe_name = Path(upload_file.filename).name.replace("/","_") #防路径遍历攻击
    file_path = destination / safe_name
     #使用NamedTemporaryFile 保证事务性
    if not upload_file.filename:     #避坑，传空文件名会报错，提前检查
    	raise HTTPException(400, "文件名不能为空")
    suffix = Path(upload_file.filename).suffix   #.suffix:获取文件后缀名
    temp_path = None
    try:   #创建临时文件
        with NamedTemporaryFile(delete=False, suffix=suffix,dir=destination) as tmp:
            tmp_path = Path(tmp.name)
        def _write_sync(): #在后台执行阻塞写入
            with open(tmp_path,"wb") as f:
                shutil.copyfileobj(upload_file.file,f)
        loop = asyncio.get_event_loop()    #获取当前正在运行的异步事件循环
        await loop.run_in_executor(None, _write_sync)  #在后台执行异步阻塞任务
        #这里
        #3.9+:await asyncio.to_thread(_write_sync)
        #原子移动（避免半写入状态被访问，造成数据混乱）
        tmp_path.replace(file_path)
        return file_path
    except Exception as e:
        #自动清理临时文件
        if 'tmp_path' in locals() and tmp_path.exists():
            tmp_path.unlink()
        raise HTTPException(500,f"文件保存失败：{e}") #抛出异常
#使用示例
async def upload_large(file: UploadFile = File(...)):
    path = await save_uplaod_file(file, UPLOAD_DIR)
    return {"size": path.stat().st_size, "filename": path.name}

参考：

NameTemporaryFile：创建临时文件，python临时文件的使用 TemporaryFile,NamedTemporaryFile-CSDN博客

asyncio: 【Python】深入理解 Python 的 asyncio 库-CSDN博客,进程阻塞

问题：使用时如果出现僵尸进程

# 强制终止所有相关进程（必须！）
taskkill /IM uvicorn.exe /F
taskkill /IM python.exe /F

# 等待2秒
Start-Sleep -Seconds 2

# 检查是否还有残留
netstat -ano | findstr :8000

# 如果还有，手动杀PID（将xxxx换成你的实际PID）
taskkill /PID xxxx /F

（2）异步分块（大文件用） 27/2000*1024(响应率)

import asyncio  
import logging  
from pathlib import Path
from tempfile import NamedTemporaryFile
from typing import Optional, Set

from fastapi import UploadFile,HTTPException

# 全局配置（实际项目中用pydanticSettings）
MAX_FILE_SIZE = 2 * 1024 * 1024 * 1024  # 2GB
ALLOWED_EXTENSIONS = {".pad", ".docx", ".zip", ".mp4", ".jpg", ".png"}
UPLOAD_DIR = Path("根路径")
UPLOAD_DIR.mkdir(mode=0o775, exist_ok=True)  #八进制权限模式，exist_ok目录存在就静默跳过，不抛异常

# 日志配置
logger = logging.getLogger(__name__)


class FileSaveError(Exception):
    '''自定义文件保存异常'''
    pass


async def save_upload_file_production(
        upload_file: UploadFile,
        destination: Path,
        *,  #
        max_size: int = MAX_FILE_SIZE,
        allowed_extensions: Optional[Set[str]] = None,  #用Set，查询速度O(1)
        chunk_size: int = 8 * 1024 * 1024,  # 8MB chunks
) -> Path:
    # 参数校验
    if not upload_file.filename:
        raise HTTPException(400, "文件名不能为空")
    filename = upload_file.filename

    # 路径遍历攻击防护
    safe_name = Path(filename).name.replace("/", "_").replace("\\", "_")
    if not safe_name or safe_name.startwith("."):  #startwith("."):检查文件名以.开头的
        raise HTTPException(400, "非法文件名")
    file_path = destination / safe_name

    # 扩展名校验
    if allowed_extensions:
        suffix = Path(safe_name).suffix.lower()
        if suffix not in allowed_extensions:
            raise HTTPException(
                400,
                f"不支持的文件类型：{suffix}，允许：{allowed_extensions}"
            )

    # 检查目录空间（小项目不用做，大项目需要做）
    temp_path: Optional[Path] = None
    try:
        # 创建临时文件（保证事务性）
        with NamedTemporaryFile(
                mode="wb",
                delete=False,
                suffix=Path(safe_name).suffix,
                dir=destination,
                prefix="tep_"
        ) as tmp:
            temp_path = Path(tmp.name)

        # 在后台线程执行阻塞IO（异步中的同步）
        def _blocking_write() -> None:  
            '''在线程池中执行的阻塞操作'''
            bytes_written = 0
            # 使用aiofiles的同步版本，因为已经在后台线程
            with open(temp_path, "wb") as f:
                while chunk := upload_file.file.read(chunk_size):  #流式读取
                    f.write(chunk)
                    bytes_written += len(chunk)

                    # 实时检查大小(防止恶意客户端谎报Content-Length)
                    if bytes_written > max_size:
                        raise FileSaveError(
                            f"文件大小超过限制：{max_size / 1024 / 1024:.2f}MB"
                        )
        # 扔到线程池执行
        loop = asyncio.get_event_loop()  # 获取循环异步事件
        await loop.run_in_executor(None, _blocking_write)  # 阻塞执行
        # 3.9向下兼容， 3.9及以上使用 await asyncio.to_thread(_write_sync)即可

        # 原子移动（类似于mysql事务，防止半写入文件被访问）
        temp_path.replace(file_path)

        # 设置权限（安全加固）
        file_path.chmod(0o644)  #系统兼容，八进制权限

        logger.info(f"文件保存成功：{file_path}")
        return file_path


    except HTTPException:
        # 让FastAPI异常直接透传
        raise
    except FileSaveError as e:
        logger.warning(f"文件保存检查失败: {e}")
        raise HTTPException(400, str(e))
    except Exception as e:
        logger.error(f"文件保存异常: {e}", exc_info=True)
        # 清理临时文件
        if temp_path and temp_path.exists():
            try:
                temp_path.unlink()
                logger.info(f"清理临时文件: {temp_path}")
            except Exception as cleanup_error:
                logger.error(f"清理临时文件失败: {cleanup_error}")
        raise HTTPException(500, f"文件保存失败: {str(e)}")
                    

使用示例：

from fastapi import File, UploadFile, APIRouter
from fastapi.responses import JSONResponse
from save_manager import save_upload_file_production, UPLOAD_DIR

app1 = APIRouter()


@app1.post(
    "/upload/large",
    summary="上传大文件",
    description="支持最大2GB文件，允许类型: pdf, docx, xlsx, zip, mp4, jpg, png",
)
async def upload_large_file(
        file: UploadFile = File(
            ...,
            description="要上传的大文件（最大2GB）"
        ),
):
    """
    大文件上传接口
    - 内存占用恒定（8MB chunks）
    - 线程池隔离阻塞IO
    - 完整安全校验
    """
    try:
        # 使用生产级保存函数
        saved_path = await save_upload_file_production(
            file,
            UPLOAD_DIR,
            max_size=2 * 1024 * 1024 * 1024,
            allowed_extensions={".pdf", ".docx", ".xlsx", ".zip", ".mp4", ".jpg", ".png"},
        )

        return JSONResponse(
            status_code=201,
            content={
                "filename": saved_path.name,
                "size": saved_path.stat().st_size,
                "location": str(saved_path),
                "message": "文件上传成功",
            }
        )
    finally:
        # 确保文件句柄被释放（FastAPI可能不会立即关闭）
        await file.close()

4.6 Reuqest对象

有些情况下我们希望能直接访问Request对象，例如我们在路径操作函数中想获取客户端的IP地址，需要在函数中声明Request类型的参数，FastAPI就回自动传递Request对象给这个参数，我们就可以获取到Request对象及其属性信息，例如header,url,cookie,seesion等

from fastapi import APIRouter,Request


app1 = APIRouter()


@app1.post("/upload")
async def upload_large_file(request: Request):
    print("URL",request.url)
    print("客户端IP地址",request.client.host)
    print("客户端宿主", request.headers.get("user-agent"))
    print("cookies",request.cookies)
    return {
        "URL":request.url,
        "cookies":request.cookies,
        "客户端IP地址":request.client.host,
        "客户端宿主":request.headers.get("user-agent"),
    }

4.7 请求静态文件

在Web开发中，需要请求很多静态文件资源（不是由服务器生成的文件），如css/js和图片文件等

什么是静态文件和动态文件？

from fastapi.staticfiles import StaticFiles

app = FastAPI()
app.mount("/static",StaticFiles(directory = "static"))

这种资源在前后端分离开发中不常用了，知道怎么对接就行，这种访问方式会拖慢服务器

4.8 响应模型相关参数

(1) response_model

前面写的那么多路径函数最终return 的都是自定义结构的字典，FastAPI提供了response_model参数，声明return响应体的模型

response_model实现了路径函数里的路径参数的路径参数 ，说白了就是数据的输入输出，输入的数据进行一次序列化，输出的数据再做一次序列化

from typing import Union

from fastapi import APIRouter, Request
from pydantic import BaseModel, EmailStr

app1 = APIRouter()
'''response_model参数的使用示例'''

class UserIn(BaseModel):
    username: str
    password: str
    email: EmailStr
    full_name: Union[str, None] = None

class UserOut(BaseModel):
    username: str
    email: EmailStr
    full_name: Union[str, None] = None
    
@app1.post("/user", response_model=UserOut)
async def creatuer(user: UserIn):
    return user

(2) response_model_eclude_unset

通过上面的例子，可以知道如何用response_model控制响应体结构，但是如果他们实际上没有存储，则可能要从姐结果中忽略他们，例如，如果model在NoSQL数据库里具有很多可选属性，但是不想发送很长的JSON响应，其中包含默认值，这时候我们需要用到 response_model_eclude_unset参数，设置为开启，它可以帮我们过滤掉那些我们没有的字段

示例：

from typing import Union,List

from fastapi import APIRouter, Request
from pydantic import BaseModel, EmailStr

app1 = APIRouter()

class Item(BaseModel):
    name: str
    description: Union[None, str] = None
    price: float
    tax: float = 10.5
    tags: List[str] = {}

items = {
    "foo": {"name":"Foo","price":50.2},
    "bar":{"name":"Bar","descriptions": "The bartenders" ,"price":50.2, "tax":10.5},
    "baz":{"name":"Baz","descirptions": "The bazers", "price":50.2, "tax":10.5, "tags":[]},
}
@app1.get("/user",response_model=Item,response_model_exclude_unset=True)
async def creatuer(item_id: str):
    return items[item_id]

(3) INCLUDE（白名单【response_model_include】）和EXCLUDE（黑名单【response_model_exclude】）

白名单：写进去的可显示

黑名单：写进去的不显示

5. ORM操作

     在大型的web开发中，我们肯定会用到数据库操作，FastAPI也支持数据库的开发，你可以用PostgreSQL，MySQL、SQLite、Oracle等。以SQLite为例。

fastapi很好，但是缺少一个合适的ORM，官方代码里使用的时sqlalchemy，Tortoise ORM（【Tortoise-ORM】 基础与数据库操作_tortoise orm-CSDN博客）是受Django启发的易于使用的异步ORM（对象关系映射器）

Tortoise ORM目前支持一下数据库：

postgreSQL >=9.4

SQLite(使用aiosqlite)

MySQL/MariaDB(使用aiomysql或者asyncmy)

5.1 ORM与驱动的安装

pip install tortoise-orm    #一定要加orm

'''驱动'''
# 方案一：安装 asyncmy（推荐）
pip install asyncmy

# 方案二：安装 aiomysql
pip install aiomysql

# 或者两个都装，让 Tortoise 自动选择
pip install asyncmy aiomysql

5.2 数据库配置以及使用

'''settings.py'''
TORTOISE_ORM = {
    'connections': {
        'default': {
            'engine': 'tortoise.backends.mysql',
            'credentials': {
                'host': '127.0.0.1',
                'port': 3306,
                'user': 'root',
                'password': '123456',
                'database': 'fastapi',
                'minsize': 1,
                'maxsize': 5,
                'charset': 'utf8mb4',
                'pool_recycle': 3600,  # 防止连接超时
            },
            'echo': True
        },
    },
    'apps': {
        'models': {
            'models': ['models', 'aerich.models'],
            'default_connection': 'default',
        }
    },
    'use_tz': False,
    'timezone': 'Asia/Shanghai'
}

'''使用'''
from fastapi import FastAPI
import uvicorn
from fastapi.staticfiles import StaticFiles

from app1.urls import app1
from settings import TORTOISE_ORM
from tortoise.contrib.fastapi import register_tortoise   #ORM
app = FastAPI()

#fastapi 一旦允许，register_ortoise(说白了就是mysql)就开始允许
register_tortoise(app = app,
                  config = TORTOISE_ORM,
                  generate_schemas = False,#如果数据库为空，则自动生成对应表单，生产环境不要开
                  add_exception_handlers = False, #生产环境不开，会泄露调试信息
                  )
app.include_router(app1, prefix='/app1', tags=["app1登录"])
app.mount("/statics", StaticFiles(directory="statics"), name="statics")

if __name__ == '__main__':
    uvicorn.run("main:app", port=8000, reload=True)

5.3 创建模型

创建models.py

'''示例'''
'''以选课系统为例'''
from tortoise.models import Model
from tortoise import fields



class Student(Model):
    id = fields.IntField(pk=True)
    name = fields.CharField(max_length=100, description="姓名")
    pwd = fields.CharField(max_length = 100, description = "密码")
    sno = fields.IntField(description="学号")
    #一对多
    clas = fields.ForeignKeyField("models.Clas", related_name="students")
    #多对多
    course = fields.ManyToManyField("models.Course", related_name="students")

class Course(Model):
    id = fields.IntField(pk=True)
    name = fields.CharField(max_length=100,description="课程名称")
    teacher = fields.ForeignKeyField("models.Teacher",related_name="teachers")


class Clas(Model):
    name = fields.CharField(max_length=100)

class Teacher(Model):
    id = fields.IntField(pk=True)
    name = fields.CharField(max_length=100)
    pwd = fields.CharField(max_length=100)
    tno = fields.IntField(description="老师编号")

数据库迁移

'''迁移工具'''
pip install aerich
#注意，如果依赖缺失，需要安装
pip install tomlkit

  数据库迁移第一步：初始化配置，只需要使用一次

aerich init -t settings.TORTOISE_ORM   #TORTOISE_ORM配置的位置
'''
初始化完会在当前目录生成一个文件：pyproject.toml和一个文件夹:migrations
pyproject: 保存配置文件路径，低版本可能是aerich.ini
migrations: 存放迁移文件
'''

数据库迁移第二步：初始化数据库，一般情况下只用一次

aerich init-db
'''
1. 此时数据库中就用相应的table
2.如果TORTOISE_ORM配置文件中的models改了名，则执行这条命令时需要增加 --app 参数，来指定你修改的名字
'''

数据库迁移第三步：更新模型并进行迁移

'''修改model类，重新生成迁移文件，比如添加一个字段'''
aerich migrate (--name) (标记修改操作) # aerich migrate --name add_colum
aerich migrate #直接使用也可以
'''进行迁移后生成了迁移文件'''
aerich upgrade   #升级版本
aerich downgrade #回滚版本
aerich history   #查看迁移历史

5.4 api 接口与restful规范

api接口

应用程序编程接口(API接口)，就是应用程序对外提供了一个操作数据的入口，这个入口可以是一个函数或者类方法，也可以是一个url地址或者网络地址，当客户端调用这个入口，应用程序则会执行对应代码操作，给客户端完成相对应的功能

当然，api 接口在工作中是比较常见的开发内容，有时候，我们会调用其他人写的api接口，有时候，我们也需要给其他人提供api接口，由此，我们需要规范一种接口规范，就像http协议一样

restful规范

为了在团队内部形成共识，防止个人习惯差异引起的混乱，我们需要找到一种大家都觉得很好的接口规范，而且这种规范能够让后端写的接口用途一目了然，减少客户端和服务端双方之间的合作成本

目前市面上大部分人使用的接口实现规范主要有：rstful,RPC

REST与技术无关，代表的一种软件架构风格，REST（表征状态转移）

GET 用来获取资源

POST用来新建资源

PUT用来更新资源

DELETE用来删除资源

注意，fastapi进程开太多会导致热重载失效（解决方法）

# 2. 查看所有 uvicorn 进程
Get-Process python | Where-Object { $_.CommandLine -like "*uvicorn*" }

# 3. 强制终止所有相关进程
Stop-Process -Name python -Force

5.4 ORM查询操作

异步： async 和 await ，就是阻塞轮询处理

这里注意，返回对象不会像django还要转一下，这里直接返回字典或者对象pytantic会直接帮你序列化或者反序列化，不用自己处理

示例：

from fastapi import APIRouter
from models import *
app1 = APIRouter()

@app1.get("/student")

async def getALLstudent():    #这里数据过大要分页，不然崩掉了
    '''
    students = await Student.all()    #跟Django一样，类似于列表,但是是一个对像，[Student(),...]
    print("students: ",students)

    for stu in students:    #如果要对toirtoise查询对象做操作，一定要加异步，因为它本身就是支持异步的，因此要在查询操作前加await,在函数前加async
        print(stu.name)
    '''
    #过滤查询
    #students = await Student.filter(name = "jack")

    #get方法
    students = await Student.get(id = 1)   #它的返回值直接就是一个模型类对象

    #模糊查询
    '''
    stu = await Student.filter(sno__gt=2001)  #gt,大于，lt，小于,加e是大于等于
    stu = await Student.filter(sno__range=[2001,30000])   #在这之间的，开区间，__in是闭区间
    print("stu:" ,stu)
    print(students.name)
    '''

    #values查询,将所查询到的QuirySet实例变成由字典组成的列表（实际上还是QuerySet实例，只不过可以看作字典）
    #stus: [{'name': 'jack', 'pwd': '123456', 'sno': 85423119}, {'name': 'mike', 'pwd': '123456', 'sno': 81423112}]
    # stus = await Student.all().values("name","pwd","sno")
    # print("stus:",stus)
    return{

        "Options":"查看所有学生"
    }

@app1.post("/student/{student_id}")

def addStudent(student_id:int):

    return {
        "Options":"添加一个学生"
    }

@app1.put("/student/{student_id}")
def updateStudent(student_id:int):

    return {
        "Options":"修改一个学生信息"
    }

@app1.delete("/student/{student_id}")
def deleteStudent(student_id:int):

    return {
        "Options":"删除一个学生信息"
    }

5.5 ORM添加操作

补充：由于我使用的是mysql8+服务器，用的是caching_sha2_password 认证方式，因此需要cryptography支持
 

pip install cryptography

'''示例'''
from typing import List

from fastapi import APIRouter
from pydantic import BaseModel, validator

from models import *

app1 = APIRouter()
class StudentIn(BaseModel):
    name: str
    pwd: str
    sno: int
    clas_id: int
    courses: List[int] = []
    @validator("name")
    def name_must_not_none(cls, value):
        assert value.isalpha(), "name must be alpha"
        return value
    @validator("sno")
    def sno_must_not_none(cls, value):
        assert value > 1000 and value < 10000, "学号大小要在1000-10000之间"
        return value
@app1.post("/student")
async def addStudent(student_in: StudentIn):
    #方式一：
    # students = Student(name=student_in.name, pwd = student_in.pwd, sno = student_in.sno, clas_id = student_in.clas_id, courses = student_in.courses)
    # await students.save()
    #方式2
    student = await Student.create(name=student_in.name, pwd = student_in.pwd, sno = student_in.sno, clas_id = student_in.clas_id)
    #多对多的关系绑定，注意，在多对多记录创建之前，其他非多对多信息需要已经创建
    choose_course = await Course.filter(id__in=student_in.courses)    #判断哪些课程在这个列表中，说白了就是找选的课程
    await Course.clear()   #清掉这个学生已经选过的课程
    await student.course.add(*choose_course)  #解构展开
    return {
        "Options": "添加一个学生",
        "data":student
    }

5.6 一对多和多对多的ORM查询

from typing import List

from fastapi import APIRouter
from pydantic import BaseModel, validator

from models import *

app1 = APIRouter()


@app1.get("/student")
async def getALLstudent():  # 这里数据过大要分页，不然崩掉了 
	#一对多查询，多对多查询
    jack = await Student.get(name = "jack")
    print(jack.name)
    print(jack.sno)
    print(await jack.clas.values("name"))
    students = await Student.all().values("name","clas__name")
    #多对多查询
    print(await jack.course.all().values("name","teacher__name"))
    #拿所有外键关联
    students = await Student.all().values("name","clas__name","course__name")
    return {
        "students":students,
        "Options": "查看所有学生"
    }

5.7 ORM操作之编辑接口开发

注意，这里全都是按照定义的字段来填的，真实需求的话不可以这样，需要做得细致一点

from typing import List

from fastapi import APIRouter
from pydantic import BaseModel, validator

from models import *

app1 = APIRouter()
class StudentIn(BaseModel):
    name: str
    pwd: str
    sno: str
    clas_id: int
    courses: List[int] = []
    @validator("name")
    def name_must_not_none(cls, value):
        assert value.isalpha(), "name must be alpha"
        return value

@app1.put("/student/{student_id}")
async def updateStudent(student_id: int, student_in: StudentIn):
    #方法一
    #student = await Student.filter(id=student_id).update(name = student_in.name, pwd = student_in.pwd, sno = student_in.sno, clas_id = student_in.clas_id)
    #方法二，简便，但是容易脏读，同时字段顺序需要一一对应，格式需要严谨
    data = student_in.dict() #字典化
    courses = data.pop("courses")  #需要先把多对多的关系去除掉
    await Student.filter(id=student_id).update(**data) #打散就可以了
    #设置多对多的选修课
    edit_stu =  await Student.get(id=student_id)
    choose_course = await Course.filter(id__in=courses)  #去除多对多中间表我们要修改的id
    await edit_stu.course.clear()  #清除掉已有的中间表
    await edit_stu.course.add(*choose_course) #把新的多对多关系加进来
    return {
        "Options": "修改一个学生信息",
        "edit_stu":edit_stu
    }

5.8 ORM操作之删除接口开发

@app1.delete("/student/{student_id}")
async def deleteStudent(student_id: int):
    delete_count = await Student.filter(id = student_id).delete()  #注意，删除，更新在filter里面，尽量用filter
    if not delete_count:
        raise HTTPException(status_code=404,detail = f"主键{student_id}不存在")
    return {
        "Options": "删除一个学生信息",
        "delete_count":delete_count
    }

tortoise_ORM和DjangoORM非常相似，ORM操作是在项目开发中重要节点，当性能较差时，才会使用原生SQL

6. 中间件和CORS组件

6.1 中间件

“中间件”是一个函数，它在每个请求被特定得路径操作之前，以及在每个响应之后工作，中间件得背，而且每个框架得中间件又不一样，

在Django中，它叫钩子函数，在Spingboot里面，叫生态函数，如果想把一个东西做得很高级，中间件必不可少

举个例子，这个东西就是，我想在任何请求之前打印个helloword，如果不用中间件，我要在每个路径函数里加一个print(“hello,world”)

fastapi是一个牛逼得微框架，不想django是“约定大于配置”，fastapi的中间件可以自己去做这玩意，灵活性很高，但是对这种基础的学习很有帮助

6.2 中间件的使用

中间件使用的情况很多，比如域名限制，黑白名单，路由权限限制，响应时间，非常灵活

示例

from datetime import time

from fastapi import FastAPI, Request
import uvicorn
from fastapi.staticfiles import StaticFiles
from fastapi.responses import Response
from app1.urls import app1
from settings import TORTOISE_ORM
from tortoise.contrib.fastapi import register_tortoise  # ORM

app = FastAPI()

# fastapi 一旦允许，register_ortoise就开始允许
register_tortoise(app=app,
                  config=TORTOISE_ORM,
                  generate_schemas=False,  # 如果数据库为空，则自动生成对应表单，生产环境不要开
                  add_exception_handlers=False,  # 生产环境不开，会泄露调试信息
                  )
app.include_router(app1, prefix='/app1', tags=["app1登录"])
app.mount("/statics", StaticFiles(directory="statics"), name="statics")


@app.middleware("http")
async def m2(request: Request, call_next):
    # 请求代码块
    print("m2 request")
    start = time.time()
    response = await call_next(request)
    # 响应代码块
    print("m2 response")
    end = time.time()
    response.headers["ProcessTimer"] = str(end - start)  # 测试响应时间
    return response


@app.middleware("http")
async def m1(request: Request, call_next):
    # 请求代码块
    print("m1 request")
    # if request.client.host in ['127.0.0.1'] :  #IP地址黑名单
    #     return Response(status_code=403,content = "黑名单")  #from fastapi.responses import Response
    # if request.url.path in ["/app1/student"]:   #路径访问限制
    #     return Response(content="visti stop")
    response = await call_next(request)
    # 响应代码块
    print("m1 response")
    return response  # 它响应的那个数据是下一个的请求体


if __name__ == '__main__':
    uvicorn.run("01quickstart:app", port=8000, reload=True)

7. CORS组件实现跨域请求

      跨域（协议，端口，ip三者只要有一个不同）请求，就是让浏览器建立连接，我们要通过浏览器发送请求或响应，这是将我们所学技术变现的最后一步，让其他浏览器客户端可以访问（有公网的前提下），只有跨域请求之后，人家的浏览器才会允许你向他发送请求，人家浏览器认为你是安全的

首先构建一个最基础的HTML界面

这里用axios

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>测试 GET 请求</title>
    <script src="https://cdn.bootcdn.net/ajax/libs/jquery/3.6.0/jquery.js"></script>
</head>
<body>
<p>点击我发送 GET 请求</p>

<script>
    $("p").click(function () {
        $.ajax({
            url: "http://127.0.0.1:8000/app1/student",
            method: "GET",
            success: function (res) {
                alert("请求成功！");
                console.log(res); // 打印后端返回的数据
            },
            error: function (xhr) {
                alert("请求失败！");
                console.error(xhr.responseText);
            }
        });
    });
</script>
</body>
</html>

可以看到：

Access to XMLHttpRequest at 'http://127.0.0.1:8000/app1/student' from origin 'http://localhost:63342' has been blocked by CORS policy: No 'Access-Control-Allow-Origin' header is present on the requested resource.

       前端请求会出现报错信息，我们缺少Access-Control-Allow-Origin这个请求头，这就是CORS请求头，即同源请求头

这里我先自己通过中间件实现一个CORS请求，由以上可知，我们需要一个Access-Control-Allow-Origin请求头，Access-Control-Allow-Origin请求头的格式是，Access-Control-Allow-Origin = ？ ，因此，可以这样实现

@app.middleware("http")
async def MyCORSMiddleware(request: Request, call_next):
     response = await call_next(request)
     response.headers["Access-Control-Allow-Origin"] = "*"   #加一个头，CORS同源请求
     return response

对CORS，fastapi有自己的完整封装库，它在实现上非常细致，以上自定义的请求非常粗糙，真实的CORS又要涉及到很多细节，包括分片请求，滑动请求，同步请求等等，这里fastapi已经帮我们写好了，可以直接用

from fastapi.middleware.cors import CORSMiddleware
from fastapi import FastAPI
app = FastAPI()
app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,  #*代表所有客户端 ,[*]也可以
    allow_credentials=True, #要不要走认证
    allow_methods=["*"], #允许通过什么方法来访问我
    allow_headers=["*"], #允许客户端带自定义的请求头
)

8. 完结撒花

对fastapi框架的学习到此为止，但想要用好fastapi，我们需要有更多的学习：

项目名称	星标数	技术栈	适合场景	学习难度
full-stack-fastapi-template	35.4K	FastAPI+React+Docker	全栈开发、快速启动	⭐⭐⭐
FastAPI-Template	持续增长	FastAPI+RBAC+安全	企业级后端、权限系统	⭐⭐⭐⭐
fastapi-production-template	1K+	FastAPI+监控+部署	生产环境、DevOps	⭐⭐⭐⭐
Django官方项目	72.8K	Django	基础学习	⭐⭐
FuAdmin	不详	Django/FastAPI	架构参考	⭐⭐⭐