Skip to content

Instantly share code, notes, and snippets.

@dima-dmytruk23
Created June 23, 2022 21:23
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save dima-dmytruk23/0183014d76ad25f4e2a0ca87ec225a10 to your computer and use it in GitHub Desktop.
Save dima-dmytruk23/0183014d76ad25f4e2a0ca87ec225a10 to your computer and use it in GitHub Desktop.
class ExtendedBrandSerializer(serializers.ModelSerializer):
brand_name = serializers.CharField(source='brand__name')
total_sales = serializers.IntegerField()
avg_sales_per_day = serializers.FloatField()
total_revenue = serializers.IntegerField()
avg_revenue_per_day = serializers.FloatField()
rating = serializers.FloatField()
feedbacks = serializers.FloatField()
base_price = serializers.FloatField(default=0)
price = serializers.FloatField(default=0)
discount = serializers.FloatField(default=0)
price_with_discount = serializers.FloatField(default=0)
max_price = serializers.FloatField()
min_price = serializers.FloatField()
avg_price = serializers.FloatField()
history = serializers.SerializerMethodField()
class Meta:
model = Good
fields = (
'brand_id',
'brand_name',
'total_sales',
'avg_sales_per_day',
'total_revenue',
'avg_revenue_per_day',
'rating',
'feedbacks',
'base_price',
'price',
'discount',
'price_with_discount',
'max_price',
'min_price',
'avg_price',
'history',
)
def get_history(self, good: dict) -> dict:
history_filter_args = {'good__brand': data['brand_id']}
if all([key in context for key in ['history_from_date', 'history_to_date']]):
history_filter_args.update({'snap_at__range': (context['from_date'], context['to_date'])})
history_filter_query = Q(**history_filter_args)
return StocksHistorySerializer(
StocksHistory.objects.extra(select={'day': 'date( snap_at )'})
.values('day')
.filter(history_filter_query)
.annotate(
sales=Sum('sales', filter=history_filter_query),
feedbacks=Sum('feedbacks', filter=history_filter_query),
rating=Avg('rating', filter=history_filter_query),
base_price=Avg('base_price', filter=history_filter_query),
price=Avg('price', filter=history_filter_query),
price_with_discount=Avg('price_with_discount', filter=history_filter_query),
revenue=Avg('revenue', filter=history_filter_query),
stock_balances=Avg('stock_balances', filter=history_filter_query),
snap_at=TruncDay('snap_at', filter=history_filter_query),
),
many=True,
).data
class GoodsTopByBrandView(generics.ListAPIView):
serializer_class = ExtendedBrandSerializer
filter_backends = [DjangoFilterBackend, filters.SearchFilter]
ordering = ['-total_sales']
def get_serializer_context(self):
context = super().get_serializer_context()
if all([key in self.request.query_params for key in ['history_from_date', 'history_to_date']]):
from_date, to_date = self._get_date_range()
context['from_date'] = from_date
context['to_date'] = to_date
return context
def get_queryset(self) -> QuerySet[Good]:
subquery_filter_args = {}
history_filter_args = {}
if all([key in self.request.query_params for key in ['history_from_date', 'history_to_date']]):
from_date, to_date = self._get_date_range()
subquery_filter_args.update({'snap_at__range': (from_date, to_date)})
history_filter_args.update({'history__snap_at__range': (from_date, to_date)})
history_filter_query = Q(**history_filter_args)
qs = (
Good.objects.values('brand_id', 'brand__name')
.annotate(
total_sales=Sum('history__sales', filter=history_filter_query),
avg_sales_per_day=Avg('history__sales', filter=history_filter_query),
total_revenue=Sum('history__revenue', filter=history_filter_query),
avg_revenue_per_day=Avg('history__revenue', filter=history_filter_query),
rating=Avg('history__rating', filter=history_filter_query),
feedbacks=Sum('history__feedbacks', filter=history_filter_query),
base_price=Avg('history__base_price', filter=history_filter_query),
price=Avg('history__price', filter=history_filter_query),
price_with_discount=Avg('history__price_with_discount', filter=history_filter_query),
discount=Avg('history__discount', filter=history_filter_query),
max_price=Max('history__price', filter=history_filter_query),
min_price=Min('history__price', filter=history_filter_query),
avg_price=Avg('history__price', filter=history_filter_query),
)
.prefetch_related(Prefetch('history', StocksHistory.objects.filter(Q(**subquery_filter_args))))
.order_by('-total_sales')
)
return qs
def _get_date_range(self) -> tuple[datetime, datetime]:
_date_format = '%Y-%m-%d'
from_date = datetime.strptime(self.request.query_params['history_from_date'], _date_format).replace(tzinfo=utc)
to_date = datetime.strptime(self.request.query_params['history_to_date'], _date_format).replace(tzinfo=utc)
return from_date, to_date
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment