FastAPI的接口编写示范代码

xiaotu

import os
import uvicorn
from fastapi import FastAPI,Query
from pydantic import BaseModel,EmailStr
from typing import Optional,Annotated

app = FastAPI()


# 1. 定义数据模型（就像是给数据定个“模具”）
class Item(BaseModel):
    name: str           # 必填字符串/必填项目就不需要赋初始值
    price: float        # 必填浮点数
    is_offer: Optional[bool] = None  # 可选布尔值，默认为 None/使用Optional更规范
    other: str = 'xxx'
    email: EmailStr # 只需要声明类型为 EmailStr，FastAPI 就会自动校验

# 使用Optional核心作用是允许为空
# 如果不加：别人看你的代码，会以为这个参数必须是字符串，只是你随手给了个 None 做初始值。
# 如果加了：你是在正式宣告——“我是故意的，这个参数在逻辑上允许为空”。这对于团队协作和代码维护非常重要。



#三种参数定义方式：A. 路径参数 (Path Parameters)
@app.get("/users/{user_id}")
def get_user(user_id: int): # 这里的 int 规范了参数必须是整数
    return {"user_id": user_id}

#三种参数定义方式：B.查询参数 (Query Parameters)
#参数跟在 ? 后面。如 /search?page=1
@app.get("/search")
def search(keyword: str, page: int = 1): # keyword 必填，page 可选且默认为 1
    return {"q": keyword, "p": page}

#三种参数定义方式：C.请求体参数 (Body Parameters) - BaseModel方式
## 使用上面定义的类进行规范
@app.post("/update")
def update_user(user: Item):
    return {"status": "updated", "data": user}

#限制参数长度
@app.get("/items/")
async def read_items(
    q: Annotated[Optional[str], Query(max_length=50)] = None):
    # 这里定义了 q 既是可选的，又限制了最大长度
    # 这里 Optional[str] 等同于 Union[str, None]
    return {"q": q}



@app.get("/")
def root():
    return {"status": "success"}


# --- POST 请求：创建/接收数据 ---
@app.post("/items/")
async def create_item(item: Item):
    # FastAPI 已经自动把 JSON 转换成了 item 对象
    # 你可以通过 item.name, item.price 访问数据
    total_price = item.price * 0.9  # 假设打个 9 折

    return {
        "message": "数据接收成功！",
        "item_received": item,
        "discounted_price": total_price,
        "other":item.other
    }

if __name__ == "__main__":
    # 获取当前文件的文件名（去掉路径和 .py 后缀）
    file_name = os.path.basename(__file__).replace(".py", "")
    #因为只有 file_name:app 的方式才能使用 reload=True
    # 动态拼接成 "文件名:app"
    uvicorn.run(f"{file_name}:app", host="0.0.0.0", port=8080, reload=True)

xiaotu

示范二

import traceback
from typing import Optional, Any, Annotated

from fastapi import FastAPI, Depends, Header, HTTPException
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from pydantic import BaseModel

# ====== 模拟 Redis（你换成真实的 r.hget 即可）======
FAKE_DB = {
    "1001": "abc123"
}

# ====== 统一返回函数 ======
def resp(code=0, msg="success", data=None, http_status=200):
    return JSONResponse(
        status_code=http_status,
        content={
            "code": code,
            "msg": msg,
            "data": data
        }
    )

# ====== 自定义业务异常 ======
class BizException(Exception):
    def __init__(self, code: int, msg: str, http_status: int = 400):
        self.code = code
        self.msg = msg
        self.http_status = http_status

# ====== 全局鉴权 ======
async def verify_auth(
    bid: Annotated[Optional[str], Header()] = None,
    token: Annotated[Optional[str], Header()] = None
):
    if not bid or not token:
        raise BizException(1001, "Missing auth", 401)

    token_redis = FAKE_DB.get(bid)

    if not token_redis:
        raise BizException(1002, "Token not found", 401)

    if token != token_redis:
        raise BizException(1003, "Invalid token", 403)

# ====== 创建 app（带全局依赖）======
app = FastAPI(dependencies=[Depends(verify_auth)])


# ====== 全局异常处理 ======

# 1️⃣ 业务异常
@app.exception_handler(BizException)
async def biz_exception_handler(request, exc: BizException):
    return resp(
        code=exc.code,
        msg=exc.msg,
        http_status=exc.http_status
    )

# 2️⃣ 参数校验异常
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc: RequestValidationError):
    return resp(
        code=1100,
        msg="Invalid parameters",
        data=exc.errors(),  # 生产环境可以去掉
        http_status=400
    )

# 3️⃣ HTTPException
@app.exception_handler(HTTPException)
async def http_exception_handler(request, exc: HTTPException):
    return resp(
        code=1000,
        msg=str(exc.detail),
        http_status=exc.status_code
    )

# 4️⃣ 全局兜底异常（500）
@app.exception_handler(Exception)
async def global_exception_handler(request, exc: Exception):
    print("🔥 服务器异常：")
    print(traceback.format_exc())

    return resp(
        code=1000,
        msg="Internal server error",
        http_status=500
    )

# ====== 数据模型 ======
class Item(BaseModel):
    name: str
    price: float
    is_offer: Optional[bool] = None


# ====== 接口示例 ======

@app.get("/ok")
async def ok():
    return resp(data={"msg": "正常返回"})


@app.get("/biz_error")
async def biz_error():
    raise BizException(2001, "业务逻辑错误", 400)


@app.get("/http_error")
async def http_error():
    raise HTTPException(status_code=404, detail="Not found")


@app.get("/server_error")
async def server_error():
    x = 1 / 0  # 故意报错
    return resp()


@app.post("/item")
async def create_item(item: Item):
    return resp(data=item.dict())