我们从Python开源项目中，提取了以下26个代码示例，用于说明如何使用django.db.models.expressions.F。
def datetimes(self, field_name, kind, order='ASC', tzinfo=None): """ Returns a list of datetime objects representing all available datetimes for the given field_name, scoped to 'kind'. """ assert kind in ("year", "month", "day", "hour", "minute", "second"), \ "'kind' must be one of 'year', 'month', 'day', 'hour', 'minute' or 'second'." assert order in ('ASC', 'DESC'), \ "'order' must be either 'ASC' or 'DESC'." if settings.USE_TZ: if tzinfo is None: tzinfo = timezone.get_current_timezone() else: tzinfo = None return self.annotate( datetimefield=DateTime(field_name, kind, tzinfo), plain_field=F(field_name) ).values_list( 'datetimefield', flat=True ).distinct().filter(plain_field__isnull=False).order_by(('-' if order == 'DESC' else '') + 'datetimefield')
def datetimes(self, field_name, kind, order='ASC', tzinfo=None): """ Returns a list of datetime objects representing all available datetimes for the given field_name, scoped to 'kind'. """ assert kind in ("year", "month", "day", "hour", "minute", "second"), \ "'kind' must be one of 'year', 'month', 'day', 'hour', 'minute' or 'second'." assert order in ('ASC', 'DESC'), \ "'order' must be either 'ASC' or 'DESC'." if settings.USE_TZ: if tzinfo is None: tzinfo = timezone.get_current_timezone() else: tzinfo = None return self.annotate( datetimefield=Trunc(field_name, kind, output_field=DateTimeField(), tzinfo=tzinfo), plain_field=F(field_name) ).values_list( 'datetimefield', flat=True ).distinct().filter(plain_field__isnull=False).order_by(('-' if order == 'DESC' else '') + 'datetimefield')
def get_dates_only(self): """ Returns the queryset containing entries having the date field only. Date is a trunc of a published_parsed datetime field. """ # NOTE: order_by influences the distinct() results here date_times = RssNotification.objects.annotate( published_parsed_date=Trunc('published_parsed', 'day', output_field=DateField()), plain_field=F('published_parsed') ).values( 'published_parsed_date' ).distinct().filter(plain_field__isnull=False).order_by( '-published_parsed_date' ).annotate(dates_count=Count('published_parsed')) return date_times
def increment_completed_steps(self, steps=1): """ Increase the value of :py:attr:`completed_steps` by the given number and save, then check for cancellation. If cancellation of the task has been requested, a TaskCanceledException will be raised to abort execution. If any special cleanup is required, this exception should be caught and handled appropriately. This method should be called often enough to provide a useful indication of progress, but not so often as to cause undue burden on the database. """ UserTaskStatus.objects.filter(pk=self.id).update(completed_steps=F('completed_steps') + steps, modified=now()) self.refresh_from_db(fields={'completed_steps', 'modified', 'state'}) if self.parent: self.parent.increment_completed_steps(steps) # pylint: disable=no-member # Was a cancellation command recently sent? if self.state == self.CANCELED and not self.is_container: raise TaskCanceledException
def dates(self, field_name, kind, order='ASC'): """ Returns a list of date objects representing all available dates for the given field_name, scoped to 'kind'. """ assert kind in ("year", "month", "day"), \ "'kind' must be one of 'year', 'month' or 'day'." assert order in ('ASC', 'DESC'), \ "'order' must be either 'ASC' or 'DESC'." return self.annotate( datefield=Date(field_name, kind), plain_field=F(field_name) ).values_list( 'datefield', flat=True ).distinct().filter(plain_field__isnull=False).order_by(('-' if order == 'DESC' else '') + 'datefield')
def post(self, request, data): if not data.get('name') and not data.get('lat'): return HttpResponseServerError("Name or coords is required.") if data.get('name'): model = {'city': City, 'subregion': Subregion, 'region': Region}.get(data.get('model')) if not model: return HttpResponseServerError("Invalid model: {}".format(data.get('model'))) obj = get_object_or_404(model, pk=data['name']) else: # http://stackoverflow.com/a/35079313 pnt = Point(float(data.get('lon', 0)), float(data.get('lat', 0)), srid=4326) order_by_expression = CombinedExpression(F('location'), '<->', GeomValue(pnt)) try: obj = City.objects.order_by(order_by_expression)[0] except IndexError: return fields = ['subregion', 'region', 'region__country', 'country'] data = {getattr(obj, key).__class__.__name__: {x: getattr(getattr(obj, key), x) for x in ['id', 'name']} for key in fields if hasattr(obj, key)} if not data.get('name'): # Is geo coord search data['city'] = {'id': obj.pk, 'name': obj.name} if hasattr(obj, 'location'): order_by_expression = CombinedExpression(F('location'), '<->', GeomValue(obj.location)) try: data['postal_code'] = PostalCode.objects.order_by(order_by_expression)[0].code except IndexError: pass data['coords'] = obj.location.coords return data
def dates(self, field_name, kind, order='ASC'): """ Returns a list of date objects representing all available dates for the given field_name, scoped to 'kind'. """ assert kind in ("year", "month", "day"), \ "'kind' must be one of 'year', 'month' or 'day'." assert order in ('ASC', 'DESC'), \ "'order' must be either 'ASC' or 'DESC'." return self.annotate( datefield=Trunc(field_name, kind, output_field=DateField()), plain_field=F(field_name) ).values_list( 'datefield', flat=True ).distinct().filter(plain_field__isnull=False).order_by(('-' if order == 'DESC' else '') + 'datefield')
def get_queryset(self): return super().get_queryset() \ .annotate(ticket_count=models.Sum( models.Case( models.When( orderitem__order__refunded=True, then=0 ), models.When( orderitem__order__billed_total='', then=0 ), models.When( orderitem__order__billed_total__isnull=True, then=0 ), default=1, output_field=models.IntegerField(), ))) \ .annotate(sold_out=models.Case( models.When( ticket_count__lt=F('capacity'), then=models.Value(False), ), default=True, output_field=models.BooleanField(), ))
def ranked_search(self, terms): return list(self.objects .annotate(rank=SearchRank(F('search'), SearchQuery(terms, config='english'))) .order_by('-rank') .values_list('id', flat=True))
def test_headline(self): query = SearchQuery('hovercraft') result = list( self.objects .annotate(match=Headline(F('title'), query)) .values_list('match', flat=True) ) self.assertEqual(result, [ 'My <b>hovercraft</b> is full of eels.', 'Spam! Spam! Spam! Spam! Spam! Spam!', ])
def incr_and_update_instance(instance, **fields): if not fields: return updates = {} for f, v in fields.items(): setattr(instance, f, getattr(instance, f) + v) updates[f] = F(f) + v instance._default_manager.filter(pk=instance.pk).update(**updates)
def increment_total_steps(self, steps): """Increase the value of :py:attr:`total_steps` by the given number and save.""" # Assume that other processes may be making concurrent changes UserTaskStatus.objects.filter(pk=self.id).update(total_steps=F('total_steps') + steps, modified=now()) self.refresh_from_db(fields={'total_steps', 'modified'}) if self.parent: self.parent.increment_total_steps(steps) # pylint: disable=no-member
def IsNotNone(*fields, default=None): """Selects whichever field is not None, in the specified order. Arguments: fields: The fields to attempt to get a value from, in order. default: The value to return in case all values are None. Returns: A Case-When expression that tries each field and returns the specified default value when all of them are None. """ when_clauses = [ expressions.When( ~expressions.Q(**{field: None}), then=expressions.F(field) ) for field in reversed(fields) ] return expressions.Case( *when_clauses, default=expressions.Value(default), output_field=CharField() )
def resolve_search(self, args, context, info): term = args['q'] people = Person.objects.filter(name__icontains=term) houses = House.objects.annotate(name=Concat(F('number'), Value(' '), F('street__name'), output_field=models.TextField())).filter(name__icontains=term) streets = Street.objects.filter(name__icontains=term) return itertools.chain(people, houses, streets)
def handle(self, *args, **options): """ Send new message notifications """ # command to run: python manage.py tunga_send_customer_emails min_date = datetime.datetime.utcnow() - relativedelta(minutes=1) # 5 minute window to read new messages customer_channels = Channel.objects.filter( type=CHANNEL_TYPE_SUPPORT, created_by__isnull=True, content_type=ContentType.objects.get_for_model(Inquirer) ).annotate(new_messages=Sum( Case( When( ~Q(action_targets__actor_content_type=F('content_type')) & Q(action_targets__gt=F('last_read')) & Q(action_targets__timestamp__lte=min_date) & Q(action_targets__verb__in=[verbs.SEND, verbs.UPLOAD]), then=1 ), default=0, output_field=IntegerField() ) ), latest_message=Max('action_targets__id')).filter(new_messages__gt=0) for channel in customer_channels: customer = channel.content_object if customer.email: activities = Action.objects.filter( channels=channel, id__gt=channel.last_read, verb__in=[verbs.SEND] ).order_by('id') messages = [activity.action_object for activity in activities] if messages: to = [customer.email] subject = "[Tunga Support] Help" ctx = { 'customer': customer, 'count': channel.new_messages, 'messages': messages, 'channel_url': '%s/customer/help/%s/' % (TUNGA_URL, channel.id) } if send_mail(subject, 'tunga/email/unread_help_messages', to, ctx): channel.last_read = channel.latest_message channel.save()
def handle(self, *args, **options): """ Send new message notifications """ # command to run: python manage.py tunga_send_message_emails utc_now = datetime.datetime.utcnow() min_date = utc_now - relativedelta(minutes=15) # 15 minute window to read new messages min_last_email_date = utc_now - relativedelta(hours=3) # Limit to 1 email every 3 hours per channel commission_date = parse('2016-08-08 00:00:00') # Don't notify about events before the commissioning date user_channels = ChannelUser.objects.filter( Q(last_email_at__isnull=True) | Q(last_email_at__lt=min_last_email_date) ).annotate(new_messages=Sum( Case( When( ~Q(channel__action_targets__actor_object_id=F('user_id')) & Q(channel__action_targets__gt=F('last_read')) & Q(channel__action_targets__timestamp__lte=min_date) & Q(channel__action_targets__timestamp__gte=commission_date) & (Q(last_email_at__isnull=True) | Q(channel__action_targets__timestamp__gt=F('last_email_at'))) & Q(channel__action_targets__verb__in=[verbs.SEND, verbs.UPLOAD]), then=1 ), default=0, output_field=IntegerField() ) )).filter(new_messages__gt=0) for user_channel in user_channels: channel_name = user_channel.channel.get_channel_display_name(user_channel.user) to = [user_channel.user.email] if user_channel.channel.type == CHANNEL_TYPE_DIRECT: conversation_subject = "New message{} from {}".format( user_channel.new_messages == 1 and '' or 's', channel_name ) else: conversation_subject = "Conversation: {}".format(channel_name) subject = conversation_subject ctx = { 'receiver': user_channel.user, 'new_messages': user_channel.new_messages, 'channel_name': channel_name, 'channel': user_channel.channel, 'channel_url': '%s/conversation/%s/' % (TUNGA_URL, user_channel.channel.id) } if send_mail(subject, 'tunga/email/unread_channel_messages', to, ctx): user_channel.last_email_at = datetime.datetime.utcnow() user_channel.save()
