FastAPI教程(一)_app = fastapi()-CSDN博客

本文链接：https://blog.csdn.net/simon_skywalker/article/details/146808364

FastAPI

简单应用

from fastapi import FastAPI

app = FastAPI()
# app是表示整个Web应用程序的顶级FastAPI对象

# 在此服务器上对URL“/hi”的请求应定向到以下函数。
# 这个装饰器只适用与get请求
@app.get("/hi")
def greet():
    return "Hello? World?"

if __name__ == "__main__":
    # 引入web服务器
    # 默认使用8000端口
    import uvicorn
    # reload参数用于在文件发生更改时重启服务器
    uvicorn.run("hello:app", reload=True)

    # 直接在浏览器访问
    # 访问http://localhost:8000/hi

    # 使用request访问
    import requests
    r = requests.get("http://localhost:8000/hi")
    print(r.json())
    # 直接在浏览器访问
    # 访问http://localhost:8000/hi

    # 使用request访问
    import requests
    r = requests.get("http://localhost:8000/hi")
    print(r.json())
    
    # 使用Httpie测试
    # 默认方法 会返回请求体
    $ http localhost:8000/hi
    # 跳过响应头 只输出响应体
    $ http -b localhost:8000/hi
    # 获得完整的响应头
    $ http -v localhost:8000/hi

HTTP Requests

URL Path

一个http请求格式

GET /hi HTTP/1.1 
Accept: /
Accept-Encoding: gzip, deflate
Connection: keep-alive
Host: localhost:8000
User-Agent: HTTPie/3.2.1

#FastAPI使用了依赖注入，需要的单数都可以在path函数中声明和提供
# 利用路径传入参数
from fastapi import FastAPI

app = FastAPI()

# 将who变量的值注入 然后输出
@app.get("/hi/{who}")
def greet(who):
    return f"Hello? {who}?"

if __name__ == "__main__":
    # 引入web服务器
    # 默认使用8000端口
    import uvicorn
    # reload参数用于在文件发生更改时重启服务器
    uvicorn.run("hello:app", reload=True)

# 在另一个窗口运行
import requests

r = requests.get("http://localhost:8000/hi/simon")
print(r.json())
# 最后的响应是一个JSON字符串
# 输出 Hello? simon?

Query Parameters
# 利用查询参数传递参数
from fastapi import FastAPI

app = FastAPI()

# 定义了一个叫做who的参数
@app.get("/hi")
def greet(who):
    return f"Hello? {who}?"

if __name__ == "__main__":
    # 引入web服务器
    # 默认使用8000端口
    import uvicorn
    # reload参数用于在文件发生更改时重启服务器
    uvicorn.run("hello:app", reload=True)
    
# 在路径中直接写入变量    
import requests
requests.get("http://localhost:8000/hi?who=simon")
r.json()

# 使用字典传入参数
params = {"who": "Mom"}
r = requests.get("http://localhost:8000/hi", params=params)

Body

# 我们可以向GET端点提供路径或查询参数，但不能提供来自请求体的值。在HTTP中，GET应该是幂等的--一个计算机术语，意思是问同样的问题，得到同样的答案。HTTP GET应该只返回内容。请求体用于在创建（POST）或更新（PUT或PATCH）时向服务器发送内容。

# 使用post 在请求体中传递参数
from fastapi import FastAPI, Body

app = FastAPI()

@app.post("/hi")
def greet(who:str = Body(embed=True)):
    return f"Hello? {who}?"


if __name__ == "__main__":
    import uvicorn
    uvicorn.run("hello:app", reload=True)

import requests
# 使用json参数来传递JSON编码数据。
r = requests.post("http://localhost:8000/hi", json={"who": "Mom"})

HTTP Header

from fastapi import FastAPI, Header

app = FastAPI()

# 在请求头中传递参数
@app.post("/hi")
def greet(who:str = Header()):
    return f"Hello? {who}?"
# $ http -v localhost:8000/hi who:Mom
# 返回请求头中内容
from fastapi import FastAPI, Header# 最后返回的是请求头中的

app = FastAPI()

@app.post("/agent")
def get_agent(user_agent:str = Header()):
    return user_agent
# 最后返回的是请求头中的user-agent内容
# FastAPI对这些内容进行了一些转化 从-到_

同一个路径函数中可以使用多个方法，可以同时从URL、查询参数、HTTP体、HTTP头、cookies等位置获取数据。

在URL中传递参数时，遵循RESTful准则是标准做法。
查询字符串通常用于提供可选参数，如分页。
body通常用于较大的输入，如整体或部分模型。

Http Response

默认情况下，FastAPI将从端点函数返回的任何内容转换为JSON。

Status Code

# 添加状态码
@app.get("/happy")
def happy(status_code=200):
    return ":)"

Headers

from fastapi import FastAPI, Body
from fastapi import Response
app = FastAPI()
# 注入响应头
@app.get("/header/{name}/{value}")
def header(name: str, value: str, response:Response):
    response.headers[name] = value
    return "normal body"

if __name__ == "__main__":
    import uvicorn
    uvicorn.run("hello:app", reload=True)
# localhost:8000/header/simon/skywalker

Response Types

常见的返回类型包括

JSONResponse (默认值)
HTMLResponse
PlainTextResponse
RedirectResponse
FileResponse
StreamingResponse

with语句

# with用于简化资源管理的上下文管理器语法结构，它能够确保代码块执行前后正确地获取和释放资源。
# with语句常见操作
# 文件操作
with open('file.txt', 'r') as f:
    content = f.read()
    # 文件会在with块结束后自动关闭

# 锁操作
with lock:
    # 临界区代码
    # 锁会在with块结束后自动释放

# 数据库连接
with connection:
    # 数据库操作
    # 连接会在with块结束后自动关闭或归还连接池

# 进行异常处理    
with pytest.raises(ValueError):
    # 期望这里的代码抛出ValueError异常
    int('not a number')

响应模型

from datetime import datetime
from pydantic import BaseClass

# 仅包含用户标签名
# 用于验证客户端请求数据
class TagIn(BaseClass):
    tag: str

# 内部模型
# 包含完整信息，包括敏感数据 (secret) 
class Tag(BaseClass):
    tag: str
    created: datetime
    secret: str

# 输出模型
# 返回必要信息 排除敏感信息
class TagOut(BaseClass):
    tag: str
    created: datetime

import datetime
from fastapi import FastAPI
from model.tag import TagIn, Tag, TagOut
import service.tag as service

app = FastAPI()

@app.post('/')
# 接受请求体中的TagIn对象
# 返回类型注解 -> TagIn
def create(tag_in: TagIn) -> TagIn:
    # tag: Tagt、ag_in: TagIn使用了参数类型注解语法
    tag: Tag = Tag(tag=tag_in.tag, created=datetime.utcnow(),
        secret="shhhh")
    service.create(tag)
    return tag_in

# 使用 response_model=TagOut 自动过滤返回数据
@app.get('/{tag_str}', response_model=TagOut)
def get_one(tag_str: str) -> TagOut:
    tag: Tag = service.get(tag_str)
    return tag

自动化文档

# 进入该页面 即可在这个页面进行自动化测试 
# 可以实现自定义参数发起请求等等操作
http://localhost:8000/docs

异步编程

异步的主要用途是避免长时间的I/O等待。

综合示例

结合上面的知识，让我们解析一个完整的示例：

import asyncio
# async 用于定义一个异步函数
# 异步函数在被调用时返回一个协程对象，这个对象需要被await或其他方式执行。

示例：
async def q():
    print("Why can't programmers tell jokes?")
    # asyncio.sleep() 是一个异步的睡眠函数，它返回一个协程，在指定的时间后完成。
    await asyncio.sleep(3)
# await用于挂起异步函数的执行，直到await后面的异步操作完成。只能在异步函数中使用。
# await`后面通常跟随一个耗时操作（例如 I/O 操作、网络请求、睡眠等），这些操作返回一个协程对象、任务或未来对象。

async def a():
    print("Timing!")

async def main():
    # asyncio用于编写异步 I/O 代码。它提供了事件循环、协程和任务等功能。
   
	# asyncio.gather() 用于并发运行多个协程，并在所有协程完成后返回它们的结果。
	# 它接受多个协程作为参数，并返回一个包含所有协程结果的列表。
    await asyncio.gather(q(), a())

if __name__ == "__main__":
    # asyncio.run()用于运行一个异步函数，启动事件循环并等待其完成。
    asyncio.run(main())
    
# 首先会运行q()
# 然后进程阻塞 切换到运行a()
# a()执行完切换回q()

# 简单的异步demo
from fastapi import FastAPI
import asyncio
import uvicorn

app = FastAPI()


@app.get("/hi")
async def greet():
    await asyncio.sleep(1)
    return {"message": "Hello? World?"}


if __name__ == "__main__":
    uvicorn.run("Test:app")

Starlette

# 直接使用Starlette编写程序
import uvicorn
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route

async def greeting(request):
    return JSONResponse('Hello? World?')

app = Starlette(debug=True, routes=[
    Route('/hi', greeting),
])

if __name__ == "__main__":
    uvicorn.run("Test:app")

Pydantic

# 基本模型定义
from pydantic import BaseModel
from typing import Optional, List
from datetime import datetime

class User(BaseModel):
    id: int
    username: str
    email: str
    is_active: bool = True
    created_at: datetime
    tags: List[str] = []
    bio: Optional[str] = None
    
# 自动类型转换
user = User(
    id="123",  # 字符串 → 整数
    username="SimonSkywalke",
    email="simon@example.com",
    created_at="2025-03-30T11:27:52"  # 字符串 → datetime
)
print(user.id)  # 输出: 123 (int类型)


from pydantic import ValidationError
try:
    invalid_user = User(
        id="not-a-number",
        username="SimonSkywalke",
        email="invalid-email",
        created_at="2025-03-30"
    )
except ValidationError as e:
    print(e.json())

简单综合应用

from pydantic import BaseModel

# 继承了Pydantic的BaseModel
class Creature(BaseModel):
    # 数据类型提示
    name: str
    country: str
    area: str
    description: str
    aka: str

thing = Creature(
    name="yeti",
    country="CN",
    area="Himalayas",
    description="Hirsute Himalayan",
    aka="Abominable Snowman")

print("Name is", thing.name)

from model import Creature

# 调用了刚才制造的模型
_creatures: list[Creature] = [
    Creature(name="yeti",
             country="CN",
             area="Himalayas",
             description="Hirsute Himalayan",
             aka="Abominable Snowman"
             ),
    Creature(name="sasquatch",
             country="US",
             area="*",
             description="Yeti's Cousin Eddie",
             aka="Bigfoot")
]

def get_creatures() -> list[Creature]:
    return _creatures

from model import Creature
from fastapi import FastAPI

app = FastAPI()

@app.get("/creature")
def get_all() -> list[Creature]:
    from data import get_creatures
    return get_creatures()


if __name__ == "__main__":
    import uvicorn
    uvicorn.run("web:app", reload=True)
# 生成的响应内容
# HTTP/1.1 200 OK
# date: Mon, 31 Mar 2025 02:44:07 GMT
# server: uvicorn
# content-length: 211
# content-type: application/json
#
# [
#   {
#     "name": "yeti",
#     "country": "CN",
#     "area": "Himalayas",
#     "description": "Hirsute Himalayan",
#     "aka": "Abominable Snowman"
#   },
#   {
#     "name": "sasquatch",
#     "country": "US",
#     "area": "*",
#     "description": "Yeti's Cousin Eddie",
#     "aka": "Bigfoot"
#   }
# ]

# FastAPI将自动将对象列表转化成JSON字符串返回
# 服务器端输出 :INFO:     127.0.0.1:61968 - "GET /creature HTTP/1.1" 200 OK

类型验证

# 错误的初始化从而引发报错
from model import Creature

dragon = Creature(
    name="dragon",
    # 会报错 因为根据上面的定义 description是一个str
    description=["incorrect", "string", "list"],
    country="*"
    )

数值验证

对于不同的数据类型可以添加一些限制

整数或浮点数(conint)

gt 大于
lt 小于
ge 大于等于
le 小于等于
multiple_of 一个数值的整数倍

字符串(constr)

min_length 最小长度
max_length 最大长度
to_upper 将内容转成大写
to_lower 将内容转成小写
regex 符合某一正则表达式

元组列表集合

min_items 最小原件数
max_items 最大元件数

from pydantic import BaseModel, constr

class Creature(BaseModel):
    # constr意味着使用了一个受限制的字符串
	name: constr(min_length=2)
    # name: str = Field(..., min_length=2)
    # Field()这个字段是必须的，并且没有默认值。
	country: str
	area: str
	description: str
	aka: str
# 会报错 因为name字段要求min_length=2
bad_creature = Creature(name="!",
description="it's a raccoon",
area="your attic")

类型转换

路径函数可以返回任何内容，就算是JSON不支持的日期对象，路径函数也通过内部函数将其转化成可以通过JSON返回的数据

# 一个检验方法
import datetime
import json
import pytest
from fastapi.encoders import jsonable_encoder

@pytest.fixture
def data():
    return datetime.datetime.now()

# 用于证明日期对象是没法直接转化成JSON对象的
# 断言接下来的代码会抛出Exception异常
def test_json_dump(data):
    with pytest.raises(Exception):
        _ = json.dumps(data)

def test_encoder(data):
    # 都能正常转换 返回非空字符串
    # 在python中被视为ture
    out = jsonable_encoder(data)
    assert out
    json_out = json.dumps(out)
    assert json_out
    # 所以assert都能通过

依赖注入

将函数需要的任何特定信息传递到函数中，然后调用它来获取特定的数据。你web处理函数不会直接调用依赖项;它在函数调用时处理。依赖项需要可被执行(callable)，可以是函数和类

简单例子

from fastapi import FastAPI, Depends, Query

app = FastAPI()

# 依赖方程
# Query(None) 表示参数是可选的.Query(...)代表参数是必须的
def user_dep(name: str = Query(None), password: str = Query(None)):
    return {"name": name, "valid": True}

# 同时也是一个单路经依赖示例
@app.get("/user")
def get_user(user: dict = Depends(user_dep)) -> dict:
    return user

# /user?name=Simon&password=secret 时，服务会返回 {"name": "Simon", "valid": true}

依赖范围

可以定义依赖关系来覆盖单个路径函数、一组路径函数或整个web应用程序。

单路径依赖

# 单路经示例二
# 依赖函数检查查询的name参数有没有缺失 不返回值
from fastapi import FastAPI, Depends, Query, HTTPException

app = FastAPI()

# the dependency function:
def check_dep(name: str = Query(None), password: str = Query(None)):
    if not name:
         # 进行异常处理
        raise HTTPException(status_code=400, detail="Name is required")
        
# the path function / web endpoint:
@app.get("/check_user", dependencies=[Depends(check_dep)])
def check_user() -> bool:
    return True

多路径依赖

# 一个简单的多路径依赖函数
from fastapi import Depends, APIRouter

def depfunc():
    pass
# ... 表示有其他参数被省略（如前缀、标签等）
# dependencies=[Depends(depfunc)] 将 depfunc 设置为此路由器下所有路由的依赖项
router = APIRouter(..., dependencies=[Depends(depfunc)])

全局依赖

from fastapi import FastAPI, Depends
# 定义了两个依赖项
def depfunc1():
    pass
 
def depfunc2():
    pass

# 将这两个依赖定义为全局依赖
# 通过在 FastAPI() 构造函数中传递 dependencies 参数，这些依赖函数将应用于应用中的所有路由。
app = FastAPI(dependencies=[Depends(depfunc1), Depends(depfunc2)])

@app.get("/main")
def get_main():
    pass