Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
minutes=self.context['event'].settings.get('reservation_time', as_type=int)
)
with self.context['event'].lock():
new_quotas = (validated_data.get('variation').quotas.filter(subevent=validated_data.get('subevent'))
if validated_data.get('variation')
else validated_data.get('item').quotas.filter(subevent=validated_data.get('subevent')))
if len(new_quotas) == 0:
raise ValidationError(
ugettext_lazy('The product "{}" is not assigned to a quota.').format(
str(validated_data.get('item'))
)
)
for quota in new_quotas:
avail = quota.availability()
if avail[0] != Quota.AVAILABILITY_OK or (avail[1] is not None and avail[1] < 1):
raise ValidationError(
ugettext_lazy('There is not enough quota available on quota "{}" to perform '
'the operation.').format(
quota.name
)
)
attendee_name = validated_data.pop('attendee_name', '')
if attendee_name and not validated_data.get('attendee_name_parts'):
validated_data['attendee_name_parts'] = {
'_legacy': attendee_name
}
seated = validated_data.get('item').seat_category_mappings.filter(subevent=validated_data.get('subevent')).exists()
if validated_data.get('seat'):
if not seated:
raise ValidationError('The specified product does not allow to choose a seat.')
@transaction.atomic
def create(self, validated_data):
options_data = validated_data.pop('options') if 'options' in validated_data else []
items = validated_data.pop('items')
question = Question.objects.create(**validated_data)
question.items.set(items)
for opt_data in options_data:
QuestionOption.objects.create(question=question, **opt_data)
return question
class QuotaSerializer(I18nAwareModelSerializer):
class Meta:
model = Quota
fields = ('id', 'name', 'size', 'items', 'variations', 'subevent', 'closed', 'close_when_sold_out')
def validate(self, data):
data = super().validate(data)
event = self.context['event']
full_data = self.to_internal_value(self.to_representation(self.instance)) if self.instance else {}
full_data.update(data)
Quota.clean_variations(full_data.get('items'), full_data.get('variations'))
Quota.clean_items(event, full_data.get('items'), full_data.get('variations'))
Quota.clean_subevent(event, full_data.get('subevent'))
return data
)
elif event.settings.display_net_prices:
n = _('{name} (+ {price} plus {taxes}% {taxname})').format(
name=label, price=money_filter(price.net, event.currency),
taxes=number_format(price.rate), taxname=price.name
)
else:
n = _('{name} (+ {price} incl. {taxes}% {taxname})').format(
name=label, price=money_filter(price.gross, event.currency),
taxes=number_format(price.rate), taxname=price.name
)
if not initial:
if avail[0] < Quota.AVAILABILITY_RESERVED:
n += ' – {}'.format(_('SOLD OUT'))
elif avail[0] < Quota.AVAILABILITY_OK:
n += ' – {}'.format(_('Currently unavailable'))
else:
if avail[1] is not None and item.do_show_quota_left:
n += ' – {}'.format(_('%(num)s currently available') % {'num': avail[1]})
if not isinstance(item_or_variation, ItemVariation) and item.picture:
n = escape(n)
n += '<br>'
n += '<a data-lightbox="{}" data-title="{}" class="productpicture" href="{}">'.format(
item.picture.url, escape(escape(item.name)), item.id
)
n += '<img alt="{}" src="{}">'.format(
thumb(item.picture, '60x60^'),
escape(item.name)
)
n += '</a>'
def execute_payment(self, request: HttpRequest, payment: OrderPayment):
try:
payment.confirm(send_mail=False)
except Quota.QuotaExceededException as e:
raise PaymentException(str(e))
cp.includes_tax = bool(price.rate)
cp.save()
err = err or error_messages['price_changed']
continue
quota_ok = True
ignore_all_quotas = cp.expires >= now_dt or (
cp.voucher and (cp.voucher.allow_ignore_quota or (cp.voucher.block_quota and cp.voucher.quota is None)))
if not ignore_all_quotas:
for quota in quotas:
if cp.voucher and cp.voucher.block_quota and cp.voucher.quota_id == quota.pk:
continue
avail = quota.availability(now_dt)
if avail[0] != Quota.AVAILABILITY_OK:
# This quota is sold out/currently unavailable, so do not sell this at all
err = err or error_messages['unavailable']
quota_ok = False
break
if quota_ok:
cp.expires = now_dt + timedelta(
minutes=event.settings.get('reservation_time', as_type=int))
cp.save()
else:
# Sorry, can't let you keep that!
delete(cp)
if err:
raise OrderError(err, errargs)
def approve(self, request, **kwargs):
send_mail = request.data.get('send_email', True)
order = self.get_object()
try:
approve_order(
order,
user=request.user if request.user.is_authenticated else None,
auth=request.auth if isinstance(request.auth, (Device, TeamAPIToken, OAuthAccessToken)) else None,
send_mail=send_mail,
)
except Quota.QuotaExceededException as e:
return Response({'detail': str(e)}, status=status.HTTP_400_BAD_REQUEST)
except OrderError as e:
return Response({'detail': str(e)}, status=status.HTTP_400_BAD_REQUEST)
return self.retrieve(request, [], **kwargs)
def _get_availability(self, ev, event):
availability = {}
if ev.presale_is_running and event.settings.event_list_availability and ev.best_availability_state is not None:
if ev.best_availability_state == Quota.AVAILABILITY_OK:
availability['color'] = 'green'
availability['text'] = ugettext('Book now')
elif event.settings.waiting_list_enabled and ev.best_availability_state >= 0:
availability['color'] = 'orange'
availability['text'] = ugettext('Waiting list')
elif ev.best_availability_state == Quota.AVAILABILITY_RESERVED:
availability['color'] = 'orange'
availability['text'] = ugettext('Reserved')
elif ev.best_availability_state < Quota.AVAILABILITY_RESERVED:
availability['color'] = 'red'
availability['text'] = ugettext('Sold out')
elif ev.presale_is_running:
availability['color'] = 'green'
availability['text'] = ugettext('Book now')
elif ev.presale_has_ended:
availability['color'] = 'red'
availability['text'] = ugettext('Sale over')
elif event.settings.presale_start_show_date and ev.presale_start:
availability['color'] = 'orange'
availability['text'] = ugettext('from %(start_date)s') % {'start_date': date_format(ev.presale_start, "SHORT_DATE_FORMAT")}
else:
availability['color'] = 'orange'
availability['text'] = ugettext('Sale soon')
return availability
if form.cleaned_data.get('type') in ('M', 'C'):
if not self.formset.is_valid():
return self.get(self.request, *self.args, **self.kwargs)
messages.success(self.request, _('The new question has been created.'))
ret = super().form_valid(form)
form.instance.log_action('pretix.event.question.added', user=self.request.user, data=dict(form.cleaned_data))
if form.cleaned_data.get('type') in ('M', 'C'):
self.save_formset(form.instance)
return ret
class QuotaList(PaginationMixin, ListView):
model = Quota
context_object_name = 'quotas'
template_name = 'pretixcontrol/items/quotas.html'
def get_queryset(self):
qs = Quota.objects.filter(
event=self.request.event
).prefetch_related(
Prefetch(
"items",
queryset=Item.objects.annotate(
has_variations=Exists(ItemVariation.objects.filter(item=OuterRef('pk')))
),
to_attr="cached_items"
),
"variations",
"variations__item"
order_to=Coalesce('max_fromto', 'max_to', 'max_from', 'date_to', 'date_from'),
)
sum_tickets_paid = Quota.objects.filter(
event=OuterRef('pk'), subevent__isnull=True
).order_by().values('event').annotate(
s=Sum('cached_availability_paid_orders')
).values(
's'
)
qs = qs.annotate(
sum_tickets_paid=Subquery(sum_tickets_paid, output_field=IntegerField())
).prefetch_related(
Prefetch('quotas',
queryset=Quota.objects.filter(subevent__isnull=True).annotate(s=Coalesce(F('size'), 0)).order_by('-s'),
to_attr='first_quotas')
)
if self.filter_form.is_valid():
qs = self.filter_form.filter_qs(qs)
return qs
def get_queryset(self):
qs = Quota.objects.filter(
event=self.request.event
).prefetch_related(
Prefetch(
"items",
queryset=Item.objects.annotate(
has_variations=Exists(ItemVariation.objects.filter(item=OuterRef('pk')))
),
to_attr="cached_items"
),
"variations",
"variations__item"
)
if self.request.GET.get("subevent", "") != "":
s = self.request.GET.get("subevent", "")
qs = qs.filter(subevent_id=s)
return qs