feat: add api stats endpoint

myapp/admin.py

@@ -1,5 +1,8 @@
 from django.contrib import admin
 from .models import *
+from django.urls import path
+from django.urls import reverse
+from django.utils.html import format_html
 from django.utils.translation import gettext_lazy as _
 
 class UserProfileAdmin(admin.ModelAdmin):
@@ -15,5 +18,49 @@ class ItemAdmin(admin.ModelAdmin):
     # Specify the filters to use in the list view
     list_filter = ('type', 'is_used', 'issue_date', 'expiry_date', 'user')
 
+
+class AppSettingsAdmin(admin.ModelAdmin):
+    list_display = ('api_token', 'updated_at', 'regenerate_token_button')
+    readonly_fields = ('api_token', 'updated_at', 'regenerate_token_button')  # Include the button here
+
+    def has_add_permission(self, request):
+        # Check if there is already an existing object in the model
+        if self.model.objects.count() >= 1:
+            return False  # Disallow adding a new object
+        else:
+            return True   # Allow adding a new object    
+
+    def regenerate_token_button(self, obj):
+        """Add a button to regenerate the API token."""
+        if obj.pk:  # Only display the button for existing objects
+            url = reverse('admin:regenerate_api_token', args=[obj.pk])
+            return format_html(
+                '<a class="button" href="{}">Regenerate API Token</a>', url
+            )
+        return "Save this object before regenerating the token."
+    regenerate_token_button.short_description = "Actions"
+    regenerate_token_button.allow_tags = True
+
+    def get_urls(self):
+        """Extend the admin URL patterns to include custom actions."""
+        urls = super().get_urls()
+        custom_urls = [
+            path('<int:pk>/regenerate-token/', self.admin_site.admin_view(self.regenerate_token), name='regenerate_api_token'),
+        ]
+        return custom_urls + urls
+
+    def regenerate_token(self, request, pk):
+        """Regenerate the API token for the selected AppSettings instance."""
+        from django.http import HttpResponseRedirect
+        from django.contrib import messages
+        try:
+            app_settings = AppSettings.objects.get(pk=pk)
+            app_settings.regenerate_api_token()
+            messages.success(request, "API token regenerated successfully!")
+        except AppSettings.DoesNotExist:
+            messages.error(request, "AppSettings instance not found.")
+        return HttpResponseRedirect(reverse('admin:myapp_appsettings_changelist'))
+
 admin.site.register(Item, ItemAdmin)
-admin.site.register(Transaction)
+admin.site.register(Transaction)
+admin.site.register(AppSettings, AppSettingsAdmin)

myapp/decorators.py

@@ -0,0 +1,31 @@
+from functools import wraps
+from django.http import HttpResponseForbidden
+from myapp.models import *
+import logging
+
+logger = logging.getLogger(__name__)
+
+def require_authorization_header_with_api_token(view_func):
+    @wraps(view_func)
+    def _wrapped_view(request, *args, **kwargs):
+        # Get the latest AppSettings object from the database
+        app_settings = AppSettings.objects.last()
+        if not app_settings:
+            logger.error('No API key configured. Returning 403 Forbidden.')
+            return HttpResponseForbidden("Unauthorized. Invalid or missing authorization token.")
+
+        # Retrieve the API token from AppSettings
+        api_token = app_settings.api_token
+
+        # Get the Authorization header from the request
+        authorization_header = request.META.get('HTTP_AUTHORIZATION')
+
+        # Check if the Authorization header is present and matches the API token
+        if authorization_header != f'Bearer {api_token}':
+            logger.info('Request with invalid or missing API token. Returning 403 Forbidden.')
+            return HttpResponseForbidden("Unauthorized. Invalid or missing authorization token.")
+
+        # Token is valid; proceed with the request
+        return view_func(request, *args, **kwargs)
+
+    return _wrapped_view

myapp/migrations/0012_appsettings.py

@@ -0,0 +1,22 @@
+# Generated by Django 5.1.3 on 2024-11-28 10:34
+
+import uuid
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('myapp', '0011_alter_item_expiry_date_alter_item_issue_date_and_more'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='AppSettings',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('api_token', models.CharField(default=uuid.uuid4, max_length=64, unique=True)),
+                ('updated_at', models.DateTimeField(auto_now=True)),
+            ],
+        ),
+    ]

myapp/migrations/0013_alter_appsettings_options.py

@@ -0,0 +1,17 @@
+# Generated by Django 5.1.3 on 2024-11-28 11:49
+
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('myapp', '0012_appsettings'),
+    ]
+
+    operations = [
+        migrations.AlterModelOptions(
+            name='appsettings',
+            options={'verbose_name': 'API Settings', 'verbose_name_plural': 'API Settings'},
+        ),
+    ]

myapp/models.py

@@ -1,6 +1,7 @@
 from django.db import models
 from django.utils import timezone
 from django.contrib.auth.models import User
+from django.core.exceptions import ValidationError
 import uuid
 
 class Item(models.Model):
@@ -58,4 +59,64 @@ class UserProfile(models.Model):
     apprise_urls = models.TextField(blank=True, null=True)
 
     def __str__(self):
-        return self.user.username
+        return self.user.username
+
+class AppSettings(models.Model):
+    """
+    Model for storing API token for authentication.
+
+    This model enforces a singleton pattern to ensure only one set of API settings exists.
+    The API token is used for authenticating API requests.
+
+    API Usage:
+    - Endpoint: /en/api/get/stats
+    - Method: GET
+    - Authorization: Requires an API token provided in the `Authorization` header
+      in the format: `Authorization: Bearer <API-TOKEN>`
+    - Description: Retrieves statistical data about items, users, and issuers.
+
+    Example:
+    ```
+    curl -H "Authorization: Bearer <API-TOKEN>" http://<your-domain>/api/get/stats
+    ```
+
+    Attributes:
+    - api_token: A unique token used for API authentication.
+    - updated_at: Timestamp of the last update to the API token.
+
+    Methods:
+    - regenerate_api_token: Generates a new API token and updates the `updated_at` field.
+    """
+
+    api_token = models.CharField(max_length=64, default=uuid.uuid4, unique=True)
+    updated_at = models.DateTimeField(auto_now=True)
+
+    class Meta:
+        verbose_name = "API Settings"
+        verbose_name_plural = "API Settings"
+
+    def regenerate_api_token(self):
+        """Generate a new API token."""
+        self.api_token = str(uuid.uuid4())  # Ensure it's saved as a string
+        self.save()
+
+    def save(self, *args, **kwargs):
+        """Override save to enforce singleton behavior and validate the API token."""
+        # Ensure only one instance exists
+        if not self.pk and AppSettings.objects.exists():
+            raise ValueError("Only one AppSettings instance is allowed.")
+
+        # Validate the API token is a valid UUID
+        if not isinstance(self.api_token, str):
+            self.api_token = str(self.api_token)  # Convert to string if it's a UUID object
+        try:
+            uuid_obj = uuid.UUID(self.api_token)  # Validate if it's a valid UUID string
+            self.api_token = str(uuid_obj)  # Normalize to UUID string format
+        except ValueError:
+            raise ValueError("The API token must be a valid UUID.")
+
+        super().save(*args, **kwargs)
+
+    def __str__(self):
+        return f"API Token (Updated: {self.updated_at})"
+

myapp/urls.py

@@ -22,6 +22,7 @@
     path('verify-apprise-urls/', views.verify_apprise_urls, name='verify_apprise_urls'),
     path('download/<uuid:item_id>/', views.download_file, name='download_file'),
     path('shared-items/', views.sharing_center, name='sharing_center'),
+    path('api/get/stats', views.get_stats, name='get_stats'),
 )
 
 admin.site.site_header = "VoucherVault"

myapp/views.py

@@ -19,6 +19,12 @@
 from django.conf import settings
 from django.utils.translation import gettext_lazy as _
 from django.contrib import messages
+from django.utils.timezone import now
+from .decorators import require_authorization_header_with_api_token
+from django.db.models import Count, Sum, Q, F, ExpressionWrapper, DecimalField
+from django.db.models.functions import Coalesce
+from django.db.models import Value
+
 
 apprise_txt = _('Apprise URLs were already configured. Will not display them again here to protect secrets. You can freely re-configure the URLs now and hit update though.')
 
@@ -468,4 +474,97 @@ def unshare_item(request, item_id, user_id):
     messages.success(request, "Item has been unshared successfully.")
 
     # Redirect back to the item view page
-    return redirect('view_item', item_uuid=item.id)
+    return redirect('view_item', item_uuid=item.id)
+
+# API
+
+@require_authorization_header_with_api_token
+
+def get_items_by_type(request, item_type):
+    authenticate_general_api_key(request)
+    items = Item.objects.filter(type=item_type).values()
+    return JsonResponse(list(items), safe=False)
+
+@require_authorization_header_with_api_token
+def get_stats(request):
+
+    # Calculate the total value of active, unused, and non-expired items, considering transactions
+    items_with_transaction_values = (
+        Item.objects.filter(is_used=False, expiry_date__gte=now())  # Exclude used and expired items
+        .annotate(
+            transaction_total=Sum('transactions__value', default=0)  # Sum of related transaction values
+        )
+        .annotate(net_value=ExpressionWrapper(F('value') + F('transaction_total'), output_field=models.DecimalField()))
+    )
+
+    total_value = round((items_with_transaction_values.aggregate(total_value=Sum('net_value'))['total_value'] or 0),2)
+
+    # Item stats
+    item_stats = {
+        "total_items": Item.objects.count(),
+        "total_value": total_value,  # Net value only for valid items
+        "vouchers": Item.objects.filter(type='voucher').count(),
+        "giftcards": Item.objects.filter(type='giftcard').count(),
+        "coupons": Item.objects.filter(type='coupon').count(),
+        "loyaltycards": Item.objects.filter(type='loyaltycard').count(),
+        "used_items": Item.objects.filter(is_used=True).count(),
+        "unused_items": Item.objects.filter(is_used=False).count(),
+        "expired_items": Item.objects.filter(expiry_date__lt=now()).count(),
+    }
+
+    # User stats
+    user_stats = {
+        "total_users": User.objects.count(),
+        "active_users": User.objects.filter(is_active=True).count(),
+        "disabled_users": User.objects.filter(is_active=False).count(),
+        "superusers": User.objects.filter(is_superuser=True).count(),
+        "staff_members": User.objects.filter(is_staff=True).count(),
+    }
+
+
+    # Calculate the total transaction values per issuer
+    issuer_transaction_totals = (
+        Item.objects.filter(is_used=False, expiry_date__gte=now())  # Only active, non-expired items
+        .values('issuer')
+        .annotate(
+            transaction_total=Coalesce(
+                Sum('transactions__value', output_field=DecimalField()), 
+                Value(0, output_field=DecimalField())
+            )  # Sum of transaction values with output field defined
+        )
+    )
+
+    # Map issuer to transaction totals for easier lookup
+    issuer_transaction_map = {item['issuer']: item['transaction_total'] for item in issuer_transaction_totals}
+
+    # Calculate issuer stats with count and base value
+    issuers = (
+        Item.objects.filter(is_used=False, expiry_date__gte=now())  # Only active, non-expired items
+        .values('issuer')
+        .annotate(
+            count=Count('issuer'),
+            base_value=Coalesce(
+                Sum('value', output_field=DecimalField()), 
+                Value(0, output_field=DecimalField())
+            )  # Sum of item values with output field defined
+        )
+        .order_by('-count')  # Optional: order by count
+    )
+
+    # Combine the values and transactions for the final total
+    issuer_stats = [
+        {
+            "issuer": issuer["issuer"],
+            "count": issuer["count"],
+            "total_value": round((issuer["base_value"] + issuer_transaction_map.get(issuer["issuer"], 0)),2),  # Add base and transaction totals
+        }
+        for issuer in issuers
+    ]
+
+
+    # Combine both stats into one response
+    return JsonResponse({
+        "item_stats": item_stats,
+        "user_stats": user_stats,
+        "issuer_stats": issuer_stats,
+    })