from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel from typing import Optional import numpy as np import time import os import sys app = FastAPI( title="FastAPI Service", description="高性能 FastAPI 服务", version="1.0.0", docs_url="/docs", redoc_url="/redoc", openapi_url="/openapi.json" ) # CORS 配置 app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # 数据模型 class Item(BaseModel): name: str description: Optional[str] = None price: float tax: Optional[float] = None class ComputeRequest(BaseModel): numbers: list[float] operation: str class ComputeResponse(BaseModel): result: float operation: str timestamp: float # 健康检查 @app.get("/health") async def health_check(): return { "status": "healthy", "service": "fastapi", "version": "1.0.0", "timestamp": time.time(), "environment": os.getenv("ENVIRONMENT", "development") } # 根路径 @app.get("/") async def root(): return { "message": "欢迎使用 FastAPI 服务", "docs": "/docs", "redoc": "/redoc", "endpoints": [ "/health", "/items", "/compute", "/info" ] } # 获取服务信息 @app.get("/info") async def get_info(): return { "service": "fastapi", "port": os.getenv("PORT", 8000), "python": { "version": os.sys.version, 'copyright': sys.copyright, 'version_info': sys.version_info, 'path': sys.path, 'api_version': sys.api_version, 'argv': sys.argv, 'defaultencoding': sys.getdefaultencoding(), 'filesystemencoding': sys.getfilesystemencoding() }, "system": { 'platform': sys.platform, 'cwd': os.getcwd(), 'name': os.name, 'curdir': os.curdir, 'path.curdir': os.path.curdir }, "environment": os.getenv("ENVIRONMENT", "development"), "startup_time": app.extra.get("startup_time", time.time()) } # 项目 CRUD 操作 items_db = [] @app.get("/items") async def read_items(skip: int = 0, limit: int = 10): return items_db[skip: skip + limit] @app.get("/items/{item_id}") async def read_item(item_id: int): if item_id < 0 or item_id >= len(items_db): raise HTTPException(status_code=404, detail="项目未找到") return items_db[item_id] @app.post("/items") async def create_item(item: Item): items_db.append(item.dict()) return {"message": "项目创建成功", "item": item} @app.put("/items/{item_id}") async def update_item(item_id: int, item: Item): if item_id < 0 or item_id >= len(items_db): raise HTTPException(status_code=404, detail="项目未找到") items_db[item_id] = item.dict() return {"message": "项目更新成功", "item": item} @app.delete("/items/{item_id}") async def delete_item(item_id: int): if item_id < 0 or item_id >= len(items_db): raise HTTPException(status_code=404, detail="项目未找到") deleted_item = items_db.pop(item_id) return {"message": "项目删除成功", "item": deleted_item} # 计算功能 @app.post("/compute", response_model=ComputeResponse) async def compute_numbers(request: ComputeRequest): if not request.numbers: raise HTTPException(status_code=400, detail="数字列表不能为空") result = 0 operation = request.operation.lower() if operation == "sum": result = sum(request.numbers) elif operation == "product": result = 1 for num in request.numbers: result *= num elif operation == "average": result = sum(request.numbers) / len(request.numbers) elif operation == "max": result = max(request.numbers) elif operation == "min": result = min(request.numbers) else: raise HTTPException(status_code=400, detail="不支持的操作类型") return ComputeResponse( result=result, operation=operation, timestamp=time.time() ) # 性能测试 @app.get("/benchmark/{iterations}") async def benchmark(iterations: int = 1000000): start_time = time.time() # 执行一些计算密集型操作 total = 0 for i in range(iterations): total += i * i end_time = time.time() return { "iterations": iterations, "total": total, "time_seconds": end_time - start_time, "operations_per_second": iterations / (end_time - start_time) if (end_time - start_time) > 0 else 0 } # 启动事件 @app.on_event("startup") async def startup_event(): app.extra["startup_time"] = time.time() print(f"FastAPI 服务启动在端口 {os.getenv('PORT', 8000)}") # 路由测试 @app.get("/hello/") def hello_json(): return {"msg": "/hello/ : Hello, FastAPI Node World!"} # 带参数的路由 @app.get("/hi//") @app.get("/hi/") # 兼容无参数情况 def hi_page(name=None): if name is None: name = "访客" return f"

你好，{name}！

返回首页" # 运行外部程序路由 @app.get("/exec/{name}/") @app.get("/exec/") # 兼容无参数情况 async def run_executable(name: str = "esample"): import subprocess # 获取当前Python文件的目录 current_dir = os.path.dirname(os.path.abspath(__file__)) # 构建程序路径 program_path = os.path.join(current_dir, "..", "bin", name) # 检查程序是否存在 if not os.path.exists(program_path): raise HTTPException( status_code=404, detail=f"程序不存在: {program_path}" ) try: # 运行程序并捕获输出 result = subprocess.run( program_path, capture_output=True, text=True, timeout=30 # 设置30秒超时 ) # 构建响应 response = { "status": "success" if result.returncode == 0 else "error", "returncode": result.returncode, "stdout": result.stdout, "stderr": result.stderr, "program_path": program_path, "executable": name } return response except Exception as e: raise HTTPException( status_code=500, detail=f"运行程序时出错: {str(e)}" ) # C++ 性能测试端点 @app.get("/cppbenchmark/") def cppbenchmark(): """性能对比测试""" # 测试数据 size = 1000 a = np.random.rand(size, size).astype(np.float64) b = np.random.rand(size, size).astype(np.float64) # Python实现（对比） def python_matmul(x, y): return np.dot(x, y) res = "ok" try: # 测试C++扩展 start = time.time() import app_cpppy # 导入C++扩展 cpp_result = app_cpppy.matrix_multiply(a, b) cpp_time = time.time() - start except Exception as e: cpp_time = time.time() - start cpp_result = 0 res = f'❌ Import failed: {e}' # 测试Python实现 start = time.time() py_result = python_matmul(a, b) py_time = time.time() - start # 验证结果一致性 diff = np.abs(cpp_result - py_result).max() return { "cpp_time": cpp_time, "python_time": py_time, "speedup": py_time / cpp_time if cpp_time > 0 else 0, "max_difference": diff, "matrix_size": size, "result": res }