fastapi异步处理模板

from fastapi import FastAPI from pydantic import BaseModel # 1. 初始化 FastAPI 应用实例 app = FastAPI( title="字符串处理接口", # 接口文档标题 description="接收单个字符串，返回处理后的字符串", # 接口文档描述 version="1.0.0" # 版本 ) # 2. 定义请求体模型(Pydantic 校验，确保输入为字符串) class StringRequest(BaseModel): input_str: str # 输入参数字段，严格校验为字符串类型 from concurrent.futures import ThreadPoolExecutor # 3. 定义接口：POST 请求，接收字符串→返回字符串 @app.post("/process-string", summary="字符串处理接口", response_description="处理后的字符串结果") async def process_string(request: StringRequest): """ 字符串处理核心接口 - **input_str**: 传入需要处理的原始字符串(必填，字符串类型) """ # 核心业务逻辑：此处可替换为你的自定义处理逻辑(如拼接、过滤、调用其他函数等) # 示例：在原始字符串后拼接「- 已处理」 processed_str = f"{request.input_str} - 已处理" loop = asyncio.get_running_loop() # 方式1：使用默认线程池 task1 = loop.run_in_executor(None, fun, request.input_str) # 返回结果：字典格式，key 为 output_str，value 为处理后的字符串 a=await asyncio.gather(task1) return a[0] @app.post("/test", summary="字符串处理接口", response_description="处理后的字符串结果") async def process_string2(request: StringRequest): """ 字符串处理核心接口 - **input_str**: 传入需要处理的原始字符串(必填，字符串类型) """ # 核心业务逻辑：此处可替换为你的自定义处理逻辑(如拼接、过滤、调用其他函数等) # 示例：在原始字符串后拼接「- 已处理」 return '1' # 4. 启动入口(直接运行该文件即可) if __name__ == "__main__": import uvicorn # 启动服务：host=0.0.0.0 允许外部访问，port=8000 端口可自定义 uvicorn.run(app, host="0.0.0.0", port=9004)