第5周学习计划：FastAPI HTTP请求与API基础实战

发表于   |   分类于 技术 , Python   |   暂无评论

Python+AI第5周学习指南：HTTP请求与API基础

一、知识点讲解

（一）requests库高级用法

作用说明：requests库是Python常用的HTTP请求工具，高级用法可实现超时控制、请求重试、会话保持等，解决接口请求中的稳定性问题，适用于各类API调用场景。

1. 核心参数详解

参数说明常用可选值及说明传参示例
timeout设置请求超时时间，避免无限等待整数（秒）、元组（(连接超时, 读取超时)）timeout=5 或 timeout=(3, 7)
headers自定义请求头，模拟浏览器或携带认证信息User-Agent（浏览器标识）、Authorization（认证）等headers={"User-Agent": "Mozilla/5.0", "token": "xxx"}
paramsGET请求参数，自动拼接到URL字典格式，键值对为参数名和参数值params={"key": "高德API密钥", "city": "济南"}
verify是否验证SSL证书，用于HTTPS请求True（默认，验证）、False（不验证）verify=False

2. 典型用法示例

（1）基础GET请求（带超时）


import requests

def get_amap_data():
    url = "https://restapi.amap.com/v3/geocode/geo"
    params = {
        "key": "你的高德API密钥",
        "address": "济南市"
    }
    try:
        # 超时设置为5秒，超过则抛出异常
        response = requests.get(url, params=params, timeout=5)
        response.raise_for_status()  # 抛出HTTP状态码异常（4xx/5xx）
        return response.json()
    except requests.exceptions.RequestException as e:
        print(f"请求失败：{e}")
        return None

（2）请求重试机制（借助urllib3）


import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def create_retry_session(retry_times=3, backoff_factor=0.5):
    """创建带重试机制的会话"""
    session = requests.Session()
    # 定义重试策略：重试3次，间隔0.5秒递增，仅重试指定状态码
    retry_strategy = Retry(
        total=retry_times,
        backoff_factor=backoff_factor,  # 间隔时间=backoff_factor*(2^(重试次数-1))
        status_forcelist=[429, 500, 502, 503, 504],  # 需要重试的HTTP状态码
        allowed_methods=["GET"]  # 仅对GET请求重试
    )
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("https://", adapter)  # 对HTTPS请求应用重试策略
    session.mount("http://", adapter)   # 对HTTP请求应用重试策略
    return session

# 使用带重试的会话请求
session = create_retry_session()
try:
    response = session.get("https://restapi.amap.com/v3/geocode/geo", 
                          params={"key": "你的密钥", "address": "济南"},
                          timeout=5)
    print(response.json())
except requests.exceptions.RequestException as e:
    print(f"多次重试后仍失败：{e}")

3. 常见陷阱提醒

  • 超时时间设置过短易导致正常请求失败，过长可能阻塞程序，建议根据接口响应速度设置（3-10秒为宜）。
  • 重试机制不可滥用，避免因频繁重试给API服务器造成压力，需结合状态码精准控制重试场景。
  • 未处理response.raise_for_status()时，4xx/5xx状态码不会主动抛出异常，易忽略请求错误。
  • API密钥直接硬编码存在安全风险，建议通过环境变量或配置文件读取。

（二）RESTful接口规范

作用说明：RESTful是一种API设计规范，基于HTTP协议，使接口具有可读性强、可扩展性高、兼容性好的特点，便于前后端协作和接口维护。

1. 核心规范详解

规范项说明常用可选值及说明示例
HTTP方法语义用HTTP方法表示操作类型，而非URL中带动词GET（查询）、POST（新增）、PUT（全量更新）、DELETE（删除）GET /api/cold-chain/route（查询冷链路线）
URL命名使用名词复数形式， lowercase+连字符（-），不使用下划线名词复数（表资源集合）、避免动词/api/cold-chain/vehicles（冷链车辆资源）
状态码返回用HTTP状态码表示请求结果，配合自定义消息200（成功）、201（创建成功）、400（参数错误）、404（资源不存在）、500（服务器错误）查询成功返回200，参数错误返回400
响应格式统一返回JSON格式，包含状态、消息、数据字段code（自定义状态码）、msg（提示信息）、data（业务数据）{"code":200,"msg":"success","data":{"route":[]}}

2. 典型用法示例（RESTful接口设计）



# 符合RESTful规范的冷链路线接口设计（后续FastAPI实现）
# 1. 查询路线（GET）
GET /api/cold-chain/routes?start=济南&end=青岛&cargo_type=冷冻食品
# 响应：200 OK
{
  "code": 200,
  "msg": "查询成功",
  "data": {
    "route_id": "RT20260116001",
    "start_point": "济南市历下区",
    "end_point": "青岛市黄岛区",
    "distance": 360.5,  # 单位：公里
    "duration": 240,    # 单位：分钟
    "path": ["济南绕城高速", "青银高速"]
  }
}

# 2. 新增路线规划（POST）
POST /api/cold-chain/routes
请求体：
{
  "start": "济南",
  "end": "烟台",
  "cargo_type": "冷藏药品",
  "temperature": 2  # 单位：℃
}
# 响应：201 Created
{
  "code": 201,
  "msg": "路线规划成功",
  "data": {"route_id": "RT20260116002"}
}

# 3. 错误响应（参数缺失）
GET /api/cold-chain/routes?start=济南
# 响应：400 Bad Request
{
  "code": 40001,
  "msg": "参数错误：缺少必填参数end（目的地）",
  "data": null
}

3. 常见陷阱提醒

  • 避免在URL中包含动词，如/getRoute、/addVehicle，违反RESTful语义，应改用HTTP方法区分操作。
  • 状态码使用混乱，如查询不到资源返回200+自定义错误，而非标准404，增加前端处理成本。
  • 响应格式不统一，部分接口返回data字段，部分直接返回数据，导致前端解析逻辑不一致。
  • 忽略资源层级关系，如嵌套资源应设计为 /api/cold-chain/vehicles/{vehicle_id}/routes（车辆下的路线）。

（三）FastAPI快速上手

作用说明：FastAPI是一款高性能的现代Python Web框架，专为构建API设计，支持自动生成Swagger文档、类型提示、异步请求，开发效率远超Flask、Django。

1. 核心参数/装饰器详解

装饰器/参数说明常用可选值及说明示例
@app.get(path)GET请求接口装饰器，定义接口路径path：接口URL路径，支持路径参数@app.get("/api/cold-chain/routes")
Query参数URL查询参数，通过函数参数定义，支持类型提示default（默认值）、None（必填）、ge/le（数值范围）start: str, end: str, temperature: float = 0.0
Path参数URL路径中的参数，用于标识具体资源嵌入URL路径，通过函数参数接收，支持类型提示@app.get("/api/routes/{route_id}")，route_id: str
Response模型定义响应数据结构，自动校验和格式化Pydantic模型，指定response_model参数response_model=RouteResponse

2. 典型用法示例

（1）基础FastAPI应用（带Swagger文档）


from fastapi import FastAPI, Query
from pydantic import BaseModel
from typing import Optional

# 1. 初始化FastAPI应用
app = FastAPI(
    title="济南冷链运输路线API",
    description="提供冷链运输最优路线查询、规划功能",
    version="1.0.0"
)

# 2. 定义响应模型（Pydantic），自动校验响应格式
class RouteData(BaseModel):
    route_id: str
    start_point: str
    end_point: str
    distance: float  # 公里
    duration: int    # 分钟

class ApiResponse(BaseModel):
    code: int
    msg: str
    data: Optional[RouteData] = None

# 3. 定义GET接口（查询路线）
@app.get("/api/cold-chain/routes", 
         response_model=ApiResponse,
         summary="查询冷链运输最优路线",
         description="根据起点、终点查询济南及周边冷链运输最优路线，支持指定货物类型")
def get_cold_chain_route(
    start: str = Query(..., description="起点城市/区域，如：济南市历下区"),
    end: str = Query(..., description="终点城市/区域，如：青岛市黄岛区"),
    cargo_type: str = Query("冷冻食品", description="货物类型，可选：冷冻食品、冷藏药品")
):
    # 模拟数据库/高德API返回数据
    route_data = RouteData(
        route_id="RT20260116001",
        start_point=start,
        end_point=end,
        distance=360.5,
        duration=240
    )
    return ApiResponse(code=200, msg="查询成功", data=route_data)

# 启动命令：uvicorn main:app --reload
# Swagger文档地址：http://127.0.0.1:8000/docs
# 备用文档地址：http://127.0.0.1:8000/redoc

3. 常见陷阱提醒

  • 忘记安装依赖包，FastAPI需手动安装：pip install fastapi uvicorn（uvicorn为ASGI服务器）。
  • 函数参数未指定类型提示，导致FastAPI无法自动生成文档和校验参数，失去核心优势。
  • 路径参数与查询参数混淆，如将资源ID放入查询参数而非路径参数，违反RESTful规范。
  • 生产环境启用--reload参数，该参数为开发热重载功能，生产环境需关闭，避免性能问题。

（四）接口异常处理

作用说明：统一处理接口调用中的各类异常（参数错误、业务异常、服务器错误等），返回标准化响应，提升接口健壮性和用户体验。

1. 核心方法详解

方法/装饰器说明常用可选值及说明示例
@app.exception_handler(异常类型)FastAPI自定义异常处理器，捕获指定类型异常HTTPException、ValueError、自定义异常类@app.exception_handler(HTTPException)
HTTPExceptionFastAPI内置HTTP异常类，携带状态码和详情status_code（HTTP状态码）、detail（异常描述）raise HTTPException(status_code=400, detail="参数错误")
自定义异常类处理业务特定异常，如API密钥无效、权限不足继承Exception，可自定义属性（如错误码）class AmapKeyError(Exception): pass

2. 典型用法示例


from fastapi import FastAPI, HTTPException, Query
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from typing import Optional

app = FastAPI(title="冷链路线API", version="1.0.0")

# 1. 自定义业务异常
class AmapKeyError(Exception):
    """高德API密钥异常"""
    pass

class RouteNotFoundError(Exception):
    """路线未找到异常"""
    def __init__(self, start: str, end: str):
        self.start = start
        self.end = end

# 2. 定义响应模型
class ApiResponse(BaseModel):
    code: int
    msg: str
    data: Optional[dict] = None

# 3. 自定义异常处理器
@app.exception_handler(AmapKeyError)
async def amap_key_exception_handler(request, exc):
    return JSONResponse(
        status_code=401,  # 未授权
        content=ApiResponse(code=40101, msg="高德API密钥无效或过期", data=None).dict()
    )

@app.exception_handler(RouteNotFoundError)
async def route_not_found_handler(request, exc):
    return JSONResponse(
        status_code=404,  # 资源不存在
        content=ApiResponse(
            code=40401, 
            msg=f"未查询到从{exc.start}到{exc.end}的冷链路线", 
            data=None
        ).dict()
    )

# 4. 接口中抛出异常
@app.get("/api/cold-chain/routes", response_model=ApiResponse)
def get_route(
    start: str = Query(...),
    end: str = Query(...),
    amap_key: str = Query(...)
):
    # 模拟密钥校验
    if amap_key != "valid_key_123":
        raise AmapKeyError()
    
    # 模拟路线查询
    if start == "济南" and end == "拉萨":
        raise RouteNotFoundError(start=start, end=end)
    
    # 正常返回
    return ApiResponse(
        code=200,
        msg="查询成功",
        data={"route_id": "RT20260116001", "distance": 360.5}
    )

3. 常见陷阱提醒

  • 异常处理器未注册，自定义异常抛出后无法被捕获，返回默认500错误。
  • 异常响应格式与正常响应不一致，导致前端需单独处理异常场景，增加复杂度。
  • 捕获异常范围过广（如直接捕获Exception），掩盖了真正的错误原因，不利于调试。
  • 未记录异常日志，异常发生后无法追溯问题，建议在异常处理器中添加日志打印。

二、实战场景

实战需求：封装高德地图API+开发济南冷链路线查询接口

实现目标：封装带超时、重试机制的高德地图路径规划API请求函数，基于FastAPI开发冷链路线查询接口，自动生成Swagger文档，返回标准化异常响应。

完整示例代码


import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
from fastapi import FastAPI, Query, HTTPException
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from typing import Optional
import logging

# ---------------------- 基础配置 ----------------------
# 日志配置（记录异常信息）
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)

# 高德API配置
AMAP_ROUTE_URL = "https://restapi.amap.com/v3/direction/driving"
AMAP_KEY = "你的高德API密钥"  # 建议替换为环境变量：os.getenv("AMAP_KEY")

# 初始化FastAPI应用
app = FastAPI(
    title="济南冷链运输最优路线查询API",
    description="基于高德地图API，提供济南及周边地区冷链运输路线查询，支持异常处理和Swagger测试",
    version="1.0.0"
)

# ---------------------- 异常定义与处理 ----------------------
# 自定义异常
class AmapRequestError(Exception):
    """高德API请求异常"""
    pass

class AmapResponseError(Exception):
    """高德API响应异常（返回错误信息）"""
    def __init__(self, err_code: str, err_msg: str):
        self.err_code = err_code
        self.err_msg = err_msg

# 响应模型
class RouteDetail(BaseModel):
    """路线详情模型"""
    route_id: str
    start_point: str
    end_point: str
    distance: float  # 总距离（公里）
    duration: int    # 总耗时（分钟）
    path: list[str]  # 路线节点列表
    toll: float      # 预估过路费（元）

class ApiResponse(BaseModel):
    """统一API响应模型"""
    code: int
    msg: str
    data: Optional[RouteDetail] = None

# 异常处理器
@app.exception_handler(AmapRequestError)
async def amap_request_handler(request, exc):
    logger.error(f"高德API请求失败：{str(exc)}")
    return JSONResponse(
        status_code=503,
        content=ApiResponse(code=50301, msg="高德API服务暂时不可用", data=None).dict()
    )

@app.exception_handler(AmapResponseError)
async def amap_response_handler(request, exc):
    logger.error(f"高德API响应错误：{exc.err_code} - {exc.err_msg}")
    return JSONResponse(
        status_code=400,
        content=ApiResponse(code=int(exc.err_code), msg=exc.err_msg, data=None).dict()
    )

# ---------------------- 高德API工具函数 ----------------------
def create_retry_session(retry_times=3, backoff_factor=0.5) -> requests.Session:
    """创建带重试机制的请求会话"""
    session = requests.Session()
    retry_strategy = Retry(
        total=retry_times,
        backoff_factor=backoff_factor,
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods=["GET"]
    )
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("https://", adapter)
    session.mount("http://", adapter)
    return session

def get_amap_route(start: str, end: str) -> dict:
    """
    调用高德地图路径规划API，获取驾驶路线（适配冷链运输，优先高速）
    :param start: 起点（经纬度或地址，如："116.481028,39.921983"或"济南市"）
    :param end: 终点（格式同起点）
    :return: 高德API返回的原始数据
    """
    session = create_retry_session()
    params = {
        "key": AMAP_KEY,
        "origin": start,
        "destination": end,
        "strategy": 2,  # 路线策略：2-高速优先（适配冷链快速运输）
        "show_fields": "cost,toll"  # 显示费用、过路费信息
    }
    try:
        # 超时设置：连接3秒，读取7秒
        response = session.get(AMAP_ROUTE_URL, params=params, timeout=(3, 7))
        response.raise_for_status()  # 抛出HTTP状态码异常
        data = response.json()
        # 校验高德API返回状态
        if data["status"] != "1":
            raise AmapResponseError(
                err_code=data["infocode"],
                err_msg=data["info"]
            )
        return data
    except requests.exceptions.RequestException as e:
        raise AmapRequestError(f"请求异常：{str(e)}") from e

# ---------------------- FastAPI接口 ----------------------
@app.get(
    path="/api/cold-chain/routes",
    response_model=ApiResponse,
    summary="查询冷链运输最优路线",
    description="基于高德地图API，查询济南及周边冷链运输路线，优先高速，返回距离、耗时、过路费等信息"
)
def query_cold_chain_route(
    start: str = Query(..., description="起点地址，如：济南市历下区、117.000923,36.675807"),
    end: str = Query(..., description="终点地址，格式同起点"),
    cargo_type: str = Query("冷冻食品", description="货物类型，用于后续扩展冷链专属路线")
):
    # 1. 调用高德API获取路线数据
    amap_data = get_amap_route(start=start, end=end)
    
    # 2. 解析高德API数据，构造业务响应
    route = amap_data["route"]["paths"][0]  # 取第一条最优路线
    route_detail = RouteDetail(
        route_id=f"RT{amap_data['timestamp']}",  # 用时间戳生成唯一路线ID
        start_point=start,
        end_point=end,
        distance=round(float(route["distance"])/1000, 1),  # 米转公里，保留1位小数
        duration=round(int(route["duration"])/60),  # 秒转分钟，取整
        path=[step["instruction"] for step in route["steps"]],  # 路线步骤
        toll=round(float(route["toll"]), 2)  # 过路费，保留2位小数
    )
    
    # 3. 返回标准化响应
    return ApiResponse(code=200, msg="查询成功", data=route_detail)

# ---------------------- 启动说明 ----------------------
# 1. 安装依赖：pip install fastapi uvicorn requests urllib3
# 2. 启动命令：uvicorn main:app --reload
# 3. 测试地址：
#    - Swagger文档：http://127.0.0.1:8000/docs
#    - 接口测试：http://127.0.0.1:8000/api/cold-chain/routes?start=济南&end=青岛&cargo_type=冷冻食品

实战技巧与参数总结

  1. 重试与超时策略：通过urllib3的Retry类实现3次重试，间隔0.5秒递增，仅对5xx和429状态码重试，避免无效重试；超时设置为(3,7)，分别控制连接超时和读取超时，平衡稳定性和效率。
  2. 高德API路线策略：参数strategy=2表示高速优先，适配冷链运输“快速送达”需求；show_fields参数指定返回费用信息，满足业务对成本预估的需求。
  3. 标准化响应设计：通过Pydantic模型强制响应格式统一，即使异常场景也返回相同结构的JSON，降低前端解析成本；路线ID用时间戳生成，确保唯一性。
  4. 日志与异常追溯：集成logging模块记录异常详情，便于问题排查；自定义异常携带具体错误信息，既友好返回给前端，又能在日志中保留调试依据。
  5. Swagger文档优化：通过FastAPI的summary、description参数完善接口说明，前端开发者可直接在/docs页面查看参数含义和响应格式，无需额外文档。

三、练习题

练习题1：封装高德地图地理编码API（带重试与超时）

题目描述

地理编码API可将地址（如“济南市历下区”）转换为经纬度，便于后续路线规划。请封装一个函数，调用高德地图地理编码API，实现地址转经纬度功能，要求包含3次重试、5秒超时机制，返回标准化结果，处理API密钥错误、地址不存在等异常。

实现思路

  1. 查阅高德地理编码API文档，确定请求URL（https://restapi.amap.com/v3/geocode/geo）和参数（key、address）。
  2. 复用之前的create_retry_session函数，创建带重试机制的会话。
  3. 定义自定义异常（如AddressNotFoundError），处理地址不存在场景。
  4. 解析API响应，提取经纬度，返回格式为{"longitude": 117.0, "latitude": 36.7}。
  5. 添加异常捕获，返回标准化错误信息。

参考代码


import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# 自定义异常
class AddressNotFoundError(Exception):
    """地址未找到异常"""
    pass

class AmapKeyError(Exception):
    """高德API密钥错误"""
    pass

def create_retry_session(retry_times=3, backoff_factor=0.5) -> requests.Session:
    """创建带重试机制的会话"""
    session = requests.Session()
    retry_strategy = Retry(
        total=retry_times,
        backoff_factor=backoff_factor,
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods=["GET"]
    )
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("https://", adapter)
    return session

def address_to_coords(address: str, amap_key: str) -> dict:
    """
    地址转经纬度（高德地理编码API）
    :param address: 具体地址，如：济南市历下区
    :param amap_key: 高德API密钥
    :return: 经纬度字典，{"longitude": 经度, "latitude": 纬度}
    """
    url = "https://restapi.amap.com/v3/geocode/geo"
    params = {"key": amap_key, "address": address}
    session = create_retry_session()
    
    try:
        response = session.get(url, params=params, timeout=5)
        response.raise_for_status()
        data = response.json()
        
        # 校验API响应状态
        if data["status"] != "1":
            if data["infocode"] == "10001":
                raise AmapKeyError("API密钥无效或过期")
            raise AddressNotFoundError(f"地址不存在：{address}")
        
        # 解析经纬度
        geocode = data["geocodes"][0]
        longitude, latitude = geocode["location"].split(",")
        return {
            "longitude": round(float(longitude), 6),
            "latitude": round(float(latitude), 6)
        }
    except requests.exceptions.RequestException as e:
        raise Exception(f"请求失败：{str(e)}") from e

# 测试
if __name__ == "__main__":
    try:
        coords = address_to_coords("济南市历下区", "你的高德API密钥")
        print("经纬度：", coords)
    except (AmapKeyError, AddressNotFoundError) as e:
        print("错误：", e)
    except Exception as e:
        print("未知错误：", e)

练习题2：开发冷链路线新增接口（POST请求）

题目描述

基于FastAPI，开发一个POST接口，用于新增冷链路线规划任务。接口需接收起点、终点、货物类型、冷藏温度等参数，调用练习题1的地址转经纬度函数，再调用高德路线API，将规划结果存储到模拟数据库（列表），返回路线ID和规划详情，支持参数校验和异常处理。

实现思路

  1. 定义请求模型（Pydantic），包含start、end、cargo_type、temperature等字段，设置参数校验（如temperature范围：-20~10℃）。
  2. 创建POST接口（/api/cold-chain/routes），接收请求参数。
  3. 调用address_to_coords函数，将起点、终点转为经纬度。
  4. 调用高德路线API，获取路线详情，构造业务数据。
  5. 用列表模拟数据库，存储路线数据，返回路线ID和详情。
  6. 统一异常处理，返回标准化响应。

参考代码


from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from typing import Optional, List
import time
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# 初始化FastAPI
app = FastAPI(title="冷链路线规划API", version="1.0.0")

# 模拟数据库（列表存储路线数据）
route_db: List[dict] = []

# 高德API配置
AMAP_KEY = "你的高德API密钥"
AMAP_ROUTE_URL = "https://restapi.amap.com/v3/direction/driving"

# ---------------------- 工具函数与异常 ----------------------
class AddressNotFoundError(Exception):
    pass

class AmapKeyError(Exception):
    pass

def create_retry_session(retry_times=3, backoff_factor=0.5) -> requests.Session:
    session = requests.Session()
    retry_strategy = Retry(total=retry_times, backoff_factor=backoff_factor, status_forcelist=[429, 500, 503])
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("https://", adapter)
    return session

def address_to_coords(address: str) -> dict:
    """地址转经纬度"""
    url = "https://restapi.amap.com/v3/geocode/geo"
    params = {"key": AMAP_KEY, "address": address}
    session = create_retry_session()
    try:
        response = session.get(url, params=params, timeout=5)
        response.raise_for_status()
        data = response.json()
        if data["status"] != "1":
            raise AddressNotFoundError(f"地址不存在：{address}")
        lon, lat = data["geocodes"][0]["location"].split(",")
        return {"lon": float(lon), "lat": float(lat)}
    except requests.exceptions.RequestException as e:
        raise Exception(f"地理编码请求失败：{str(e)}") from e

# ---------------------- 模型定义 ----------------------
class RouteCreateRequest(BaseModel):
    """路线创建请求模型"""
    start: str = Field(..., description="起点地址，如：济南市历下区")
    end: str = Field(..., description="终点地址，如：青岛市黄岛区")
    cargo_type: str = Field(..., description="货物类型，如：冷冻食品、冷藏药品")
    temperature: float = Field(..., ge=-20, le=10, description="冷藏温度（℃），范围：-20~10")

class RouteCreateResponse(BaseModel):
    """路线创建响应模型"""
    code: int
    msg: str
    data: dict

# ---------------------- POST接口 ----------------------
@app.post("/api/cold-chain/routes", response_model=RouteCreateResponse, summary="新增冷链路线规划")
def create_cold_chain_route(req: RouteCreateRequest):
    try:
        # 1. 地址转经纬度
        start_coords = address_to_coords(req.start)
        end_coords = address_to_coords(req.end)
        
        # 2. 调用高德路线API
        session = create_retry_session()
        params = {
            "key": AMAP_KEY,
            "origin": f"{start_coords['lon']},{start_coords['lat']}",
            "destination": f"{end_coords['lon']},{end_coords['lat']}",
            "strategy": 2  # 高速优先
        }
        response = session.get(AMAP_ROUTE_URL, params=params, timeout=5)
        response.raise_for_status()
        amap_data = response.json()
        if amap_data["status"] != "1":
            raise Exception(f"路线规划失败：{amap_data['info']}")
        
        # 3. 构造路线数据
        route = amap_data["route"]["paths"][0]
        route_id = f"RT{int(time.time())}"  # 时间戳作为唯一ID
        route_info = {
            "route_id": route_id,
            "start": req.start,
            "end": req.end,
            "cargo_type": req.cargo_type,
            "temperature": req.temperature,
            "distance": round(float(route["distance"])/1000, 1),
            "duration": round(int(route["duration"])/60),
            "toll": round(float(route.get("toll", 0)), 2)
        }
        
        # 4. 存储到模拟数据库
        route_db.append(route_info)
        
        # 5. 返回响应
        return RouteCreateResponse(
            code=201,
            msg="路线规划成功",
            data=route_info
        )
    except AddressNotFoundError as e:
        return RouteCreateResponse(code=404, msg=str(e), data={})
    except AmapKeyError as e:
        return RouteCreateResponse(code=401, msg=str(e), data={})
    except Exception as e:
        return RouteCreateResponse(code=500, msg=f"服务器错误：{str(e)}", data={})

# 测试：启动后通过Swagger文档（/docs）发送POST请求
# 请求体示例：
# {
#   "start": "济南市历下区",
#   "end": "青岛市黄岛区",
#   "cargo_type": "冷冻食品",
#   "temperature": -5
# }

练习题3：开发路线查询接口（按ID查询）

题目描述

基于练习题2的模拟数据库，开发一个GET接口（/api/cold-chain/routes/{route_id}），通过路线ID查询路线详情。要求支持路径参数校验，处理路线ID不存在的异常，返回标准化响应，补充到FastAPI应用中，自动生成Swagger文档。

实现思路

  1. 定义GET接口，使用路径参数route_id接收查询条件。
  2. 遍历模拟数据库（route_db），根据route_id查找对应的路线数据。
  3. 若找到路线，返回详情；若未找到，抛出异常并返回404响应。
  4. 统一响应格式，与之前接口保持一致。

参考代码


# 续练习题2的代码，添加以下内容
from fastapi import Path

class RouteQueryResponse(BaseModel):
    """路线查询响应模型"""
    code: int
    msg: str
    data: Optional[dict] = None

@app.get(
    "/api/cold-chain/routes/{route_id}",
    response_model=RouteQueryResponse,
    summary="按ID查询冷链路线详情"
)
def query_route_by_id(
    route_id: str = Path(..., description="路线唯一ID，如：RT1736999999")
):
    # 遍历模拟数据库查找路线
    for route in route_db:
        if route["route_id"] == route_id:
            return RouteQueryResponse(
                code=200,
                msg="查询成功",
                data=route
            )
    # 未找到路线
    return RouteQueryResponse(
        code=404,
        msg=f"路线ID不存在：{route_id}",
        data=None
    )

# 测试：
# 1. 先通过POST接口创建路线，获取route_id
# 2. 访问：http://127.0.0.1:8000/api/cold-chain/routes/你的route_id
# 3. 未找到时返回404提示，找到时返回路线详情

（注：文档部分内容可能由 AI 生成）

添加新评论