[#106] Implement Traffic Statistics API

digipus/settings.py

@@ -46,6 +46,7 @@ INSTALLED_APPS = [
     "register.apps.RegisterConfig",
     "administration.apps.AdministrationConfig",
     'crispy_forms',
+    "traffic_statistics",
 ]
 
 MIDDLEWARE = [

digipus/urls.py

@@ -24,4 +24,5 @@ urlpatterns = [
     path("", include("authentication.urls"), name="auth"),
     path("", include("app.urls"), name="app"),
     path("administration/", include("administration.urls")),
+    path("statistics/", include("traffic_statistics.urls")),
 ] + static(settings.MEDIA_URL, document_root=settings.MEDIA_ROOT)

traffic_statistics/apps.py

+from django.apps import AppConfig
+
+
+class TrafficStatisticsConfig(AppConfig):
+    name = "traffic_statistics"

traffic_statistics/constants.py

+# URL Parameters
+
+DATE_FORMAT = "%d-%m-%Y"
+DATE_FORMAT_FORMAL = "dd-mm-yyyy"
+
+PARAM_START_DATE = "start_date"
+PARAM_END_DATE = "end_date"
+
+# Errors
+
+ERROR_NULL_DATE_PARAM = "{} parameter must be given and cannot be empty"
+ERROR_INVALID_DATE_PARAM = "{} parameter is invalid, accepted format: " + DATE_FORMAT_FORMAL
+
+# Fields
+
+FIELD_LIKES = "likes"
+FIELD_COMMENTS = "comments"
+FIELD_VIEWS = "views"
+FIELD_DOWNLOADS = "downloads"

traffic_statistics/tests.py

+import json
+from datetime import date, datetime, time, timedelta
+from django.apps import apps
+from django.test import TestCase
+from django.urls import resolve
+from django.utils.timezone import localdate, make_aware
+from random import randrange
+
+from app.models import (
+    Comment,
+    DownloadStatistics,
+    Like,
+    ViewStatistics,
+)
+from traffic_statistics.apps import TrafficStatisticsConfig
+from traffic_statistics.constants import (
+    DATE_FORMAT,
+    PARAM_END_DATE,
+    PARAM_START_DATE,
+
+    ERROR_INVALID_DATE_PARAM,
+    ERROR_NULL_DATE_PARAM,
+
+    FIELD_COMMENTS,
+    FIELD_DOWNLOADS,
+    FIELD_LIKES,
+    FIELD_VIEWS,
+)
+from traffic_statistics.views import StatisticsAPIView
+
+
+def generate_traffic_data(traffic_per_dates):
+
+    def random_time_of_day(date):
+        # to test the daily grouping on any time of the day
+        initial_datetime = datetime.combine(date, time())
+
+        SECS_IN_A_DAY = 24 * 60 * 60
+        generated_datetime = initial_datetime + timedelta(
+            seconds=randrange(SECS_IN_A_DAY)
+        )
+        return generated_datetime
+
+    traffic_objects = [Comment, DownloadStatistics, Like, ViewStatistics]
+    labels = [FIELD_COMMENTS, FIELD_DOWNLOADS, FIELD_LIKES, FIELD_VIEWS]
+
+    traffic_count = {}
+
+    for traffic_date, counts in traffic_per_dates:
+
+        traffic_date_str = traffic_date.strftime(DATE_FORMAT)
+        traffic_count[traffic_date_str] = {}
+
+        i = 0
+        for traffic_object, label in zip(traffic_objects, labels):
+            count = counts[i]
+            for _ in range(count):
+                timestamp = make_aware(random_time_of_day(traffic_date))
+                traffic_object.objects.create(timestamp=timestamp)
+
+            traffic_count[traffic_date_str][label] = count
+
+    return traffic_count
+
+
+class StatisticsAPITest(TestCase):
+
+    @classmethod
+    def setUpTestData(cls):
+        cls.base_url = "/statistics/api/"
+
+    def test_app_name_config(self):
+        self.assertEqual(TrafficStatisticsConfig.name, "traffic_statistics")
+        self.assertEqual(
+            apps.get_app_config("traffic_statistics").name, "traffic_statistics")
+
+    def test_url_resolves_to_statistics_api_view(self):
+        found = resolve(self.base_url)
+        self.assertEqual(found.func.__name__, StatisticsAPIView.as_view().__name__)
+
+    def _run_error_assertion(self, url, error):
+        response = self.client.get(url)
+        data = json.loads(response.content)
+        self.assertEqual(response.status_code, 400)
+        self.assertEqual(data["error"], error)
+
+    def test_api_returns_error_on_unprovided_date_params(self):
+        # both params are empty
+        self._run_error_assertion(
+            self.base_url,
+            ERROR_NULL_DATE_PARAM.format(PARAM_START_DATE),
+        )
+        # start date param is empty
+        self._run_error_assertion(
+            self.base_url + "?end_date=01-10-2020",
+            ERROR_NULL_DATE_PARAM.format(PARAM_START_DATE),
+        )
+        # end date param is empty
+        self._run_error_assertion(
+            self.base_url + "?start_date=01-10-2020",
+            ERROR_NULL_DATE_PARAM.format(PARAM_END_DATE),
+        )
+
+    def test_api_returns_error_on_invalid_date_params_format(self):
+        # invalid start date
+        self._run_error_assertion(
+            self.base_url + "?start_date=20-20-2020&end_date=01-10-2020",
+            ERROR_INVALID_DATE_PARAM.format(PARAM_START_DATE),
+        )
+        # invalid end date
+        self._run_error_assertion(
+            self.base_url + "?start_date=30-09-2020&end_date=2020-10-01",
+            ERROR_INVALID_DATE_PARAM.format(PARAM_END_DATE),
+        )
+
+    def test_api_success_on_valid_date_params(self):
+        response = self.client.get(
+            self.base_url + "?start_date=01-01-2020&end_date=01-01-2020"
+        )
+        self.assertEqual(response.status_code, 200)
+
+    def test_api_returns_correct_daily_traffic_count(self):
+        traffic_dates = [
+            # (date, [comments, downloads, likes, views])
+            (date(year=2020, month=9, day=27), [0, 1, 2, 3]),
+            (date(year=2020, month=9, day=28), [1, 2, 3, 0]),
+            (date(year=2020, month=9, day=29), [2, 3, 0, 1]),
+            (date(year=2020, month=9, day=30), [3, 0, 1, 2]),
+        ]
+
+        traffic_count = generate_traffic_data(traffic_dates)
+        start_date = traffic_dates[0][0].strftime(DATE_FORMAT)
+        end_date = traffic_dates[3][0].strftime(DATE_FORMAT)
+
+        response = self.client.get(
+            self.base_url + f"?start_date={start_date}&end_date={end_date}")
+        self.assertEqual(response.status_code, 200)
+        self.assertJSONEqual(response.content, traffic_count)
+
+    def test_api_returns_up_to_today_at_max(self):
+        today = localdate()
+        tomorrow = today + timedelta(days=1)
+
+        tomorrow_str = tomorrow.strftime(DATE_FORMAT)
+
+        response = self.client.get(
+            self.base_url + f"?start_date={tomorrow_str}&end_date={tomorrow_str}")
+        data = json.loads(response.content)
+
+        self.assertEqual(response.status_code, 200)
+        self.assertFalse(tomorrow_str in data.keys())

traffic_statistics/urls.py

+from django.urls import path
+
+from traffic_statistics.views import StatisticsAPIView
+
+app_name = "traffic_statistics"
+
+urlpatterns = [
+    path("api/", StatisticsAPIView.as_view()),
+]

traffic_statistics/utils/requests.py

+from datetime import datetime
+
+from traffic_statistics.constants import (
+    DATE_FORMAT,
+    PARAM_END_DATE,
+    PARAM_START_DATE,
+
+    ERROR_INVALID_DATE_PARAM,
+    ERROR_NULL_DATE_PARAM,
+)
+
+
+def validate_date_param(date_str_param, param_name):
+    if date_str_param is None:
+        raise ValueError(ERROR_NULL_DATE_PARAM.format(param_name))
+
+    try:
+        return datetime.strptime(date_str_param, DATE_FORMAT).date()
+    except ValueError:
+        raise ValueError(ERROR_INVALID_DATE_PARAM.format(param_name))
+
+
+def extract_date_range_params(request):
+    start_date_param = request.GET.get(PARAM_START_DATE)
+    end_date_param = request.GET.get(PARAM_END_DATE)
+
+    start_date = validate_date_param(start_date_param, PARAM_START_DATE)
+    end_date = validate_date_param(end_date_param, PARAM_END_DATE)
+
+    return start_date, end_date

traffic_statistics/utils/statistics.py

+from datetime import timedelta
+from django.db.models import Count
+from django.db.models.functions import TruncDate
+
+from traffic_statistics.constants import (
+    DATE_FORMAT,
+    FIELD_COMMENTS,
+    FIELD_DOWNLOADS,
+    FIELD_LIKES,
+    FIELD_VIEWS,
+)
+
+
+def get_daily_traffic_count(model, start_date, end_date):
+    queryset = model.objects \
+        .annotate(date=TruncDate('timestamp')) \
+        .filter(date__gte=start_date, date__lte=end_date) \
+        .values('date') \
+        .annotate(count=Count('date')) \
+        .values_list('date', 'count')
+    return queryset
+
+
+def generate_initial_traffic_count_dict(start_date, end_date):
+    date_range_dict = {}
+
+    iter_date = start_date
+    while iter_date <= end_date:
+        date_str = iter_date.strftime(DATE_FORMAT)
+
+        date_range_dict[date_str] = {
+            FIELD_COMMENTS: 0,
+            FIELD_DOWNLOADS: 0,
+            FIELD_LIKES: 0,
+            FIELD_VIEWS: 0,
+        }
+        iter_date += timedelta(days=1)
+
+    return date_range_dict

traffic_statistics/views.py

+from django.http import JsonResponse
+from django.views.generic import ListView
+from django.utils.timezone import localdate
+
+from app.models import (
+    Comment,
+    DownloadStatistics,
+    Like,
+    ViewStatistics,
+)
+from traffic_statistics.constants import (
+    DATE_FORMAT,
+    FIELD_COMMENTS,
+    FIELD_DOWNLOADS,
+    FIELD_LIKES,
+    FIELD_VIEWS,
+)
+from traffic_statistics.utils.requests import extract_date_range_params
+from traffic_statistics.utils.statistics import (
+    generate_initial_traffic_count_dict,
+    get_daily_traffic_count,
+)
+
+
+class StatisticsAPIView(ListView):
+
+    def get(self, *args, **kwargs):
+        try:
+            start_date, end_date = extract_date_range_params(self.request)
+        except Exception as e:
+            return JsonResponse(data={"error": str(e)}, status=400)
+
+        today = localdate()
+        start_date = min(start_date, today)
+        end_date = min(end_date, today)
+
+        statistics_dict = generate_initial_traffic_count_dict(start_date, end_date)
+
+        daily_traffic_counts = [
+            get_daily_traffic_count(Like, start_date, end_date),
+            get_daily_traffic_count(Comment, start_date, end_date),
+            get_daily_traffic_count(ViewStatistics, start_date, end_date),
+            get_daily_traffic_count(DownloadStatistics, start_date, end_date),
+        ]
+        labels = [FIELD_LIKES, FIELD_COMMENTS, FIELD_VIEWS, FIELD_DOWNLOADS]
+
+        for (daily_all_traffic_count, label) in zip(daily_traffic_counts, labels):
+
+            for daily_traffic_count in daily_all_traffic_count:
+                date, count = daily_traffic_count
+
+                date_str = date.strftime(DATE_FORMAT)
+                statistics_dict[date_str][label] = count
+
+        return JsonResponse(data=statistics_dict, status=200)