app_fast/app_fast.py

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/<name>/")
@app.get("/hi/")  # 兼容无参数情况
def hi_page(name=None):
    if name is None:
        name = "访客"
    return f"<h1>你好，{name}！</h1><a href='/'>返回首页</a>"

# 运行外部程序路由
@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
    }