用 Codex 或 Claude 帮你安装 复制这段 Prompt，粘贴到 Codex、Claude 或其他助手里，让它检查 Skill 页面并帮你完成安装。
直接命令不会经过审查 Prompt；运行前请先检查来源。
npx skills add https://github.com/microwind/ai-skills --skill fastapi-api命令会保持在同一行。复制前请横向滚动并检查完整内容。
想先保存到本地？可下载 SkillsMP 当前能够提供的文件。
基于 SOC 职业分类
正在显示 SKILL.md
| name | FastAPI高性能API |
| description | 当开发FastAPI应用时，分析路由设计，优化异步处理，解决性能问题。验证API架构，设计高性能服务，和最佳实践。 |
| license | MIT |
FastAPI是Python最现代化的高性能Web框架。不当的FastAPI配置会导致性能问题、安全漏洞和维护困难。在开发FastAPI应用前需要仔细分析架构需求。
核心原则: 好的FastAPI应用应该高性能、类型安全、自动文档化、易于测试。坏的FastAPI应用会导致性能瓶颈、类型错误和维护困难。
始终:
触发短语:
问题:
API响应慢，并发能力差
错误示例:
- 同步阻塞操作
- 数据库连接未复用
- 缺乏缓存机制
- 内存泄漏问题
解决方案:
1. 使用异步操作避免阻塞
2. 配置数据库连接池
3. 添加适当的缓存策略
4. 实施内存监控和清理
问题:
数据验证不严格，类型错误频发
错误示例:
- 缺少输入验证
- 类型注解不完整
- 验证规则不合理
- 错误信息不清晰
解决方案:
1. 完善Pydantic模型定义
2. 添加严格的类型注解
3. 实施自定义验证规则
4. 提供清晰的错误信息
问题:
API文档不完整，用户体验差
错误示例:
- 缺少响应模型
- 参数描述缺失
- 示例数据不完整
- 文档配置错误
解决方案:
1. 完善响应模型定义
2. 添加详细的参数描述
3. 提供完整的示例数据
4. 正确配置文档选项
import ast
import os
import json
from typing import List, Dict, Any, Optional, Set
from dataclasses import dataclass
from pathlib import Path
import importlib.util
import sys
@dataclass
class FastAPIRoute:
"""FastAPI路由信息"""
path: str
method: str
function_name: str
file: str
line: int
issues: List[str]
@dataclass
class FastAPIModel:
"""FastAPI模型信息"""
name: str
file: str
line: int
fields: List[str]
issues: List[str]
@dataclass
class FastAPIIssue:
"""FastAPI问题"""
severity: str # critical, high, medium, low
type: str
file: str
message: str
suggestion: str
line: Optional[int] = None
:
():
.app_path = Path(app_path)
.routes: [FastAPIRoute] = []
.models: [FastAPIModel] = []
.issues: [FastAPIIssue] = []
() -> [, ]:
:
python_files = .find_python_files()
file_path python_files:
.analyze_python_file(file_path)
.analyze_dependencies()
.analyze_configuration()
.generate_report()
Exception e:
{: }
() -> [Path]:
python_files = []
root, dirs, files os.walk(.app_path):
dirs[:] = [d d dirs d.startswith() d [, , , ]]
file files:
file.endswith():
python_files.append(Path(root) / file)
python_files
() -> :
:
(file_path, , encoding=) f:
content = f.read()
tree = ast.parse(content)
fastapi_apps = .find_fastapi_apps(tree, (file_path))
.analyze_routes(tree, (file_path), fastapi_apps)
.analyze_models(tree, (file_path))
.analyze_code_quality(tree, (file_path))
Exception e:
.issues.append(FastAPIIssue(
severity=,
=,
file=(file_path),
message=,
suggestion=
))
() -> []:
fastapi_apps = []
node ast.walk(tree):
(node, ast.Assign):
target node.targets:
(target, ast.Name):
((node.value, ast.Call)
(node.value.func, ast.Name)
node.value.func. == ):
fastapi_apps.append(target.)
((node.value, ast.Call)
(node.value.func, ast.Attribute)
node.value.func.attr == ):
fastapi_apps.append(target.)
fastapi_apps
() -> :
node ast.walk(tree):
(node, ast.Call):
((node.func, ast.Attribute)
node.func.attr [, , , , , , ]):
route_info = .extract_route_info(node, file_path, fastapi_apps)
route_info:
.routes.append(route_info)
.validate_route(route_info)
() -> [FastAPIRoute]:
:
method = node.func.attr.upper()
path =
node.args (node.args[], ast.Str):
path = node.args[].s
node.args (node.args[], ast.Constant):
path = node.args[].value
function_name =
line = node.lineno
parent = .find_parent_function(node)
parent (parent, ast.FunctionDef):
function_name = parent.name
function_name:
FastAPIRoute(
path=path,
method=method,
function_name=function_name,
file=file_path,
line=line,
issues=[]
)
Exception:
() -> [ast.AST]:
() -> :
.is_restful_path(route.path):
route.issues.append()
.issues.append(FastAPIIssue(
severity=,
=,
file=route.file,
message=,
suggestion=,
line=route.line
))
.has_parameters(route.path) .has_validation(route.function_name):
route.issues.append()
.issues.append(FastAPIIssue(
severity=,
=,
file=route.file,
message=,
suggestion=,
line=route.line
))
() -> :
verbs = [, , , , , , , ]
path_parts = path.strip().split()
part path_parts:
part.lower() verbs:
() -> :
path path
() -> :
() -> :
node ast.walk(tree):
(node, ast.ClassDef):
.is_pydantic_model(node):
model_info = .parse_pydantic_model(node, file_path)
model_info:
.models.append(model_info)
.validate_model(model_info)
() -> :
node.bases:
base node.bases:
(base, ast.Name) base. == :
(base, ast.Attribute) base.attr == :
() -> [FastAPIModel]:
node.name:
model_info = FastAPIModel(
name=node.name,
file=file_path,
line=node.lineno,
fields=[],
issues=[]
)
item node.body:
(item, ast.AnnAssign) (item.target, ast.Name):
model_info.fields.append(item.target.)
model_info
() -> :
(model.fields) == :
model.issues.append()
.issues.append(FastAPIIssue(
severity=,
=,
file=model.file,
message=,
suggestion=,
line=model.line
))
model.name.endswith() model.name.endswith():
model.issues.append()
.issues.append(FastAPIIssue(
severity=,
=,
file=model.file,
message=,
suggestion=,
line=model.line
))
() -> :
node ast.walk(tree):
(node, ast.FunctionDef):
complexity = .calculate_complexity(node)
complexity > :
.issues.append(FastAPIIssue(
severity=,
=,
file=file_path,
message=,
suggestion=,
line=node.lineno
))
.check_async_usage(tree, file_path)
() -> :
complexity =
child ast.walk(node):
(child, (ast.If, ast.For, ast.While, ast.Try)):
complexity +=
(child, ast.BoolOp):
complexity += (child.values) -
complexity
() -> :
has_async_def =
has_await =
node ast.walk(tree):
(node, ast.AsyncFunctionDef):
has_async_def =
(node, ast.Await):
has_await =
has_async_def has_await:
.issues.append(FastAPIIssue(
severity=,
=,
file=file_path,
message=,
suggestion=
))
() -> :
requirements_path = .app_path /
pyproject_path = .app_path /
requirements_path.exists():
.analyze_requirements_txt(requirements_path)
pyproject_path.exists():
.analyze_pyproject_toml(pyproject_path)
:
.issues.append(FastAPIIssue(
severity=,
=,
file=,
message=,
suggestion=
))
() -> :
:
(requirements_path, , encoding=) f:
requirements = f.read().strip().split()
required_deps = [, ]
performance_deps = [, ]
dep required_deps:
(dep.lower() req.lower() req requirements):
.issues.append(FastAPIIssue(
severity=,
=,
file=(requirements_path),
message=,
suggestion=
))
dep performance_deps:
(dep.lower() req.lower() req requirements):
.issues.append(FastAPIIssue(
severity=,
=,
file=(requirements_path),
message=,
suggestion=
))
Exception e:
.issues.append(FastAPIIssue(
severity=,
=,
file=(requirements_path),
message=,
suggestion=
))
() -> :
:
(pyproject_path, , encoding=) f:
content = f.read()
content.lower():
.issues.append(FastAPIIssue(
severity=,
=,
file=(pyproject_path),
message=,
suggestion=
))
Exception e:
.issues.append(FastAPIIssue(
severity=,
=,
file=(pyproject_path),
message=,
suggestion=
))
() -> :
main_files = [, , ]
has_main_file =
main_file main_files:
(.app_path / main_file).exists():
has_main_file =
has_main_file:
.issues.append(FastAPIIssue(
severity=,
=,
file=,
message=,
suggestion=
))
() -> [, ]:
summary = {
: (.issues),
: ([i i .issues i.severity == ]),
: ([i i .issues i.severity == ]),
: ([i i .issues i.severity == ]),
: ([i i .issues i.severity == ]),
: (.routes),
: (.models)
}
recommendations = .generate_recommendations()
{
: summary,
: [.route_to_dict(r) r .routes],
: [.model_to_dict(m) m .models],
: [.issue_to_dict(i) i .issues],
: recommendations,
: .calculate_health_score(summary)
}
() -> [, ]:
{
: route.path,
: route.method,
: route.function_name,
: route.file,
: route.line,
: route.issues
}
() -> [, ]:
{
: model.name,
: model.file,
: model.line,
: model.fields,
: model.issues
}
() -> [, ]:
{
: issue.severity,
: issue.,
: issue.file,
: issue.message,
: issue.suggestion,
: issue.line
}
() -> [[, ]]:
recommendations = []
issue_types = {}
issue .issues:
issue_types[issue.] = issue_types.get(issue., ) +
issue_types.get(, ) > :
recommendations.append({
: ,
: ,
: ,
suggestion:
})
issue_types.get(, ) > :
recommendations.append({
: ,
: ,
: ,
suggestion:
})
issue_types.get(, ) > :
recommendations.append({
: ,
: ,
: ,
suggestion:
})
recommendations
() -> :
score =
score -= summary[] *
score -= summary[] *
score -= summary[] *
score -= summary[] *
(, score)
:
():
.app_path = Path(app_path)
() -> [, ]:
optimizations = []
dependency_optimization = .optimize_dependencies()
dependency_optimization:
optimizations.append(dependency_optimization)
config_optimization = .optimize_configuration()
config_optimization:
optimizations.append(config_optimization)
performance_optimization = .optimize_performance()
performance_optimization:
optimizations.append(performance_optimization)
{
: optimizations,
: {
: (optimizations),
: .estimate_improvements(optimizations)
}
}
() -> [[, ]]:
requirements_path = .app_path /
requirements_path.exists():
{
: ,
: ,
:
}
:
(requirements_path, , encoding=) f:
requirements = f.read().strip().split()
performance_deps = [, , ]
missing_deps = []
dep performance_deps:
(dep.lower() req.lower() req requirements):
missing_deps.append(dep)
missing_deps:
{
: ,
: ,
:
}
Exception:
() -> [[, ]]:
env_file = .app_path /
env_file.exists():
{
: ,
: ,
:
}
() -> [[, ]]:
main_files = [, , ]
main_file main_files:
file_path = .app_path / main_file
file_path.exists():
:
(file_path, , encoding=) f:
content = f.read()
content:
{
: ,
: ,
:
}
Exception:
() -> [, ]:
improvements = {
: ,
: ,
:
}
opt optimizations:
opt[] == :
improvements[] +=
opt[] == :
improvements[] +=
opt[] == :
improvements[] +=
improvements
():
analyzer = FastAPIAnalyzer()
report = analyzer.analyze_application()
()
()
()
()
()
()
rec report[]:
()
optimizer = FastAPIOptimizer()
optimization = optimizer.optimize_application()
()
opt optimization[]:
()
__name__ == :
main()
import time
import asyncio
from typing import Dict, Any, List
from fastapi import FastAPI, Request, Response
from fastapi.middleware import Middleware
from starlette.middleware.base import BaseHTTPMiddleware
from dataclasses import dataclass
import psutil
import uvicorn
@dataclass
class RequestMetrics:
"""请求指标"""
method: str
path: str
status_code: int
duration: float
timestamp: float
user_agent: str
ip: str
@dataclass
class PerformanceMetrics:
"""性能指标"""
cpu_percent: float
memory_percent: float
memory_mb: float
active_connections: int
requests_per_second: float
class FastAPIPerformanceMonitor:
def __init__(self, app: FastAPI):
self.app = app
self.requests: List[RequestMetrics] = []
self.start_time = time.time()
self.setup_monitoring()
():
.app.add_middleware(PerformanceMiddleware, monitor=)
():
.get_metrics()
():
{: , : time.time()}
():
metrics = RequestMetrics(
method=request.method,
path=request.url.path,
status_code=response.status_code,
duration=duration,
timestamp=time.time(),
user_agent=request.headers.get(, ),
ip=request.client.host request.client
)
.requests.append(metrics)
(.requests) > :
.requests.pop()
() -> [, ]:
request_stats = .calculate_request_stats()
performance_stats = .get_performance_stats()
route_stats = .calculate_route_stats()
{
: time.time(),
: time.time() - .start_time,
: request_stats,
: performance_stats,
: route_stats
}
() -> [, ]:
.requests:
{
: ,
: ,
: ,
: ,
: ,
:
}
total_requests = (.requests)
uptime = time.time() - .start_time
requests_per_second = total_requests / uptime uptime >
response_times = [req.duration req .requests]
average_response_time = (response_times) / (response_times)
error_requests = [req req .requests req.status_code >= ]
error_rate = (error_requests) / total_requests * total_requests >
{
: total_requests,
: requests_per_second,
: average_response_time,
: error_rate,
: .calculate_percentile(response_times, ),
: .calculate_percentile(response_times, )
}
() -> PerformanceMetrics:
PerformanceMetrics(
cpu_percent=psutil.cpu_percent(),
memory_percent=psutil.virtual_memory().percent,
memory_mb=psutil.virtual_memory().used / / ,
active_connections=(psutil.net_connections()),
requests_per_second=.calculate_current_rps()
)
() -> [[, ]]:
route_stats = {}
req .requests:
route_key =
route_key route_stats:
route_stats[route_key] = {
: ,
: ,
: ,
:
}
stats = route_stats[route_key]
stats[] +=
stats[] += req.duration
stats[] = stats[] / stats[]
req.status_code >= :
stats[] +=
(
[
{: route, **stats}
route, stats route_stats.items()
],
key= x: x[],
reverse=
)[:]
() -> :
values:
sorted_values = (values)
index = ((sorted_values) * percentile / )
sorted_values[(index, (sorted_values) - )]
() -> :
now = time.time()
recent_requests = [req req .requests now - req.timestamp < ]
(recent_requests) /
():
():
().__init__(app)
.monitor = monitor
():
start_time = time.time()
response = call_next(request)
duration = time.time() - start_time
.monitor.record_request(request, response, duration)
response
():
():
():
start_time = time.time()
:
result = func(*args, **kwargs)
result
:
duration = time.time() - start_time
wrapper
decorator
():
app = FastAPI(title=)
monitor = FastAPIPerformanceMonitor(app)
():
{: }
():
asyncio.sleep()
{: []}
():
asyncio.sleep()
{: user_data, : }
app
__name__ == :
app = create_app()
uvicorn.run(
app,
host=,
port=,
log_level=,
access_log=
)