| name | Flask/Django分析器 |
| description | 当审查Flask/Django代码、规划项目结构、调试框架问题或优化框架性能时，分析Flask/Django Web框架的最佳实践和架构模式。 |
| license | MIT |
Flask/Django分析器技能
概述
Web框架实现快速开发，但不良模式会创建维护噩梦。
核心原则: Web框架实现快速开发，但不良模式会创建维护噩梦。
何时使用
始终:
- 审查Flask/Django代码
- 规划Flask/Django项目结构
- 调试框架问题
- 优化框架性能
- 架构设计评审
- 安全性分析
触发短语:
- "分析Flask代码"
- "Django项目架构"
- "Web框架最佳实践"
- "Flask/Django性能优化"
- "框架安全检查"
- "项目结构规划"
Flask/Django分析功能
代码质量
- MVC/MVT模式检查
- 蓝图/应用结构分析
- 模型设计验证
- 视图函数审查
- 模板使用检查
性能分析
- 数据库查询优化
- 缓存策略分析
- 中间件性能检查
- 静态文件优化
- 会话管理评估
安全检查
- CSRF保护验证
- SQL注入防护
- XSS漏洞检测
- 认证授权检查
- 敏感数据保护
常见Flask/Django问题
架构问题
问题:
项目结构混乱，违反框架设计原则
错误示例:
- 所有代码放在一个文件
- 模型与视图耦合
- 缺少模块化设计
- 业务逻辑分散
解决方案:
1. 遵循MVC/MVT架构
2. 使用蓝图/应用分离
3. 创建清晰的模块结构
4. 实现关注点分离
数据库问题
问题:
数据库设计不当，查询效率低下
错误示例:
- 缺少索引优化
- N+1查询问题
- 事务管理不当
- 数据模型设计不合理
解决方案:
1. 优化数据库索引
2. 使用select_related/prefetch_related
3. 合理管理事务
4. 规范数据模型设计
安全问题
问题:
Web应用存在安全漏洞
错误示例:
- 缺少CSRF保护
- SQL注入风险
- XSS攻击漏洞
- 认证机制不完善
解决方案:
1. 启用CSRF保护
2. 使用ORM防止SQL注入
3. 模板自动转义
4. 实现完善的认证系统
代码实现示例
Flask代码分析器
import ast
import os
import re
from typing import List, Dict, Any, Optional, Tuple
from dataclasses import dataclass
from collections import defaultdict
@dataclass
class FlaskIssue:
"""Flask代码问题"""
file_path: str
line_number: int
column: int
severity: str
message: str
rule_id: str
suggestion: Optional[str] = None
@dataclass
class RouteMetrics:
"""路由指标"""
endpoint: str
methods: List[str]
line_number: int
has_auth: bool
has_validation: bool
complexity: int
docstring: bool
@dataclass
class ModelMetrics:
"""模型指标"""
name: str
fields_count: int
relationships_count: int
has_indexes: bool
has_validators: bool
docstring: bool
:
():
.project_root = project_root
.issues: [FlaskIssue] = []
.route_metrics: [RouteMetrics] = []
.model_metrics: [ModelMetrics] = []
.config_files: [] = []
.template_files: [] = []
() -> [, ]:
results = {
: .analyze_project_structure(),
: .analyze_routes(),
: .analyze_models(),
: .analyze_templates(),
: .analyze_config(),
: .analyze_security(),
: .analyze_performance(),
: .issues,
: .generate_summary()
}
results
() -> [, ]:
structure = {
: ,
: ,
: ,
: ,
: ,
: {},
: []
}
root, dirs, files os.walk(.project_root):
rel_path = os.path.relpath(root, .project_root)
structure[][rel_path] = {
: dirs,
: files
}
files files:
(os.path.join(root, files ), ) f:
content = f.read()
content:
structure[] =
dirs ( f.lower() f files):
structure[] =
files files:
structure[] =
dirs dirs:
structure[] =
dirs dirs:
structure[] =
structure[]:
structure[].append({
: ,
: ,
:
})
structure[]:
structure[].append({
: ,
: ,
:
})
structure
() -> [, ]:
route_files = .find_files([, , , ])
file_path route_files:
.analyze_route_file(file_path)
{
: (.route_metrics),
: .route_metrics,
: [issue issue .issues issue.rule_id.startswith()]
}
() -> :
:
(file_path, , encoding=) f:
content = f.read()
tree = ast.parse(content, filename=file_path)
node ast.walk(tree):
(node, ast.FunctionDef):
decorators = [d d node.decorator_list
(d, ast.Call)
(d.func, ast.Name)
d.func. [, ]]
decorators:
.analyze_route_function(node, file_path, decorators)
Exception e:
.issues.append(FlaskIssue(
file_path=file_path,
line_number=,
column=,
severity=,
message=,
rule_id=
))
() -> :
route_info = .extract_route_info(decorators[])
has_auth = .check_route_auth(node)
has_validation = .check_route_validation(node)
complexity = .calculate_complexity(node)
has_docstring = ast.get_docstring(node)
metrics = RouteMetrics(
endpoint=route_info.get(, ),
methods=route_info.get(, []),
line_number=node.lineno,
has_auth=has_auth,
has_validation=has_validation,
complexity=complexity,
docstring=has_docstring
)
.route_metrics.append(metrics)
has_auth (method [, , ] method metrics.methods):
.issues.append(FlaskIssue(
file_path=file_path,
line_number=node.lineno,
column=,
severity=,
message=,
rule_id=,
suggestion=
))
has_validation (method [, , ] method metrics.methods):
.issues.append(FlaskIssue(
file_path=file_path,
line_number=node.lineno,
column=,
severity=,
message=,
rule_id=,
suggestion=
))
complexity > :
.issues.append(FlaskIssue(
file_path=file_path,
line_number=node.lineno,
column=,
severity=,
message=,
rule_id=,
suggestion=
))
() -> [, ]:
info = {: , : []}
decorator.args:
(decorator.args[], ast.Str):
info[] = decorator.args[].s
(decorator.args[], ast.Constant):
info[] = decorator.args[].value
keyword decorator.keywords:
keyword.arg == :
(keyword.value, ast.):
methods = []
elt keyword.value.elts:
(elt, ast.Str):
methods.append(elt.s)
(elt, ast.Constant):
methods.append(elt.value)
info[] = methods
info
() -> :
auth_decorators = []
decorator node.decorator_list:
(decorator, ast.Name):
decorator. [, , ]:
auth_decorators.append(decorator.)
(decorator, ast.Call):
(decorator.func, ast.Name):
decorator.func. [, , ]:
auth_decorators.append(decorator.func.)
(auth_decorators) >
() -> :
validation_patterns = [
,
,
,
,
,
,
]
child ast.walk(node):
(child, ast.Call):
(child.func, ast.Attribute):
(pattern child.func.attr pattern validation_patterns):
(child.func, ast.Name):
child.func. validation_patterns:
() -> [, ]:
model_files = .find_files([, ])
file_path model_files:
.analyze_model_file(file_path)
{
: (.model_metrics),
: .model_metrics,
: [issue issue .issues issue.rule_id.startswith()]
}
() -> :
:
(file_path, , encoding=) f:
content = f.read()
tree = ast.parse(content, filename=file_path)
node ast.walk(tree):
(node, ast.ClassDef):
.is_model_class(node):
.analyze_model_class(node, file_path)
Exception e:
.issues.append(FlaskIssue(
file_path=file_path,
line_number=,
column=,
severity=,
message=,
rule_id=
))
() -> :
base node.bases:
(base, ast.Name):
base. [, , ]:
(base, ast.Attribute):
base.attr == :
decorator node.decorator_list:
(decorator, ast.Name):
decorator. == :
() -> :
fields_count =
relationships_count =
has_indexes =
has_validators =
item node.body:
(item, ast.Assign):
target item.targets:
(target, ast.Name):
(item.value, ast.Call):
(item.value.func, ast.Name):
item.value.func. [, , , ]:
fields_count +=
keyword item.value.keywords:
keyword.arg == keyword.value.value :
has_indexes =
(item.value.func, ast.Attribute):
item.value.func.attr [, , ]:
item.value.func.attr == :
relationships_count +=
:
fields_count +=
has_docstring = ast.get_docstring(node)
metrics = ModelMetrics(
name=node.name,
fields_count=fields_count,
relationships_count=relationships_count,
has_indexes=has_indexes,
has_validators=has_validators,
docstring=has_docstring
)
.model_metrics.append(metrics)
fields_count > :
.issues.append(FlaskIssue(
file_path=file_path,
line_number=node.lineno,
column=,
severity=,
message=,
rule_id=,
suggestion=
))
has_docstring:
.issues.append(FlaskIssue(
file_path=file_path,
line_number=node.lineno,
column=,
severity=,
message=,
rule_id=,
suggestion=
))
() -> [, ]:
template_dirs = [, ]
template_dir template_dirs:
template_path = os.path.join(.project_root, template_dir)
os.path.exists(template_path):
.analyze_template_directory(template_path)
{
: (.template_files),
: [issue issue .issues issue.rule_id.startswith()]
}
() -> :
root, dirs, files os.walk(template_dir):
file files:
file.endswith((, , , )):
file_path = os.path.join(root, file)
.analyze_template_file(file_path)
() -> :
:
(file_path, , encoding=) f:
content = f.read()
.template_files.append(file_path)
.check_template_security(file_path, content)
.check_template_structure(file_path, content)
Exception e:
.issues.append(FlaskIssue(
file_path=file_path,
line_number=,
column=,
severity=,
message=,
rule_id=
))
() -> :
lines = content.split()
line_num, line (lines, ):
line line:
.issues.append(FlaskIssue(
file_path=file_path,
line_number=line_num,
column=line.find(),
severity=,
message=,
rule_id=,
suggestion=
))
line line line line:
() -> :
lines = content.split()
has_extends = ( line line lines)
has_extends file_path.endswith():
is_base_template = ( line line lines)
is_base_template:
.issues.append(FlaskIssue(
file_path=file_path,
line_number=,
column=,
severity=,
message=,
rule_id=,
suggestion=
))
() -> [, ]:
config_files = .find_files([, , , ])
file_path config_files:
.analyze_config_file(file_path)
{
: (.config_files),
: [issue issue .issues issue.rule_id.startswith()]
}
() -> :
:
(file_path, , encoding=) f:
content = f.read()
.check_config_security(file_path, content)
.check_config_structure(file_path, content)
Exception e:
.issues.append(FlaskIssue(
file_path=file_path,
line_number=,
column=,
severity=,
message=,
rule_id=
))
() -> :
lines = content.split()
line_num, line (lines, ):
line line:
value = line.split()[].strip()
value.startswith() value.startswith():
.issues.append(FlaskIssue(
file_path=file_path,
line_number=line_num,
column=line.find(),
severity=,
message=,
rule_id=,
suggestion=
))
line line:
.issues.append(FlaskIssue(
file_path=file_path,
line_number=line_num,
column=line.find(),
severity=,
message=,
rule_id=,
suggestion=
))
() -> :
has_env_vars = content content
has_env_vars:
.issues.append(FlaskIssue(
file_path=file_path,
line_number=,
column=,
severity=,
message=,
rule_id=,
suggestion=
))
() -> [, ]:
security_issues = []
.has_csrf_protection():
security_issues.append({
: ,
: ,
: ,
:
})
.has_security_headers():
security_issues.append({
: ,
: ,
: ,
:
})
{
: security_issues,
: .calculate_security_score(security_issues)
}
() -> :
root, dirs, files os.walk(.project_root):
file files:
file.endswith():
file_path = os.path.join(root, file)
:
(file_path, , encoding=) f:
content = f.read()
content.lower() content:
:
() -> :
() -> [, ]:
performance_issues = []
.has_cache_config():
performance_issues.append({
: ,
: ,
: ,
:
})
.has_connection_pool():
performance_issues.append({
: ,
: ,
: ,
:
})
{
: performance_issues,
: .calculate_performance_score(performance_issues)
}
() -> :
() -> :
() -> :
complexity =
child ast.walk(node):
(child, (ast.If, ast.While, ast.For, ast.ExceptHandler)):
complexity +=
(child, ast.BoolOp):
complexity += (child.values) -
complexity
() -> []:
found_files = []
root, dirs, files os.walk(.project_root):
file files:
file filenames:
found_files.append(os.path.join(root, file))
found_files
() -> :
score =
issue issues:
issue[] == :
score -=
issue[] == :
score -=
issue[] == :
score -=
(, score)
() -> :
score =
issue issues:
issue[] == :
score -=
issue[] == :
score -=
issue[] == :
score -=
(, score)
() -> [, ]:
total_issues = (.issues)
error_count = ([i i .issues i.severity == ])
warning_count = ([i i .issues i.severity == ])
info_count = ([i i .issues i.severity == ])
{
: total_issues,
: error_count,
: warning_count,
: info_count,
: (, - error_count * - warning_count * - info_count),
: .generate_recommendations()
}
() -> [[, ]]:
recommendations = []
issue_counts = defaultdict()
issue .issues:
issue_counts[issue.rule_id] +=
issue_counts[] > :
recommendations.append({
: ,
: ,
:
})
issue_counts[] > :
recommendations.append({
: ,
: ,
:
})
issue_counts[] > :
recommendations.append({
: ,
: ,
:
})
recommendations
:
():
.project_root = project_root
.issues: [FlaskIssue] = []
() -> [, ]:
{
: .analyze_settings(),
: .analyze_urls(),
: .analyze_views(),
: .analyze_models(),
: .analyze_admin(),
: .issues
}
() -> [, ]:
settings_files = [, , ]
issues = []
settings_file settings_files:
settings_path = os.path.join(.project_root, settings_file)
os.path.exists(settings_path):
{: issues}
() -> [, ]:
url_files = []
issues = []
root, dirs, files os.walk(.project_root):
file files:
file == :
{: issues}
() -> [, ]:
view_files = []
issues = []
root, dirs, files os.walk(.project_root):
file files:
file == :
{: issues}
() -> [, ]:
model_files = []
issues = []
root, dirs, files os.walk(.project_root):
file files:
file == :
{: issues}
() -> [, ]:
admin_files = []
issues = []
root, dirs, files os.walk(.project_root):
file files:
file == :
{: issues}
():
project_path =
flask_analyzer = FlaskAnalyzer(project_path)
flask_result = flask_analyzer.analyze_project()
()
()
()
django_analyzer = DjangoAnalyzer(project_path)
django_result = django_analyzer.analyze_project()
()
()
__name__ == :
main()
Flask/Django性能分析器
import time
import psutil
import threading
from typing import Dict, List, Any, Callable
from dataclasses import dataclass
from functools import wraps
@dataclass
class PerformanceMetric:
"""性能指标"""
name: str
value: float
unit: str
timestamp: float
metadata: Dict[str, Any]
class FlaskDjangoPerformanceAnalyzer:
def __init__(self):
self.metrics: List[PerformanceMetric] = []
self.is_monitoring = False
self.monitor_thread = None
def start_monitoring(self, interval: int = 5) -> None:
"""开始性能监控"""
self.is_monitoring = True
self.monitor_thread = threading.Thread(
target=self._monitor_loop,
args=(interval,),
daemon=True
)
self.monitor_thread.start()
() -> :
.is_monitoring =
.monitor_thread:
.monitor_thread.join()
() -> :
.is_monitoring:
._collect_system_metrics()
time.sleep(interval)
() -> :
cpu_percent = psutil.cpu_percent()
.record_metric(, cpu_percent, )
memory = psutil.virtual_memory()
.record_metric(, memory.percent, )
.record_metric(, memory.available, )
disk = psutil.disk_usage()
.record_metric(, disk.percent, )
.record_metric(, disk.free, )
() -> :
metric = PerformanceMetric(
name=name,
value=value,
unit=unit,
timestamp=time.time(),
metadata=metadata {}
)
.metrics.append(metric)
(.metrics) > :
.metrics = .metrics[-:]
() -> :
() -> :
():
start_time = time.time()
start_memory = psutil.Process().memory_info().rss
:
result = func(*args, **kwargs)
success =
error =
Exception e:
success =
error = (e)
:
end_time = time.time()
end_memory = psutil.Process().memory_info().rss
.record_metric(,
end_time - start_time, ,
{: success, : error})
.record_metric(,
end_memory - start_memory, ,
{: success, : error})
result
wrapper
decorator
() -> :
() -> :
():
start_time = time.time()
:
result = func(*args, **kwargs)
query_count = (result, , )
success =
Exception e:
query_count =
success =
:
end_time = time.time()
.record_metric(,
end_time - start_time, ,
{: query_count, : success})
.record_metric(,
query_count, ,
{: success})
result
wrapper
decorator
() -> :
request_data:
.record_metric(,
request_data[], ,
{: request_data.get(, )})
request_data:
.record_metric(,
request_data[], ,
{: request_data.get(, )})
() -> [, ]:
report = {
: time.time(),
: ._generate_summary(),
: ._group_metrics_by_type(),
: ._analyze_trends(),
: ._generate_performance_recommendations()
}
report
() -> [, ]:
.metrics:
{: }
metrics_by_type = defaultdict()
metric .metrics:
metrics_by_type[metric.name].append(metric)
summary = {}
metric_type, metrics metrics_by_type.items():
metrics:
values = [m.value m metrics]
summary[metric_type] = {
: (values) / (values),
: (values),
: (values),
: (values),
: metrics[].unit
}
summary
() -> [, [PerformanceMetric]]:
grouped = defaultdict()
metric .metrics:
grouped[metric.name].append(metric)
(grouped)
() -> [, ]:
trends = {}
grouped = ._group_metrics_by_type()
metric_type, metrics grouped.items():
(metrics) >= :
recent = metrics[-].value
earliest = metrics[].value
recent > earliest * :
trends[metric_type] =
recent < earliest * :
trends[metric_type] =
:
trends[metric_type] =
trends
() -> [[, ]]:
recommendations = []
summary = ._generate_summary()
summary summary[][] > :
recommendations.append({
: ,
: ,
: ,
:
})
summary summary[][] > :
recommendations.append({
: ,
: ,
: ,
:
})
request_metrics = [m m .metrics m.name]
request_metrics:
avg_response_time = (m.value m request_metrics) / (request_metrics)
avg_response_time > :
recommendations.append({
: ,
: ,
: ,
:
})
recommendations
():
analyzer = FlaskDjangoPerformanceAnalyzer()
analyzer.start_monitoring()
():
time.sleep()
():
time.sleep()
[]
home_view()
get_users()
analyzer.analyze_request_performance({
: ,
: ,
:
})
time.sleep()
analyzer.stop_monitoring()
report = analyzer.generate_performance_report()
(, report)
__name__ == :
main()
Flask/Django最佳实践
项目结构
- 应用工厂: 使用create_app模式
- 蓝图组织: 按功能模块分离
- 配置分离: 环境变量管理配置
- 测试覆盖: 完整的测试套件
数据库设计
- 模型规范: 遵循数据库设计原则
- 索引优化: 合理创建索引
- 查询优化: 避免N+1问题
- 事务管理: 正确使用事务
安全实践
- CSRF保护: 启用跨站请求伪造保护
- 输入验证: 验证所有用户输入
- SQL注入防护: 使用ORM参数化查询
- 认证授权: 实现完善的权限控制
性能优化
- 缓存策略: 合理使用Redis/Memcached
- 数据库优化: 连接池和查询优化
- 静态文件: CDN加速和压缩
- 异步任务: 使用Celery处理耗时任务
相关技能
- python-analyzer - Python代码分析
- sql-optimizer - SQL优化
- web-security - Web安全
- api-design - API设计
