一、实现按周期统计是一个很常用的功能，在使用python的DRF中也可以实现该功能。
按照周进行汇总
SELECT DATE_FORMAT(exams_dat,"%Y-%u") as weeks_month, profession_id, Sum('score_total')
FROM tb_production_statistics
GROUP BY profession_id,weeks_month
按照月进行汇总
SELECT DATE_FORMAT(exams_dat,"%Y-%m") as weeks_month, profession_id, Sum('score_total')
FROM tb_production_statistics
GROUP BY profession_id,weeks_month
SQL表示形式
按周进行汇总
from django.db.models.functions import TruncWeek,TruncMonth
Article.objects
.annotate(week_mont=TruncWeek('exams_date')) # Truncate to month and add to select list
.values('profession_id') # Group By profession_id
.annotate(score_total=Sum('score_total')) # Select the sum of the grouping
.values('week_mont', 'profession_id') # (might be redundant, haven't tested) select month and count
按月进行汇总
from django.db.models.functions import TruncWeek,TruncMonth
Article.objects
.annotate(week_mont=TruncMonth('exams_date')) # Truncate to month and add to select list
.values('profession_id') # Group By profession_id
.annotate(score_total=Sum('score_total')) # Select the sum of the grouping
.values('week_mont', 'profession_id') # (might be redundant, haven't tested) select month and count
Django表示形式
二、GroupBy之后部分防止数据信息，可以把某个字段拼接起来
SELECT `tb_score_summary`.`profession_id`, MIN(`tb_score_summary`.`exams_date`) AS `exams_date`, SUM(`tb_score_summary`.`score_total`) AS `score_total`, GROUP_CONCAT(`tb_score_summary`.`total_concat` SEPARATOR ";") AS `total_concat` FROM `tb_score_summary` GROUP BY `tb_score_summary`.`profession_id`SQL表示形式
在Django中需要自定义一个GroupConcat类
class GroupConcat(Aggregate):
function = 'GROUP_CONCAT'
template = '%(function)s(%(distinct)s%(expressions)s%(ordering)s%(separator)s)'
def __init__(self, expression, distinct=False, ordering=None, separator=';', **extra):
super(GroupConcat, self).__init__(
expression,
distinct='DISTINCT ' if distinct else '',
ordering=' ORDER BY %s' % ordering if ordering is not None else '',
separator=' SEPARATOR "%s"' % separator,
output_field=CharField(),
**extra)
GroupConcat
按周进行汇总,并且把total_concat拼接
from django.db.models.functions import TruncWeek,TruncMonth
Article.objects
.annotate(week_mont=TruncWeek('exams_date')) # Truncate to month and add to select list
.values('profession_id') # Group By profession_id
.annotate(score_total=Sum('score_total'), total_concat=GroupConcat('total_concat', separator=';')
) # Select the sum of the grouping
.values('week_mont', 'profession_id') # (might be redundant, haven't tested) select month and count
按月进行汇总,并且把total_concat拼接
from django.db.models.functions import TruncWeek,TruncMonth
Article.objects
.annotate(week_mont=TruncMonth('exams_date')) # Truncate to month and add to select list
.values('profession_id') # Group By profession_id
.annotate(score_total=Sum('score_total'), total_concat=GroupConcat('total_concat', separator=';')
) # Select the sum of the grouping
.values('week_mont', 'profession_id') # (might be redundant, haven't tested) select month and count
Django表示形式
三、实现按月统计不同专业的总成绩，并把每个学生的考试总成绩拼接再一起
from django.urls import path, re_path
from api_service.report.views import ScoreListView, ScoreDetailView, ScoreSummaryListView
urlpatterns = [
path('score_summary/list', ScoreSummaryListView.as_view()), # 成绩汇总列表
]
url.py
from django.db import models
class ScoreModels(models.Model):
gender = models.SmallIntegerField(verbose_name='性别')
profession_id = models.IntegerField(verbose_name='专业id')
profession_name = models.CharField(max_length=255, verbose_name='专业名称')
course_id = models.IntegerField(verbose_name='课程id')
course_name = models.CharField(max_length=255, verbose_name='课程名称')
student_id = models.IntegerField(verbose_name='学生ID')
student_name = models.CharField(max_length=255, verbose_name='学生名称')
exams_version = models.IntegerField(verbose_name='考试版本')
exams_date = models.DateField(verbose_name='考试日期')
score_total = models.BigIntegerField(verbose_name='考试成绩')
add_time = models.DateTimeField(verbose_name='创建时间')
modify_time = models.DateTimeField(verbose_name='修改时间')
class Meta:
managed = True
db_table = 'tb_score'
verbose_name = '考试成绩表'
class ScoreSummaryModels(models.Model):
profession_id = models.IntegerField(verbose_name='专业id')
exams_version = models.IntegerField(verbose_name='考试版本')
exams_date = models.DateField(verbose_name='考试日期')
score_total = models.BigIntegerField(verbose_name='考试成绩')
total_concat = models.TextField(verbose_name='以{标注员：总成绩}组成的json字符串')
add_time = models.DateTimeField(verbose_name='创建时间')
modify_time = models.DateTimeField(verbose_name='修改时间')
class Meta:
managed = True
db_table = 'tb_score_summary'
verbose_name = '考试成绩汇总表'
model.py
import django_filters
import logging
from django_filters import rest_framework as filter_set
from api_service.report.models import ScoreModels, ScoreSummaryModels
from api_service.report.public_methods import date_range_filter
logger = logging.getLogger("api_service")
class ScoreFilter(filter_set.FilterSet):
profession_id = filter_set.CharFilter(lookup_expr='exact')
gender = django_filters.BaseInFilter(lookup_expr='in')
exams_date = filter_set.CharFilter(field_name='exams_date', method=date_range_filter)
class Meta:
models = ScoreModels
fields = ('profession_id', 'gender', 'exams_date')
class ScoreSummaryFilter(filter_set.FilterSet):
profession_id = filter_set.CharFilter(lookup_expr='exact')
exams_date = filter_set.CharFilter(field_name='exams_date', method=date_range_filter)
class Meta:
models = ScoreSummaryModels
fields = ('profession_id', 'exams_date')
filters.py
class ScoresSummarySerializer(serializers.ModelSerializer):
class Meta:
model = ScoreSummaryModels
fields = ('profession_id', 'exams_date', 'score_total', 'total_concat',)
def to_representation(self, instance):
data = super().to_representation(instance)
period_type_no = self.context['request'].query_params.get('period_type_no')
if period_type_no == str(PeriodType.PERIOD_TYPE_CODE_MONTH):
cur_year = datetime.strptime(data['exams_date'], '%Y-%m-%d').year
cur_month = datetime.strptime(data['exams_date'], '%Y-%m-%d').month
data['exams_date'] = "%d年%d月" % (cur_year, cur_month)
return data
serializers.py
class ScoreSummaryListView(ListAPIView):
queryset = ScoreSummaryModels.objects.all()
serializer_class = ScoresSummarySerializer
filter_backends = (filters.OrderingFilter, DjangoFilterBackend,)
filter_class = ScoreSummaryFilter
pagination_class = StandardBasePagination
source_summary_list_values = ('profession_id', 'exams_date', 'score_total', 'total_concat',)
def filter_queryset(self, queryset):
for backend in list(self.filter_backends):
queryset = backend().filter_queryset(self.request, queryset, self)
if queryset.exists():
params = self.request.query_params
period_type_no = params.get('period_type_no')
if period_type_no == str(PeriodType.PERIOD_TYPE_CODE_WEEK):
period_type_str = "TruncWeek('exams_date')"
elif period_type_no == str(PeriodType.PERIOD_TYPE_CODE_MONTH):
period_type_str = "TruncMonth('exams_date')"
elif period_type_no == str(PeriodType.PERIOD_TYPE_CODE_DAY):
period_type_str = "TruncDay('exams_date')"
queryset = self.statistic_by_period(queryset, period_type_str)
return queryset
def statistic_by_period(self, queryset, period_type):
queryset = queryset.annotate(week_month=eval(period_type)) \
.values("profession_id", ) \
.annotate(
exams_date=Min('exams_date'),
score_total=Sum('score_total'),
total_concat=GroupConcat('total_concat', separator=';')
).values(*self.source_summary_list_values)
return queryset
view.py
class GroupConcat(Aggregate):
function = 'GROUP_CONCAT'
template = '%(function)s(%(distinct)s%(expressions)s%(ordering)s%(separator)s)'
def __init__(self, expression, distinct=False, ordering=None, separator=';', **extra):
super(GroupConcat, self).__init__(
expression,
distinct='DISTINCT ' if distinct else '',
ordering=' ORDER BY %s' % ordering if ordering is not None else '',
separator=' SEPARATOR "%s"' % separator,
output_field=CharField(),
**extra)
GroupConcat
标签：GroupConcat,models,profession,Django,76,score,exams,total,id From： https://www.cnblogs.com/YK2012/p/15968931.html