• 日期: 2026-05-30
  • 标签: 程序员

一.FastAPI基础

1.FastAPI概述

FastAPI是一个基于python的高性能Web框架，专用于快速构建API接口服务
图片
这里同步与异步应该与单线程和多线程差不多
图片
图片
Pydantic类型提示与验证，可以通过:类型说明变量的类型
图片
还包含了可交互式文档，可以进行测试API
图片
总结：FastAPI的好处有异步性能高，开发效率高，自动生成文档。

2.第一个FastAPI程序

from fastapi import FastAPI

#创建FastAPI实例
app=FastAPI()

#路由将根目录和root函数联系起来
@app.get("/")
async def root():
    return {"message":"Hello World"}

@app.get("/hello/{name}")
async def say_hello(name:str):
    return {"message":f"hello {name}"}

# 添加启动代码
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=8000)

图片
图片
图片

3.路径参数与查询参数与请求体参数

3.1路径参数和Path注解

图片

from fastapi import FastAPI

#创建FastAPI实例
app=FastAPI()

#装饰器将根目录和root函数联系起来
@app.get("/")
async def root():
    return {"message":"Hello World"}

@app.get("/book/{id}")
async def get_book(id:int):  #这里的id对应url传入的id，采用冒号的形式解释什么类型
    return {"id":id,"title":f"这是第{id}本书"}

# 添加启动代码
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=8000)

图片
Path注解
图片

from fastapi import FastAPI,Path

#创建FastAPI实例
app=FastAPI()

#路由将根目录和root函数联系起来
@app.get("/")
async def root():
    return {"message":"Hello World"}

@app.get("/book/{id}")
async def get_book(id:int=Path(...,gt=0,lt=101,description="书籍id，取值1-100")):  #这里的id对应url传入的id，采用冒号的形式解释什么类型
    return {"id":id,"title":f"这是第{id}本书"}

@app.get("/author/{name}")
async def get_name(name:str=Path(...,min_length=2,max_lengt=10)):
    return {"msg":f"这是{name}的信息"}


# 添加启动代码
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=8000)

3.2查询参数和Query注解

声明的参数不是路径参数时，路径操作函数会把该参数自动转化为查询参数

from fastapi import FastAPI,Query

#创建FastAPI实例
app=FastAPI()

#路由将根目录和root函数联系起来
@app.get("/")
async def root():
    return {"message":"Hello World"}

#需求 查询新闻 -> 分页 ,skip：跳过的记录数，limit：返回的记录数
@app.get("/news/news_list")
async def get_news_list(
        skip:int=Query(0,description="跳过的记录数",lt=10),  #这里的默认值是0和10
        limit:int=Query(10,description="返回的记录数")
):
    return {"skip":skip,"limit":limit}


# 添加启动代码
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=8000)

图片

3.3请求体参数和Field注解

图片
这里是post或put方式，而上面的查询参数属于get方式
图片

from fastapi import FastAPI
from pydantic import BaseModel

#创建FastAPI实例
app=FastAPI()

#路由将根目录和root函数联系起来
@app.get("/")
async def root():
    return {"message":"Hello World"}

#注册：用户名和密码 -> str
class User(BaseModel):
     username:str
     password:str

#请求方式改为post
@app.post("/register")
async def register(user:User):
    return user


# 添加启动代码
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=8000)

图片
Field注解
图片

from fastapi import FastAPI
from pydantic import BaseModel,Field

#创建FastAPI实例
app=FastAPI()

#路由将根目录和root函数联系起来
@app.get("/")
async def root():
    return {"message":"Hello World"}

#注册：用户名和密码 -> str
class User(BaseModel):
     username:str=Field(default="张三",min_length=2,max_length=10,description="用户名，长度要求2-10个字")
     password:str=Field(min_length=3,max_length=20)

#请求方式改为post
@app.post("/register")
async def register(user:User):
    return user


# 添加启动代码
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=8000)

图片

4.响应类型

4.1json

from fastapi import FastAPI,HTTPException

#创建FastAPI实例
app=FastAPI()

#路由将根目录和root函数联系起来
@app.get("/")
async def root():
    return {"message":"Hello World"}

# 添加启动代码
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=8000)

4.2HTML

from fastapi import FastAPI
from fastapi.responses import HTMLResponse

#创建FastAPI实例
app=FastAPI()

#路由将根目录和root函数联系起来
@app.get("/")
async def root():
    return {"message":"Hello World"}

#接口 -> 响应 HTML代码
@app.get("/html",response_class=HTMLResponse)
async def get_html():
    return "
这是一级标题
"


# 添加启动代码
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=8000)

4.3文件

from fastapi import FastAPI
from fastapi.responses import FileResponse

#创建FastAPI实例
app=FastAPI()

#路由将根目录和root函数联系起来
@app.get("/")
async def root():
    return {"message":"Hello World"}

#接口：返回一张图片内容
@app.get("/file")
async def get_file():
    path="./file/1.jpg"
    return FileResponse(path)


# 添加启动代码
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=8000)

4.4自定义响应数据格式

from fastapi import FastAPI
from pydantic import BaseModel

#创建FastAPI实例
app=FastAPI()

#路由将根目录和root函数联系起来
@app.get("/")
async def root():
    return {"message":"Hello World"}

#需求：新闻接口 ->  响应数据类型 id title content
class News(BaseModel):
    id:int
    title:str
    content:str

@app.get("/news/{id}",response_model=News)
async def get_news(id:int):
    return {
        "id":id,
        "title":f"这是第{id}本书",
        "content":"这是一本好书"
    }


# 添加启动代码
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=8000)

5.异常处理

from fastapi import FastAPI,HTTPException

#创建FastAPI实例
app=FastAPI()

#路由将根目录和root函数联系起来
@app.get("/")
async def root():
    return {"message":"Hello World"}

#序求：按 id 查询新闻-> 1-6
@app.get("/news/{id}")
async def get_news(id:int):
    id_list=[1,2,3,4,5,6]
    if id not in id_list:
        raise HTTPException(status_code=404,detail="查找的新闻不存在")
    return {"id":id}

# 添加启动代码
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=8000)

二.FastAPI进阶

1.中间件

@app.middleware("http")
async def middleware1(request,call_next):
    print("中间件1 start")
    response=await call_next(request)
    print("中间件1 end")
    return response

2.依赖注入

from fastapi import FastAPI,Query,Depends  #2.导入Depends

#创建FastAPI实例
app=FastAPI()

#路由将根目录和root函数联系起来
@app.get("/")
async def root():
    return {"message":"Hello World"}

#分页参数逻辑共用：新闻列表和用户列表
#1.依赖项
async def common_parameters(
        skip:int=Query(0,gr=0),
        limit:int=Query(0,le=60)
):
    return {"skip":skip,"limit":limit}

@app.get("/news/news_list")
async def get_news_list(commons=Depends(common_parameters)):
    return commons

@app.get("/user/user_list")
async def get_user_list(commons=Depends(common_parameters)):
    return commons

# 添加启动代码
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=8000)

3.ORM

安装：pip install sqlalchemy[asyncio] aiomysql

3.1ORM建表

import datetime

from fastapi import FastAPI
from sqlalchemy import DateTime, func, String, Float
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.orm import DeclarativeBase,Mapped,mapped_column

#创建FastAPI实例
app=FastAPI()

#路由将根目录和root函数联系起来
@app.get("/")
async def root():
    return {"message":"Hello World"}

#1.创建异步引擎
ASYNC_DATABASE_URL="mysql+aiomysql://root:123456@localhost:3306/FastAPI_first"
async_engine=create_async_engine(
    ASYNC_DATABASE_URL,
    echo=True,     #可选，输出SQL日志
    pool_size=10,  #设置连接池活跃的连接数
    max_overflow=20  #允许额外的连接数
)

#2.定义模型类：基类+表对应的模型类
#基类：创建时间，更新时间；书籍表：id，书名，作者，价格，出版社
class Base(DeclarativeBase):
    create_time:Mapped[datetime]=mapped_column(DateTime,insert_default=func.now(),default=func.now,comment="创建时间")
    update_time:Mapped[datetime]=mapped_column(DateTime,insert_default=func.now(),default=func.now,onupdate=func.now(),comment="修改时间")


class Book(Base):
    __tablename__="book"
    id:Mapped[int]=mapped_column(primary_key=True,comment="书籍id")
    bookname:Mapped[str]=mapped_column(String(255),comment="书名")
    author:Mapped[str]=mapped_column(String(255),comment="作者")
    price:Mapped[float]=mapped_column(Float,comment="价格")
    publisher:Mapped[str]=mapped_column(String(255),comment="出版社")

#3：建表：定义函数建表 -> FastAPI 启动的时候调用建表的函数
async def create_tables():
    #获取异步引擎
    async with async_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)  #Base 模型类的元数据创建

@app.on_event("startup")
async def startup_event():
    await create_tables()

# 添加启动代码
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=8000)

原文地址: https://www.cveoy.top/t/topic/qGL3 著作权归作者所有。请勿转载和采集!

免费AI点我,无需注册和登录