优化后的 api_device_water_usage_statistics 接口
概述
为了提升 api_device_water_usage_statistics 接口的性能、可维护性和可扩展性，进行了全面的优化。优化内容涵盖了代码结构调整、数据库查询优化、缓存策略实施、错误处理增强以及性能监控等方面。以下将详细介绍各项优化策略及其实现方式。
1. 代码结构优化
优化代码结构的主要目的是提高代码的可读性和可维护性，减少冗余代码。通过将复杂的逻辑拆分为独立的函数，实现模块化管理。
1.1 函数模块化
将位置过滤、统计处理等逻辑独立成函数，避免主函数过于庞大和复杂。
位置过滤函数
根据传入的位置信息筛选设备，避免重复代码。
def filter_devices_by_location(location_l1_id, location_l2_id, location_l3_id):
"""
根据位置信息筛选设备
"""
location = None
if location_l3_id:
location = Location.objects.filter(id=location_l3_id).first()
elif location_l2_id:
location = Location.objects.filter(id=location_l2_id).first()
elif location_l1_id:
location = Location.objects.filter(id=location_l1_id).first()
if (location_l3_id or location_l2_id or location_l1_id) and not location:
return None, error_response(message='指定位置不存在')
device_query = DeviceInfoModel.objects.all()
if location:
if location_l3_id:
device_query = device_query.filter(location_id=location_l3_id)
else:
leaf_locations = location.get_leafnodes()
device_query = device_query.filter(location__in=leaf_locations)
return device_query, None
统计处理函数
根据统计类型（日/月）处理用水量数据。
def handle_statistics(device_query, search_time, statistics_type):
"""
根据统计类型处理用水量数据
"""
if statistics_type == '日':
return handle_daily_statistics(device_query, search_time)
elif statistics_type == '月':
return handle_monthly_statistics(device_query, search_time)
else:
return error_response('统计类型必须为 "日" 或 "月"')
1.2 主函数优化
@custom_require_http_methods(['GET'])
@json_request_handler
@token_validation_handler
def api_device_water_usage_statistics(request):
"""
用水量统计接口
"""
# 获取请求参数
search_time = request.json_data.get('search_time') # 查询时间，格式为 "2024" 或 "2024-03"
if not search_time:
return error_response('请指定查询时间')
statistics_type = request.json_data.get('statistics_type', '日') # 统计类型，默认为 "日"
if statistics_type not in ['日', '月']:
return error_response('统计类型必须为 "日" 或 "月"')
# 获取位置信息
location_l1_id = request.json_data.get('location_l1_id') # 一级位置信息
location_l2_id = request.json_data.get('location_l2_id') # 二级位置信息
location_l3_id = request.json_data.get('location_l3_id') # 三级位置信息
# 根据位置筛选设备
device_query, error = filter_devices_by_location(location_l1_id, location_l2_id, location_l3_id)
if error:
return error
if not device_query.exists():
return error_response('未查询到设备数据')
# 处理统计
return handle_statistics(device_query, search_time, statistics_type)
2. 数据库查询优化
优化数据库查询以减少查询次数和提高查询效率，主要包括使用 Django ORM 的高级功能以及避免不必要的查询。
2.1 使用 select_related 和 prefetch_related
预加载相关表数据，减少数据库查询次数。
def filter_devices_by_location(location_l1_id, location_l2_id, location_l3_id):
"""
根据位置信息筛选设备，并预加载相关的位置信息
"""
location = None
if location_l3_id:
location = Location.objects.filter(id=location_l3_id).first()
elif location_l2_id:
location = Location.objects.filter(id=location_l2_id).first()
elif location_l1_id:
location = Location.objects.filter(id=location_l1_id).first()
if (location_l3_id or location_l2_id or location_l1_id) and not location:
return None, error_response(message='指定位置不存在')
device_query = DeviceInfoModel.objects.select_related('location')
if location:
if location_l3_id:
device_query = device_query.filter(location_id=location_l3_id)
else:
leaf_locations = location.get_leafnodes()
device_query = device_query.filter(location__in=leaf_locations)
return device_query, None
2.2 使用 annotate 和 aggregate
直接在数据库中进行统计计算，减少数据传输和处理开销。
from django.db.models import Sum
from django.db.models.functions import TruncMonth
def handle_daily_statistics(device_query, search_time):
"""
处理按日统计的逻辑
"""
try:
search_date = datetime.datetime.strptime(search_time, "%Y-%m")
except ValueError as e:
return error_response(f'日期格式错误：{e}')
first_day = datetime.date(search_date.year, search_date.month, 1)
last_day = (datetime.date(search_date.year, search_date.month + 1, 1) - timedelta(days=1)
if search_date.month < 12
else datetime.date(search_date.year + 1, 1, 1) - timedelta(days=1))
# 查询设备用水量
all_daily_data = (
DeviceDailyUsageModel.objects.filter(statistics_date__range=(first_day, last_day))
.exclude(water_use__gt=100000000)
.values("license_no", "statistics_date")
.annotate(daily_water_use=Sum("water_use"))
)
# 按 license_no 分组
data_by_license_no = defaultdict(dict)
for entry in all_daily_data:
data_by_license_no[entry["license_no"]][entry["statistics_date"]] = entry["daily_water_use"]
# 获取设备信息
device_info_dict = fetch_device_info(device_query)
# 初始化结果
specific = []
for device in device_query:
daily_usage = data_by_license_no.get(device.license_no, {})
daily_water_usage = {}
total_water_use = 0
for current_day in (first_day + timedelta(days=i) for i in range((last_day - first_day).days + 1)):
water_use = daily_usage.get(current_day, decimal.Decimal('0.000')) # 无数据时填充 0
daily_water_usage[current_day] = water_use
total_water_use += water_use
# 生成设备统计数据
device_statistics_data = {
"water_usage": list(daily_water_usage.values()),
"total": total_water_use,
}
device_statistics_data.update(device_info_dict.get(device.license_no, {}))
specific.append(device_statistics_data)
return success_response(data=specific)
2.3 避免原生 SQL 查询
使用 Django ORM 替代原生 SQL 查询，提升安全性和可维护性。
def fetch_device_info(device_query):
"""
获取设备的详细信息
"""
license_nos = device_query.values_list('license_no', flat=True)
device_info = DeviceInfoModel.objects.filter(license_no__in=license_nos).select_related('location')
device_info_dict = {}
for device in device_info:
device_info_dict[device.license_no] = {
"location_1_name": device.location.parent.parent.name if device.location and device.location.parent and device.location.parent.parent else "",
"location_2_name": device.location.parent.name if device.location and device.location.parent else "",
"location_3_name": device.location.name if device.location else "",
"owner_name": device.owner_name,
"well_number": device.well_number,
"water_permit_e_certificate_code": device.water_permit_e_certificate_code,
"device_id": device.device_id
}
return device_info_dict
3. 缓存策略
使用缓存来存储频繁查询的数据，减少数据库访问次数，提升接口响应速度。
3.1 缓存设备信息
from django.core.cache import cache
def fetch_device_info(device_query):
"""
获取设备的详细信息，使用缓存优化
"""
license_nos = list(device_query.values_list('license_no', flat=True))
cache_key = f"device_info_{'_'.join(license_nos)}"
device_info_dict = cache.get(cache_key)
if not device_info_dict:
device_info = DeviceInfoModel.objects.filter(license_no__in=license_nos).select_related('location')
device_info_dict = {}
for device in device_info:
device_info_dict[device.license_no] = {
"location_1_name": device.location.parent.parent.name if device.location and device.location.parent and device.location.parent.parent else "",
"location_2_name": device.location.parent.name if device.location and device.location.parent else "",
"location_3_name": device.location.name if device.location else "",
"owner_name": device.owner_name,
"well_number": device.well_number,
"water_permit_e_certificate_code": device.water_permit_e_certificate_code,
"device_id": device.device_id
}
cache.set(cache_key, device_info_dict, timeout=60 * 60) # 缓存1小时
return device_info_dict
4. 错误处理与日志记录
增强错误处理机制，确保接口在异常情况下能够稳定运行，并记录必要的日志以便后续调试。
4.1 增强异常处理
import logging
logger = logging.getLogger(__name__)
def handle_daily_statistics(device_query, search_time):
"""
处理按日统计的逻辑
"""
try:
search_date = datetime.datetime.strptime(search_time, "%Y-%m")
except ValueError as e:
logger.error(f"日期格式错误：{search_time}, 错误信息：{e}")
return error_response(f'日期格式错误：%Y-%m {e}')
first_day = datetime.date(search_date.year, search_date.month, 1)
last_day = (datetime.date(search_date.year, search_date.month + 1, 1) - timedelta(days=1)
if search_date.month < 12
else datetime.date(search_date.year + 1, 1, 1) - timedelta(days=1))
try:
# 查询设备用水量
all_daily_data = (
DeviceDailyUsageModel.objects.filter(statistics_date__range=(first_day, last_day))
.exclude(water_use__gt=100000000)
.values("license_no", "statistics_date")
.annotate(daily_water_use=Sum("water_use"))
)
except Exception as e:
logger.error(f"查询设备用水量失败：{e}")
return error_response('查询失败，请稍后重试')
# 按 license_no 分组
data_by_license_no = defaultdict(dict)
for entry in all_daily_data:
data_by_license_no[entry["license_no"]][entry["statistics_date"]] = entry["daily_water_use"]
# 获取设备信息
device_info_dict = fetch_device_info(device_query)
# 初始化结果
specific = []
for device in device_query:
daily_usage = data_by_license_no.get(device.license_no, {})
daily_water_usage = {}
total_water_use = 0
for current_day in (first_day + timedelta(days=i) for i in range((last_day - first_day).days + 1)):
water_use = daily_usage.get(current_day, decimal.Decimal('0.000')) # 无数据时填充 0
daily_water_usage[current_day] = water_use
total_water_use += water_use
# 生成设备统计数据
device_statistics_data = {
"water_usage": list(daily_water_usage.values()),
"total": total_water_use,
}
try:
device_statistics_data.update(device_info_dict.get(device.license_no, {}))
except Exception as e:
logger.error(f"合并设备信息失败：{e}")
logger.debug(f"设备统计数据：{device_statistics_data}")
specific.append(device_statistics_data)
return success_response(data=specific)
5. 性能测试与监控
在优化过程中，需要通过性能测试和监控工具来验证优化效果，并持续监控接口的运行状况。
5.1 使用 Django Debug Toolbar
分析 SQL 查询、请求/响应时间等，帮助发现性能瓶颈。
- 安装：
pip install django-debug-toolbar - 配置步骤请参考 Django Debug Toolbar 官方文档
5.2 使用 New Relic 监控
实时监控应用性能，检测潜在问题。
- 安装：
pip install newrelic - 配置步骤请参考 New Relic 官方文档
6. 优化后的完整代码示例
以下是整合以上优化策略后的完整接口代码，包含详细的中文注释。
from django.db.models import Sum
from django.db.models.functions import TruncMonth
from collections import defaultdict
import datetime
import decimal
from django.core.cache import cache
import logging
logger = logging.getLogger(__name__)
@custom_require_http_methods(['GET'])
@json_request_handler
@token_validation_handler
def api_device_water_usage_statistics(request):
"""
用水量统计接口
"""
# 获取请求参数
search_time = request.json_data.get('search_time') # 查询时间，格式为 "2024" 或 "2024-03"
if not search_time:
return error_response('请指定查询时间')
statistics_type = request.json_data.get('statistics_type', '日') # 统计类型，默认为 "日"
if statistics_type not in ['日', '月']:
return error_response('统计类型必须为 "日" 或 "月"')
# 获取位置信息
location_l1_id = request.json_data.get('location_l1_id') # 一级位置信息
location_l2_id = request.json_data.get('location_l2_id') # 二级位置信息
location_l3_id = request.json_data.get('location_l3_id') # 三级位置信息
# 根据位置筛选设备
device_query, error = filter_devices_by_location(location_l1_id, location_l2_id, location_l3_id)
if error:
return error
if not device_query.exists():
return error_response('未查询到设备数据')
# 处理统计
return handle_statistics(device_query, search_time, statistics_type)
def filter_devices_by_location(location_l1_id, location_l2_id, location_l3_id):
"""
根据位置信息筛选设备，并预加载相关的位置信息
"""
location = None
if location_l3_id:
location = Location.objects.filter(id=location_l3_id).first()
elif location_l2_id:
location = Location.objects.filter(id=location_l2_id).first()
elif location_l1_id:
location = Location.objects.filter(id=location_l1_id).first()
if (location_l3_id or location_l2_id or location_l1_id) and not location:
return None, error_response(message='指定位置不存在')
device_query = DeviceInfoModel.objects.select_related('location')
if location:
if location_l3_id:
device_query = device_query.filter(location_id=location_l3_id)
else:
leaf_locations = location.get_leafnodes()
device_query = device_query.filter(location__in=leaf_locations)
return device_query, None
def handle_statistics(device_query, search_time, statistics_type):
"""
根据统计类型处理用水量数据
"""
if statistics_type == '日':
return handle_daily_statistics(device_query, search_time)
elif statistics_type == '月':
return handle_monthly_statistics(device_query, search_time)
else:
return error_response('统计类型必须为 "日" 或 "月"')
def handle_daily_statistics(device_query, search_time):
"""
处理按日统计的逻辑
"""
try:
search_date = datetime.datetime.strptime(search_time, "%Y-%m")
except ValueError as e:
logger.error(f"日期格式错误：{search_time}, 错误信息：{e}")
return error_response(f'日期格式错误：%Y-%m {e}')
first_day = datetime.date(search_date.year, search_date.month, 1)
last_day = (datetime.date(search_date.year, search_date.month + 1, 1) - timedelta(days=1)
if search_date.month < 12
else datetime.date(search_date.year + 1, 1, 1) - timedelta(days=1))
try:
# 查询设备用水量
all_daily_data = (
DeviceDailyUsageModel.objects.filter(statistics_date__range=(first_day, last_day))
.exclude(water_use__gt=100000000)
.values("license_no", "statistics_date")
.annotate(daily_water_use=Sum("water_use"))
)
except Exception as e:
logger.error(f"查询设备用水量失败：{e}")
return error_response('查询失败，请稍后重试')
# 按 license_no 分组
data_by_license_no = defaultdict(dict)
for entry in all_daily_data:
data_by_license_no[entry["license_no"]][entry["statistics_date"]] = entry["daily_water_use"]
# 获取设备信息
device_info_dict = fetch_device_info(device_query)
# 初始化结果
specific = []
for device in device_query:
daily_usage = data_by_license_no.get(device.license_no, {})
daily_water_usage = {}
total_water_use = 0
for current_day in (first_day + timedelta(days=i) for i in range((last_day - first_day).days + 1)):
water_use = daily_usage.get(current_day, decimal.Decimal('0.000')) # 无数据时填充 0
daily_water_usage[current_day] = water_use
total_water_use += water_use
# 生成设备统计数据
device_statistics_data = {
"water_usage": list(daily_water_usage.values()),
"total": total_water_use,
}
try:
device_statistics_data.update(device_info_dict.get(device.license_no, {}))
except Exception as e:
logger.error(f"合并设备信息失败：{e}")
logger.debug(f"设备统计数据：{device_statistics_data}")
specific.append(device_statistics_data)
return success_response(data=specific)
def handle_monthly_statistics(device_query, search_time):
"""
处理按月统计的逻辑
"""
try:
search_date = datetime.datetime.strptime(search_time, "%Y")
except ValueError as e:
logger.error(f"日期格式错误：{search_time}, 错误信息：{e}")
return error_response(f'日期格式错误：%Y {e}')
first_day = datetime.date(search_date.year, 1, 1)
last_day = datetime.date(search_date.year, 12, 31)
try:
# 查询设备用水量
all_monthly_data = (
DeviceDailyUsageModel.objects.filter(statistics_date__range=(first_day, last_day))
.exclude(water_use__gt=100000000)
.annotate(month=TruncMonth("statistics_date"))
.values("license_no", "month")
.annotate(monthly_water_use=Sum("water_use"))
)
except Exception as e:
logger.error(f"查询设备用水量失败：{e}")
return error_response('查询失败，请稍后重试')
# 按 license_no 分组
data_by_license_no = defaultdict(dict)
for entry in all_monthly_data:
data_by_license_no[entry["license_no"]][entry["month"]] = entry["monthly_water_use"]
# 获取设备信息
device_info_dict = fetch_device_info(device_query)
# 初始化结果
specific = []
for device in device_query:
monthly_usage = data_by_license_no.get(device.license_no, {})
monthly_water_usage = {}
total_water_use = 0
for month in range(1, 13):
current_month = datetime.date(search_date.year, month, 1)
water_use = monthly_usage.get(current_month, decimal.Decimal('0.000')) # 无数据时填充 0
monthly_water_usage[current_month] = water_use
total_water_use += water_use
# 生成设备统计数据
device_statistics_data = {
"water_usage": list(monthly_water_usage.values()),
"total": total_water_use,
}
try:
device_statistics_data.update(device_info_dict.get(device.license_no, {}))
except Exception as e:
logger.error(f"合并设备信息失败：{e}")
logger.debug(f"设备统计数据：{device_statistics_data}")
specific.append(device_statistics_data)
return success_response(data=specific)
def fetch_device_info(device_query):
"""
获取设备的详细信息，使用缓存优化
"""
license_nos = list(device_query.values_list('license_no', flat=True))
cache_key = f"device_info_{'_'.join(license_nos)}"
device_info_dict = cache.get(cache_key)
if not device_info_dict:
try:
device_info = DeviceInfoModel.objects.filter(license_no__in=license_nos).select_related('location')
device_info_dict = {}
for device in device_info:
device_info_dict[device.license_no] = {
"location_1_name": device.location.parent.parent.name if device.location and device.location.parent and device.location.parent.parent else "",
"location_2_name": device.location.parent.name if device.location and device.location.parent else "",
"location_3_name": device.location.name if device.location else "",
"owner_name": device.owner_name,
"well_number": device.well_number,
"water_permit_e_certificate_code": device.water_permit_e_certificate_code,
"device_id": device.device_id
}
cache.set(cache_key, device_info_dict, timeout=60 * 60) # 缓存1小时
except Exception as e:
logger.error(f"获取设备信息失败：{e}")
return {}
return device_info_dict
7. 总结
通过上述优化策略，api_device_water_usage_statistics 接口在性能、可维护性和可扩展性方面获得了显著提升。主要优化点包括：
-
**代码结构优化**：模块化函数，提升代码可读性和可维护性。
-
**数据库查询优化**：使用 Django ORM 高级功能减少查询次数，提高查询效率。
-
**缓存策略实施**：缓存频繁查询的数据，降低数据库负载，加快响应速度。
-
**错误处理与日志记录**：增强异常处理，记录详细日志，便于问题排查。
-
**性能测试与监控**：使用工具进行性能分析和实时监控，持续优化接口性能。
这些优化措施不仅提升了接口的当前性能，还为未来的扩展和维护奠定了坚实的基础。建议在实际应用中持续监控接口表现，根据实际需求进一步调整优化策略。
参考资料
- 《两万字的性能优化指南！39个策略提升接口性能！》
- 《5个接口性能提升的通用技巧》
- 《接口性能优化的11个小技巧》
Last updated January 6, 2025