| name | Django Web框架 |
| description | 当开发Django应用时,分析项目结构,优化ORM查询,解决性能问题。验证MVC架构,设计企业级Web应用,和最佳实践。 |
| license | MIT |
Django Web框架技能
概述
Django是Python最强大的Web框架。不当的Django配置会导致性能问题、安全漏洞和维护困难。在开发Django应用前需要仔细分析项目需求。
核心原则: 好的Django应用应该遵循MTV模式、高效ORM、安全配置、易于扩展。坏的Django应用会导致性能瓶颈、安全风险和维护困难。
何时使用
始终:
- 开发企业级Web应用时
- 构建内容管理系统时
- 实现用户认证系统时
- 配置数据库ORM时
- 处理表单验证时
触发短语:
- "Django Web框架"
- "Python Web开发"
- "ORM查询优化"
- "MVC架构设计"
- "Django项目结构"
- "企业级Web应用"
Django Web框架功能
MTV架构
- 模型层设计
- 视图层处理
- 模板层渲染
- URL路由配置
- 中间件系统
ORM系统
- 模型定义
- 查询优化
- 关系映射
- 数据迁移
- 原生SQL支持
用户认证
管理后台
- 自动管理界面
- 自定义管理
- 批量操作
- 搜索过滤
- 导入导出
常见Django问题
ORM性能问题
问题:
数据库查询慢,N+1查询问题
错误示例:
- 缺少select_related/prefetch_related
- 循环查询数据库
- 未使用数据库索引
- 复杂聚合查询
解决方案:
1. 使用select_related优化外键查询
2. 使用prefetch_related优化多对多查询
3. 添加适当的数据库索引
4. 使用缓存机制
安全配置问题
问题:
安全漏洞,数据泄露风险
错误示例:
- 调试模式在生产环境开启
- CSRF保护未启用
- SQL注入风险
- XSS攻击漏洞
解决方案:
1. 生产环境关闭DEBUG模式
2. 启用CSRF保护中间件
3. 使用ORM防止SQL注入
4. 模板自动转义防止XSS
项目结构问题
问题:
代码组织混乱,维护困难
错误示例:
- 应用职责不清晰
- 配置文件混乱
- 静态文件管理不当
- 模板结构不合理
解决方案:
1. 合理划分应用职责
2. 分环境配置管理
3. 统一静态文件处理
4. 优化模板继承结构
代码实现示例
Django项目分析器
import os
import re
import ast
from pathlib import Path
from typing import List, Dict, Any, Optional, Set
from dataclasses import dataclass
import importlib.util
@dataclass
class DjangoApp:
"""Django应用信息"""
name: str
path: str
models: List[str]
views: List[str]
urls: List[str]
issues: List[str]
@dataclass
class DjangoModel:
"""Django模型信息"""
name: str
file: str
fields: List[str]
relationships: List[str]
issues: List[str]
@dataclass
class DjangoIssue:
"""Django问题"""
severity: str
type: str
file: str
message: str
suggestion: str
line: Optional[] =
:
():
.project_path = Path(project_path)
.apps: [DjangoApp] = []
.models: [DjangoModel] = []
.issues: [DjangoIssue] = []
() -> [, ]:
:
.check_project_structure()
.analyze_settings()
.scan_apps()
.analyze_models()
.analyze_views()
.analyze_urls()
.generate_report()
Exception e:
{: }
() -> :
manage_py = .project_path /
manage_py.exists():
.issues.append(DjangoIssue(
severity=,
=,
file=,
message=,
suggestion=
))
settings_py = .project_path /
settings_py.exists():
settings_dir = .project_path /
settings_dir.exists():
.issues.append(DjangoIssue(
severity=,
=,
file=,
message=,
suggestion=
))
() -> :
settings_files = []
settings_py = .project_path /
settings_py.exists():
settings_files.append(settings_py)
:
settings_dir = .project_path /
settings_dir.exists():
file settings_dir.glob():
file.name.startswith():
settings_files.append(file)
settings_file settings_files:
.analyze_settings_file(settings_file)
() -> :
:
(settings_file, , encoding=) f:
content = f.read()
content:
.issues.append(DjangoIssue(
severity=,
=,
file=(settings_file),
message=,
suggestion=
))
content content.startswith():
content content.split()[].split()[]:
.issues.append(DjangoIssue(
severity=,
=,
file=(settings_file),
message=,
suggestion=
))
content:
.issues.append(DjangoIssue(
severity=,
=,
file=(settings_file),
message=,
suggestion=
))
required_middlewares = [
,
,
,
,
,
]
middleware required_middlewares:
middleware content:
.issues.append(DjangoIssue(
severity=,
=,
file=(settings_file),
message=,
suggestion=
))
content:
.issues.append(DjangoIssue(
severity=,
=,
file=(settings_file),
message=,
suggestion=
))
Exception e:
.issues.append(DjangoIssue(
severity=,
=,
file=(settings_file),
message=,
suggestion=
))
() -> :
item .project_path.iterdir():
item.is_dir() item.name.startswith() item.name != :
.is_django_app(item):
app_info = .analyze_app(item)
app_info:
.apps.append(app_info)
() -> :
init_file = app_path /
init_file.exists():
required_files = [, ]
file_name required_files:
(app_path / file_name).exists():
() -> [DjangoApp]:
app_name = app_path.name
app_info = DjangoApp(
name=app_name,
path=(app_path),
models=[],
views=[],
urls=[],
issues=[]
)
models_file = app_path /
models_file.exists():
app_info.models = .extract_model_names(models_file)
views_file = app_path /
views_file.exists():
app_info.views = .extract_view_names(views_file)
urls_file = app_path /
urls_file.exists():
app_info.urls = .extract_url_patterns(urls_file)
.validate_app(app_info)
app_info
() -> []:
:
(models_file, , encoding=) f:
content = f.read()
model_names = []
tree = ast.parse(content)
node ast.walk(tree):
(node, ast.ClassDef):
.is_django_model(node):
model_names.append(node.name)
model_names
Exception:
[]
() -> :
node.bases:
base node.bases:
(base, ast.Name) base. == :
(base, ast.Attribute):
base.attr == :
() -> []:
:
(views_file, , encoding=) f:
content = f.read()
view_names = []
tree = ast.parse(content)
node ast.walk(tree):
(node, ast.FunctionDef):
.is_django_view(node):
view_names.append(node.name)
(node, ast.ClassDef):
.is_django_class_view(node):
view_names.append(node.name)
view_names
Exception:
[]
() -> :
node.args.args (node.args.args) >= :
first_arg = node.args.args[]
(first_arg, ast.arg) first_arg.arg == :
() -> :
node.bases:
base node.bases:
(base, ast.Name):
view_bases = [, , , , , , ]
base. view_bases:
(base, ast.Attribute):
base.attr view_bases:
() -> []:
:
(urls_file, , encoding=) f:
content = f.read()
url_patterns = []
path_pattern =
re_pattern =
paths = re.findall(path_pattern, content)
urls = re.findall(re_pattern, content)
url_patterns.extend(paths)
url_patterns.extend(urls)
url_patterns
Exception:
[]
() -> :
(app.models) == :
app.issues.append()
.issues.append(DjangoIssue(
severity=,
=,
file=app.path,
message=,
suggestion=
))
(app.views) == :
app.issues.append()
.issues.append(DjangoIssue(
severity=,
=,
file=app.path,
message=,
suggestion=
))
(app.urls) == :
app.issues.append()
.issues.append(DjangoIssue(
severity=,
=,
file=app.path,
message=,
suggestion=
))
() -> :
app .apps:
models_file = Path(app.path) /
models_file.exists():
.analyze_model_file(models_file, app.name)
() -> :
:
(models_file, , encoding=) f:
content = f.read()
tree = ast.parse(content)
node ast.walk(tree):
(node, ast.ClassDef) .is_django_model(node):
model_info = .parse_django_model(node, (models_file))
model_info:
.models.append(model_info)
.validate_model(model_info)
Exception e:
.issues.append(DjangoIssue(
severity=,
=,
file=(models_file),
message=,
suggestion=
))
() -> [DjangoModel]:
node.name:
model_info = DjangoModel(
name=node.name,
file=file_path,
fields=[],
relationships=[],
issues=[]
)
item node.body:
(item, ast.Assign):
target item.targets:
(target, ast.Name):
field_name = target.
model_info.fields.append(field_name)
.is_relationship_field(item):
model_info.relationships.append(field_name)
model_info
() -> :
node.value:
relationship_fields = [
, , ,
]
(node.value, ast.Call):
(node.value.func, ast.Name):
node.value.func. relationship_fields
(node.value.func, ast.Attribute):
node.value.func.attr relationship_fields
() -> :
(model.fields) == :
model.issues.append()
.issues.append(DjangoIssue(
severity=,
=,
file=model.file,
message=,
suggestion=,
line=
))
model.fields model.fields:
has_primary_key =
field model.fields:
field.lower() [, , ]:
has_primary_key =
has_primary_key:
model.issues.append()
.issues.append(DjangoIssue(
severity=,
=,
file=model.file,
message=,
suggestion=
))
() -> :
app .apps:
views_file = Path(app.path) /
views_file.exists():
.analyze_view_file(views_file, app.name)
() -> :
:
(views_file, , encoding=) f:
content = f.read()
tree = ast.parse(content)
node ast.walk(tree):
(node, ast.FunctionDef) .is_django_view(node):
.validate_view_function(node, (views_file))
(node, ast.ClassDef) .is_django_class_view(node):
.validate_class_view(node, (views_file))
Exception e:
.issues.append(DjangoIssue(
severity=,
=,
file=(views_file),
message=,
suggestion=
))
() -> :
has_return =
child ast.walk(node):
(child, ast.Return):
has_return =
has_return:
.issues.append(DjangoIssue(
severity=,
=,
file=file_path,
message=,
suggestion=,
line=node.lineno
))
() -> :
has_valid_base =
base node.bases:
(base, ast.Name) base. == :
has_valid_base =
(base, ast.Attribute) base.attr == :
has_valid_base =
has_valid_base:
.issues.append(DjangoIssue(
severity=,
=,
file=file_path,
message=,
suggestion=,
line=node.lineno
))
() -> :
main_urls = .project_path /
main_urls.exists():
.analyze_url_file(main_urls, )
app .apps:
app_urls = Path(app.path) /
app_urls.exists():
.analyze_url_file(app_urls, app.name)
() -> :
:
(urls_file, , encoding=) f:
content = f.read()
content:
.issues.append(DjangoIssue(
severity=,
=,
file=(urls_file),
message=,
suggestion=
))
context == :
content:
.issues.append(DjangoIssue(
severity=,
=,
file=(urls_file),
message=,
suggestion=
))
Exception e:
.issues.append(DjangoIssue(
severity=,
=,
file=(urls_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 == ]),
: (.apps),
: (.models)
}
recommendations = .generate_recommendations()
{
: summary,
: [.app_to_dict(app) app .apps],
: [.model_to_dict(model) model .models],
: [.issue_to_dict(issue) issue .issues],
: recommendations,
: .calculate_health_score(summary)
}
() -> [, ]:
{
: app.name,
: app.path,
: app.models,
: app.views,
: app.urls,
: app.issues
}
() -> [, ]:
{
: model.name,
: model.file,
: model.fields,
: model.relationships,
: 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)
:
():
.project_path = Path(project_path)
() -> [, ]:
optimizations = []
config_optimization = .optimize_configuration()
config_optimization:
optimizations.append(config_optimization)
db_optimization = .optimize_database()
db_optimization:
optimizations.append(db_optimization)
performance_optimization = .optimize_performance()
performance_optimization:
optimizations.append(performance_optimization)
{
: optimizations,
: {
: (optimizations),
: .estimate_improvements(optimizations)
}
}
() -> [[, ]]:
settings_files = .find_settings_files()
settings_file settings_files:
:
(settings_file, , encoding=) f:
content = f.read()
content:
{
: ,
: ,
:
}
Exception:
() -> [[, ]]:
settings_files = .find_settings_files()
settings_file settings_files:
:
(settings_file, , encoding=) f:
content = f.read()
content:
{
: ,
: ,
:
}
Exception:
() -> [[, ]]:
settings_files = .find_settings_files()
settings_file settings_files:
:
(settings_file, , encoding=) f:
content = f.read()
content:
{
: ,
: ,
:
}
Exception:
() -> [Path]:
settings_files = []
settings_py = .project_path /
settings_py.exists():
settings_files.append(settings_py)
settings_dir = .project_path /
settings_dir.exists():
file settings_dir.glob():
file.name.startswith():
settings_files.append(file)
settings_files
() -> [, ]:
improvements = {
: ,
: ,
:
}
opt optimizations:
opt[] == :
improvements[] +=
opt[] == :
improvements[] +=
opt[] == :
improvements[] +=
improvements
():
analyzer = DjangoAnalyzer()
report = analyzer.analyze_project()
()
()
()
()
()
()
rec report[]:
()
optimizer = DjangoOptimizer()
optimization = optimizer.optimize_project()
()
opt optimization[]:
()
__name__ == :
main()
Django性能监控器
import time
import psutil
from django.conf import settings
from django.db import connection
from django.core.cache import cache
from django.http import JsonResponse
from django.views.decorators.http import require_http_methods
from django.utils.decorators import method_decorator
from django.views import View
from dataclasses import dataclass
from typing import Dict, Any, List
import threading
import json
@dataclass
class RequestMetrics:
"""请求指标"""
method: str
path: str
status_code: int
duration: float
timestamp: float
user_agent: str
ip: str
db_queries: int
@dataclass
class DatabaseMetrics:
"""数据库指标"""
queries_count: int
queries_time: float
slow_queries: List[str]
connection_usage: int
class DjangoPerformanceMonitor:
_instance = None
_lock = threading.Lock()
():
cls._instance :
cls._lock:
cls._instance :
cls._instance = ().__new__(cls)
cls._instance
():
(, ):
.requests: [RequestMetrics] = []
.start_time = time.time()
.initialized =
():
metrics = RequestMetrics(
method=request.method,
path=request.path,
status_code=response.status_code,
duration=duration,
timestamp=time.time(),
user_agent=request.META.get(, ),
ip=.get_client_ip(request),
db_queries=db_queries
)
.requests.append(metrics)
(.requests) > :
.requests.pop()
() -> :
x_forwarded_for = request.META.get()
x_forwarded_for:
ip = x_forwarded_for.split()[]
:
ip = request.META.get()
ip
() -> [, ]:
request_stats = .calculate_request_stats()
performance_stats = .get_performance_stats()
database_stats = .get_database_stats()
cache_stats = .get_cache_stats()
{
: time.time(),
: time.time() - .start_time,
: request_stats,
: performance_stats,
: database_stats,
: cache_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 >
db_queries = [req.db_queries req .requests]
average_db_queries = (db_queries) / (db_queries)
{
: total_requests,
: requests_per_second,
: average_response_time,
: error_rate,
: .calculate_percentile(response_times, ),
: .calculate_percentile(response_times, ),
: average_db_queries
}
() -> [, ]:
{
: psutil.cpu_percent(),
: psutil.virtual_memory().percent,
: psutil.virtual_memory().used / / ,
: (psutil.net_connections()),
: (psutil.pids())
}
() -> [, ]:
:
connections = connection.connection
{
: connections ,
: connections.vendor connections ,
: connections.settings_dict.get(, ) connections ,
: connections.settings_dict.get(, ) connections
}
Exception:
{
: ,
:
}
() -> [, ]:
:
cache_key =
cache.(cache_key, , )
cache_value = cache.get(cache_key)
cache.delete(cache_key)
{
: cache_value == ,
: (settings, , {}).get(, {}).get(, )
}
Exception:
{
: ,
:
}
() -> :
values:
sorted_values = (values)
index = ((sorted_values) * percentile / )
sorted_values[(index, (sorted_values) - )]
:
():
.get_response = get_response
.monitor = DjangoPerformanceMonitor()
():
start_time = time.time()
initial_queries = (connection.queries)
response = .get_response(request)
duration = time.time() - start_time
settings.DEBUG:
db_queries = (connection.queries) - initial_queries
:
db_queries =
.monitor.record_request(request, response, duration, db_queries)
response
():
():
monitor = DjangoPerformanceMonitor()
metrics = monitor.get_metrics()
JsonResponse(metrics)
():
():
monitor = DjangoPerformanceMonitor()
health_status = {
: ,
: time.time(),
: time.time() - monitor.start_time
}
:
django.db connection
connection.cursor() cursor:
cursor.execute()
health_status[] =
Exception e:
health_status[] =
health_status[] =
:
cache.(, , )
cache_value = cache.get()
cache_value == :
health_status[] =
:
health_status[] =
cache.delete()
Exception e:
health_status[] =
status_code = health_status[] ==
JsonResponse(health_status, status=status_code)
():
():
start_time = time.time()
:
result = func(*args, **kwargs)
result
:
duration = time.time() - start_time
()
wrapper
():
():
queryset = func(*args, **kwargs)
(queryset, ):
model = queryset.model
field model._meta.fields:
field.is_relation field.many_to_one:
queryset = queryset.select_related(field.name)
queryset
wrapper
Django Web框架最佳实践
项目结构
- 应用划分: 合理划分应用职责
- 配置管理: 分环境配置管理
- 静态文件: 统一静态文件处理
- 模板组织: 优化模板继承结构
- 测试覆盖: 完善的测试体系
ORM优化
- 查询优化: 使用select_related和prefetch_related
- 索引优化: 添加适当的数据库索引
- 批量操作: 使用批量创建和更新
- 缓存策略: 合理使用查询缓存
- 连接池: 配置数据库连接池
安全配置
- DEBUG设置: 生产环境关闭DEBUG
- SECRET_KEY: 使用环境变量管理
- CSRF保护: 启用CSRF中间件
- SQL注入: 使用ORM防止注入
- XSS防护: 模板自动转义
性能优化
- 缓存配置: 配置Redis或Memcached
- 静态文件: 使用CDN服务静态文件
- 数据库优化: 优化数据库配置
- 中间件优化: 精简中间件配置
- 监控告警: 实时性能监控
相关技能
- python-development - Python开发
- database-optimization - 数据库优化
- web-security - Web安全
- restful-api-design - RESTful API设计