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 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", 8001),
        "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', 8001)}")


# 路由测试
@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>"