القائمة
يُحدَّد Prompt الذي يراجع المصدر أولًا بشكل افتراضي. يمكنك التبديل إلى أمر مباشر أو تنزيل نسخة محلية.
اقرأ SKILL.md وأي ملفات مرافقة يعرضها SkillsMP قبل أن تقرر التثبيت.
استنادا إلى تصنيف SOC المهني
التثبيت باستخدام Codex أو Claude انسخ هذا Prompt والصقه في Codex أو Claude أو مساعد آخر ليراجع صفحة Skill ويثبّتها لك.
يتجاوز الأمر المباشر Prompt المخصّص للمراجعة. افحص المصدر قبل تشغيله.
npx skills add https://github.com/microwind/ai-skills --skill fastapi-apiيبقى الأمر في سطر واحد. مرّر أفقيًا لمراجعته كاملًا قبل النسخ.
تفضّل نسخة محلية؟ نزّل الملفات المتاحة حاليًا لدى SkillsMP.
عرض 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=
)