Spaces:
Sleeping
Sleeping
File size: 5,317 Bytes
0c62be0 9429cb8 7a86764 0c62be0 7a86764 0c62be0 7a86764 0c62be0 b482470 0c62be0 b482470 0c62be0 7a86764 0c62be0 7a86764 ba1bdb9 7a86764 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 | 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>" |