Python-fastapi框架

 2025/05/28 

fastapi支持快速搭建api，并且原生支持协程，异步特性

在搭建api这方面比flask更加专精

快速开始

pip install fastapi uvicorn pydantic

vim main.py
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
async def root():
    return {"message": "Hello World"}
    
# 启动命令
uvicorn main:app --reload

Query类

专门用于定义查询参数（URL 中的 ?key=value 参数）

自动验证类型、数值控制范围（不符会直接返回错误）
自动转换类型，如传入数字，自动转换为字符串（未开启严格模式时）
自动生成文档

@app.get("/items/")
async def list_items(
    page: int = Query(1, description="页码"),  # 查询参数，未传值时默认为1
    size: int = Query(10, ge=1)               # ≥1的整数，未传值时默认为10
):
    return {"page": page, "size": size}
    
# 查询方式
GET /items/?page=2&size=20

特性	Query 参数	Pydantic 模型
数据来源	URL 的 `?` 后面	请求体（JSON/表单等）
数据结构	扁平键值对	支持嵌套的复杂结构
验证能力	基本类型和简单约束	支持复杂验证（自定义逻辑等）
典型应用场景	分页、过滤、排序	用户注册、订单提交、配置更新
文档显示位置	Swagger 的 “Parameters” 部分	Swagger 的 “Request body” 部分

Pydantic模型

负责处理结构化数据，比如从前端传送过来一个json数据

同上，自动验证数据类型
自动生成文档

需要处理的json：
{  
"name": "张三", 
"age": 25,  
"email": "zhangsan@example.com",  
"password": "123456"
}

from fastapi import FastAPI
from pydantic import BaseModel, EmailStr, Field
# EmailStr用以验证email格式
# Field用来限制数据范围

class UserCreate(BaseModel):
		name: str
		age: int = Field(ge=18, le=100, description="年龄需在18-100岁之间") # 数据范围
		email: EmailStr
		password: str = Field(..., min_length=6, pattern="^[A-Za-z0-9]+$")  # 密码规则

app = FastAPI()
@app.post("/register")
async def register_user(user: UserCreate):  # 直接使用模型作为参数
    # 此时 user 已经是验证后的对象
    return {
        "msg": "注册成功",
        "data": user.model_dump()  # 将模型转为字典
    }

uvicorn main:app --reload

项目实践

@staticmethod
@app.get("/api/read_base_sw_switch")
async def read_base_type():
    try:
        result = subprocess.run(['xxxx'],timeout=10,
                                capture_output=True,text=True,check=True)
        if result.stdout == "Test":
            return {"code": 0,"data": "test","msg": "success"}
        elif result.stdout == "Fleet":
            return {"code": 0,"data": "fleet","msg": "success"}
        else:
            return {"code": 1,"data": "","msg": "error not get the execpted value"}

    except subprocess.TimeoutExpired:
        raise HTTPException(504, "Command timed out")
    except subprocess.CalledProcessError as e:
        raise HTTPException(400, f"Command failed: {e.stderr.decode()}")
    except Exception as e:
        raise HTTPException(500, f"Internal error: {str(e)}")

@staticmethod
@app.post("/api/set_base_sw_switch")
async def switch_base_type(base_type: dict):
    try:
        type_value = base_type['type']
        if type_value == "test":
            result = subprocess.run([xxxx],
                                    timeout=10,capture_output=True,check=True)
        elif type_value == "fleet":
            result = subprocess.run([xxxx],
                                    timeout=10,capture_output=True,check=True)
        else:
            return {"code": 1, "data": "","msg": "Invalid 'type' value: must be 'test' or 'fleet'"}
        return JSONResponse(
        status_code=200,
        content={
            "code": 0,
            "data": "",
            "msg": "successful set"})

    except subprocess.TimeoutExpired:
        raise HTTPException(504, "Command timed out")
    except subprocess.CalledProcessError as e:
        error_detail = e.stderr.decode().strip() or "No error details"
        raise HTTPException(500, f"Command failed: {error_detail}")
    except Exception as e:
        raise HTTPException(500, f"Internal error: {str(e)}")