Python django.db.models 模块，ExpressionWrapper() 实例源码

def get_queryset(self):
        empty_str = ExpressionWrapper(V(''), output_field=CharField())
        future_meeting = models.MeetingHistory.objects.latest('date')

        return models.PresentHistory.objects.values(
            date=F('meeting__date'),
            presentation_type=F('present_type'),
            presenter_name=F('presenter__name'),
            present_content=F('content'),
        ).exclude(meeting__date=future_meeting.date).order_by().union(
            models.MeetingSkip.objects.all().values(
                'date',
                presentation_type=Concat(V('Postponed: '), 'reason'),
                presenter_name=empty_str,
                present_content=empty_str,
            ).filter(date__lte=date.today()).order_by()
        ).order_by('-date')

def get_user_queryset(self):

        qs = super(UserViewSet, self).get_user_queryset()

        if self.action == 'retrieve':

            if self.request.user.is_authenticated():
                qs = qs.annotate(
                    followed=models.ExpressionWrapper(
                        models.Count(
                            models.Subquery(
                                User.following.through.objects.filter(
                                    to_user_id=self.request.user.id,
                                    from_user_id=models.OuterRef('id')
                                ).values('id')
                            )
                        ),
                        output_field=models.BooleanField()
                    )
                )

        return qs

def get_queryset(self):
        user = self.get_user_object()

        try:
            rel_field = getattr(user, self.allowed_actions[self.action])
        except KeyError:
            raise NotFound

        if self.request.user.is_authenticated():
            rel_field = rel_field.annotate(
                followed=models.ExpressionWrapper(
                    models.Count(
                        models.Subquery(
                            User.following.through.objects.filter(
                                to_user_id=self.request.user.id,
                                from_user_id=models.OuterRef('id')
                            ).values('id')
                        )
                    ),
                    output_field=models.BooleanField()
                )
            )

        return rel_field.order_by('id')

def with_bounds(self):
        return self.annotate(
            start=Min('statuses__creation_date'),
            end=Max('statuses__creation_date')).annotate(
                statuses_delta=ExpressionWrapper(
                    F('end') - F('start'),
                    output_field=models.DurationField()
                ))

def duration_expr():
        return ExpressionWrapper(
            Cast(F('max_frame') - F('min_frame'), models.FloatField()) / F('video__fps'),
            models.FloatField())

def height_expr():
        return ExpressionWrapper(F('bbox_y2') - F('bbox_y1'), models.FloatField())

def stats(self):
        return self.order_by('category').values('category')\
            .annotate(count=models.Count('id'))\
            .annotate(
                unread=models.ExpressionWrapper(
                    models.F('count') - models.Sum('has_read'),
                    output_field=models.IntegerField()))

def users_stats(self, users):

        if isinstance(users, models.QuerySet):
            users = models.Subquery(users)

        qs = self.order_by('user').values('user')\
            .filter(user__in=users)\
            .annotate(count=models.Count('id'))\
            .annotate(
                unread=models.ExpressionWrapper(
                    models.F('count') - models.Sum('has_read'),
                    output_field=models.IntegerField()
                )
        )
        return qs

def get_avg_duration(query_set):
    duration = ExpressionWrapper(F('updated_datetime') - F('requested_datetime'), output_field=fields.DurationField())
    duration_list = query_set.annotate(duration=duration).values_list("duration", flat=True)
    if not duration_list:
        return datetime.timedelta(0)
    return sum(duration_list, datetime.timedelta(0)) / len(duration_list)


# Returns median duration of closed issues (updated_datetime - requested_datetime)
# from given category. Returns a tuple (days, hours)

def get_median_duration(query_set):
    duration = ExpressionWrapper(F('updated_datetime') - F('requested_datetime'), output_field=fields.DurationField())
    duration_list = sorted(query_set.annotate(duration=duration).values_list("duration", flat=True))
    if not duration_list:
        return datetime.timedelta(0)
    return duration_list[(len(duration_list) - 1) // 2]


# Concerts timedelta into millisoconds

def get_context_data(self, **kwargs):
        context = super(PlayerList, self).get_context_data(**kwargs)
        players = context['player_list']

        players = players.annotate(
            match_count=Count('matchplayer'),
            wins=Count(Case(
                When(
                    matchplayer__team=F('matchplayer__match__winner'), then=1)
                )
            ),
            winrate=ExpressionWrapper(
                F('wins') * Decimal('100') / F('match_count'),
                output_field=FloatField()
            )
        )

        if not players:
            # no games this season yet, nothing to calc
            return context

        max_vals = players.aggregate(Max('dota_mmr'), Max('score'), Max('ladder_mmr'))
        score_max = max_vals['score__max']
        dota_mmr_max = max_vals['dota_mmr__max']
        ladder_mmr_max = max_vals['ladder_mmr__max']

        matches_max = max(player.match_count for player in players)

        for player in players:
            player.score_percent = float(player.score) / score_max * 100
            player.dota_mmr_percent = float(player.dota_mmr) / dota_mmr_max * 100
            player.ladder_mmr_percent = float(player.ladder_mmr) / ladder_mmr_max * 100
            player.matches_percent = float(player.match_count) / matches_max * 100

        context.update({
            'player_list': players,
        })

        return context


# TODO: inherit PlayersBest and PlayersSuccessful from PlayerList

def overview(request, event_url_name):
    event = get_object_or_404(Event, url_name=event_url_name)

    # permission
    if not event.is_admin(request.user):
        return nopermission(request)

    num_helpers = event.helper_set.count()

    num_coordinators = 0
    timeline = {}
    for helper in event.helper_set.all():
        if helper.is_coordinator:
            num_coordinators += 1
        else:
            day = helper.timestamp.strftime('%Y-%m-%d')
            if day in timeline:
                timeline[day] += 1
            else:
                timeline[day] = 1

    num_vegetarians = event.helper_set.filter(vegetarian=True).count()

    num_shift_slots = Shift.objects.filter(job__event=event).aggregate(
        Sum('number'))['number__sum']

    empty_slots_expr = ExpressionWrapper(F('number') - F('num_helpers'),
                                         output_field=fields.IntegerField())
    num_empty_shift_slots = Shift.objects.filter(job__event=event) \
                            .annotate(num_helpers=Count('helper')) \
                            .annotate(empty_slots=empty_slots_expr) \
                            .aggregate(Sum('empty_slots'))['empty_slots__sum']

    total_duration = ExpressionWrapper((F('end') - F('begin')) * F('number'),
                                       output_field=fields.DurationField())
    hours_total = Shift.objects.filter(job__event=event) \
                       .annotate(duration=total_duration) \
                       .aggregate(Sum('duration'))['duration__sum']

    # sum up timeline
    timeline = OrderedDict(sorted(timeline.items()))
    timeline_sum = OrderedDict()
    tmp = 0
    for day in timeline:
        tmp += timeline[day]
        timeline_sum[day] = tmp

    # render
    context = {'event': event,
               'num_helpers': num_helpers,
               'num_coordinators': num_coordinators,
               'num_vegetarians': num_vegetarians,
               'num_shift_slots': num_shift_slots,
               'num_empty_shift_slots': num_empty_shift_slots,
               'hours_total': hours_total,
               'timeline': timeline_sum}
    return render(request, 'statistic/overview.html', context)