Shayan/site_backend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix permissions and prepare admin login for deploy

Browse Source 
- Admin-only: list contacts, create/edit campaigns and compositions
- Public: submit contact, mine contacts by email/phone, read campaigns/compositions
- Add admin_response field on ContactUs
- Update Postman base_url to https://zoneco.org/api with token save

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Shayan Azadi
2026-07-22 22:52:44 +03:30
parent 97127d5772
commit 149fb614b0
8 changed files with 425 additions and 220 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
api/admin.py
View File

@@ -4,9 +4,9 @@ from .models import ContactUs, Composition, Campaign, CompositionImage
@admin.register(ContactUs)
class ContactUsAdmin(admin.ModelAdmin):
list_display = ['name', 'email_or_phone', 'category', 'created_at']
list_display = ['name', 'email_or_phone', 'category', 'has_admin_response', 'created_at']
list_filter = ['category', 'created_at']
search_fields = ['name', 'email_or_phone', 'description']
search_fields = ['name', 'email_or_phone', 'description', 'admin_response']
readonly_fields = ['created_at', 'updated_at']
date_hierarchy = 'created_at'
@@ -17,12 +17,19 @@ class ContactUsAdmin(admin.ModelAdmin):
('پیام', {
'fields': ('description',)
}),
('پاسخ ادمین', {
'fields': ('admin_response',)
}),
('اطلاعات زمانی', {
'fields': ('created_at', 'updated_at'),
'classes': ('collapse',)
}),
)
@admin.display(boolean=True, description='پاسخ ادمین')
def has_admin_response(self, obj):
return bool(obj.admin_response and obj.admin_response.strip())
class CompositionImageInline(admin.TabularInline):
model = CompositionImage

22
api/migrations/0003_contact_admin_response.py Normal file
View File

@@ -0,0 +1,22 @@
# Generated by Django 4.2.7 on 2026-07-22 19:20
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('api', '0002_composition_images'),
]
operations = [
migrations.AddField(
model_name='contactus',
name='admin_response',
field=models.TextField(blank=True, default='', help_text='Admin reply visible to the contact submitter', verbose_name='پاسخ ادمین'),
),
migrations.AddIndex(
model_name='contactus',
index=models.Index(fields=['email_or_phone'], name='api_contact_email_o_fedc42_idx'),
),
]

7
api/models.py
View File

@@ -27,6 +27,12 @@ class ContactUs(models.Model):
choices=CATEGORY_CHOICES,
verbose_name='دسته‌بندی'
)
admin_response = models.TextField(
blank=True,
default='',
verbose_name='پاسخ ادمین',
help_text='Admin reply visible to the contact submitter',
)
created_at = models.DateTimeField(auto_now_add=True, verbose_name='تاریخ ایجاد')
updated_at = models.DateTimeField(auto_now=True, verbose_name='تاریخ بروزرسانی')
@@ -37,6 +43,7 @@ class ContactUs(models.Model):
indexes = [
models.Index(fields=['-created_at']),
models.Index(fields=['category']),
models.Index(fields=['email_or_phone']),
]
def __str__(self):

7
api/serializers.py
View File

@@ -32,11 +32,14 @@ class AdminLoginSerializer(serializers.Serializer):
class ContactUsSerializer(serializers.ModelSerializer):
"""
Serializer for ContactUs model.
Public create cannot set admin_response (enforced in the viewset).
"""
class Meta:
model = ContactUs
fields = ['id', 'name', 'email_or_phone', 'description', 'category',
'created_at', 'updated_at']
fields = [
'id', 'name', 'email_or_phone', 'description', 'category',
'admin_response', 'created_at', 'updated_at',
]
read_only_fields = ['id', 'created_at', 'updated_at']
def validate_email_or_phone(self, value):

216
api/tests.py
View File

@@ -6,13 +6,13 @@ from django.core.files.uploadedfile import SimpleUploadedFile
from django.utils import timezone
from PIL import Image
from rest_framework import status
from rest_framework.authtoken.models import Token
from rest_framework.test import APITestCase
from .models import Campaign, Composition, CompositionImage, ContactUs
def make_test_image(name='test.jpg', color='red', size=(100, 100)):
"""Create a minimal valid JPEG for upload tests."""
buffer = io.BytesIO()
Image.new('RGB', size, color=color).save(buffer, format='JPEG')
buffer.seek(0)
@@ -21,6 +21,14 @@ def make_test_image(name='test.jpg', color='red', size=(100, 100)):
class ContactUsAPITests(APITestCase):
def setUp(self):
User = get_user_model()
self.admin = User.objects.create_user(
username='admin_test',
password='test_admin_pass_123',
is_staff=True,
is_superuser=True,
)
self.token = Token.objects.create(user=self.admin)
self.contact = ContactUs.objects.create(
name='Ali Reza',
email_or_phone='ali@example.com',
@@ -28,12 +36,20 @@ class ContactUsAPITests(APITestCase):
category='پشتیبانی',
)
def test_list_contacts(self):
def test_list_contacts_requires_admin(self):
response = self.client.get('/api/contact-us/')
self.assertIn(response.status_code, (
status.HTTP_401_UNAUTHORIZED,
status.HTTP_403_FORBIDDEN,
))
def test_list_contacts_as_admin(self):
self.client.credentials(HTTP_AUTHORIZATION=f'Token {self.token.key}')
response = self.client.get('/api/contact-us/')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['count'], 1)
def test_create_contact(self):
def test_create_contact_public(self):
payload = {
'name': 'Sara',
'email_or_phone': '09121234567',
@@ -43,69 +59,101 @@ class ContactUsAPITests(APITestCase):
response = self.client.post('/api/contact-us/', payload, format='json')
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
self.assertEqual(ContactUs.objects.count(), 2)
self.assertEqual(response.data.get('admin_response'), '')
def test_retrieve_contact(self):
response = self.client.get(f'/api/contact-us/{self.contact.id}/')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['name'], 'Ali Reza')
def test_public_cannot_set_admin_response_on_create(self):
payload = {
'name': 'Sara',
'email_or_phone': 'sara@example.com',
'description': 'Hello',
'category': 'سایر',
'admin_response': 'should be ignored',
}
response = self.client.post('/api/contact-us/', payload, format='json')
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
contact = ContactUs.objects.get(pk=response.data['id'])
self.assertEqual(contact.admin_response, '')
def test_by_category(self):
response = self.client.get('/api/contact-us/by_category/?category=پشتیبانی')
def test_mine_contacts(self):
ContactUs.objects.create(
name='Other',
email_or_phone='other@example.com',
description='x',
category='سایر',
)
response = self.client.get(
'/api/contact-us/mine/?email_or_phone=ali@example.com'
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(len(response.data), 1)
self.assertEqual(response.data[0]['email_or_phone'], 'ali@example.com')
def test_by_category_missing_param(self):
response = self.client.get('/api/contact-us/by_category/')
def test_mine_requires_email_or_phone(self):
response = self.client.get('/api/contact-us/mine/')
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
def test_update_requires_auth(self):
response = self.client.patch(
f'/api/contact-us/{self.contact.id}/',
{'name': 'Updated'},
format='json',
)
def test_by_category_requires_admin(self):
response = self.client.get('/api/contact-us/by_category/?category=پشتیبانی')
self.assertIn(response.status_code, (
status.HTTP_401_UNAUTHORIZED,
status.HTTP_403_FORBIDDEN,
))
def test_admin_can_reply(self):
self.client.credentials(HTTP_AUTHORIZATION=f'Token {self.token.key}')
response = self.client.patch(
f'/api/contact-us/{self.contact.id}/',
{'admin_response': 'We will help you.'},
format='json',
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['admin_response'], 'We will help you.')
class CompositionAPITests(APITestCase):
def setUp(self):
User = get_user_model()
self.admin = User.objects.create_user(
username='testadmin', password='testpass123', is_staff=True
username='admin_test',
password='test_admin_pass_123',
is_staff=True,
)
self.token = Token.objects.create(user=self.admin)
self.composition = Composition.objects.create(
name='Test Composition',
description='Test description',
)
def test_list_compositions(self):
def test_list_compositions_public(self):
response = self.client.get('/api/compositions/')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['count'], 1)
def test_create_composition_json(self):
def test_create_composition_requires_admin(self):
payload = {'name': 'New Comp', 'description': 'Desc'}
response = self.client.post('/api/compositions/', payload, format='json')
self.assertIn(response.status_code, (
status.HTTP_401_UNAUTHORIZED,
status.HTTP_403_FORBIDDEN,
))
def test_create_composition_as_admin(self):
self.client.credentials(HTTP_AUTHORIZATION=f'Token {self.token.key}')
payload = {'name': 'New Comp', 'description': 'Desc'}
response = self.client.post('/api/compositions/', payload, format='json')
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
self.assertEqual(response.data['name'], 'New Comp')
self.assertEqual(response.data['images'], [])
def test_retrieve_composition(self):
def test_retrieve_composition_public(self):
response = self.client.get(f'/api/compositions/{self.composition.id}/')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertIn('images', response.data)
self.assertIn('main_image', response.data)
self.assertIn('created_at', response.data)
def test_by_created_at(self):
def test_by_created_at_public(self):
now = timezone.now()
Composition.objects.create(
name='Old Composition',
description='Old',
)
Composition.objects.create(name='Old Composition', description='Old')
Composition.objects.filter(name='Old Composition').update(
created_at=now - timedelta(days=10)
)
@@ -118,17 +166,8 @@ class CompositionAPITests(APITestCase):
self.assertIn('Test Composition', names)
self.assertNotIn('Old Composition', names)
def test_by_created_at_missing_params(self):
response = self.client.get('/api/compositions/by-created-at/')
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
def test_by_created_at_invalid_datetime(self):
response = self.client.get(
'/api/compositions/by-created-at/?from=not-a-date'
)
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
def test_create_with_multiple_images(self):
def test_create_with_multiple_images_as_admin(self):
self.client.credentials(HTTP_AUTHORIZATION=f'Token {self.token.key}')
img1 = make_test_image('img1.jpg', 'red')
img2 = make_test_image('img2.jpg', 'blue')
response = self.client.post(
@@ -145,12 +184,8 @@ class CompositionAPITests(APITestCase):
self.assertEqual(len(response.data['images']), 2)
self.assertTrue(response.data['main_image']['is_main'])
composition = Composition.objects.get(name='Multi Image')
self.assertEqual(composition.images.count(), 2)
self.assertEqual(composition.main_image.is_main, True)
def test_add_images(self):
self.client.force_authenticate(user=self.admin)
def test_add_images_as_admin(self):
self.client.credentials(HTTP_AUTHORIZATION=f'Token {self.token.key}')
img = make_test_image('added.jpg', 'green')
response = self.client.post(
f'/api/compositions/{self.composition.id}/add-images/',
@@ -159,10 +194,9 @@ class CompositionAPITests(APITestCase):
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(len(response.data['images']), 1)
self.assertTrue(response.data['main_image']['is_main'])
def test_set_main_image(self):
self.client.force_authenticate(user=self.admin)
def test_set_main_image_as_admin(self):
self.client.credentials(HTTP_AUTHORIZATION=f'Token {self.token.key}')
img1 = CompositionImage.objects.create(
composition=self.composition,
image=make_test_image('a.jpg'),
@@ -184,8 +218,8 @@ class CompositionAPITests(APITestCase):
self.assertFalse(img1.is_main)
self.assertTrue(img2.is_main)
def test_delete_image(self):
self.client.force_authenticate(user=self.admin)
def test_delete_image_as_admin(self):
self.client.credentials(HTTP_AUTHORIZATION=f'Token {self.token.key}')
img = CompositionImage.objects.create(
composition=self.composition,
image=make_test_image('del.jpg'),
@@ -197,20 +231,16 @@ class CompositionAPITests(APITestCase):
self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
self.assertFalse(CompositionImage.objects.filter(pk=img.id).exists())
def test_update_requires_auth(self):
response = self.client.patch(
f'/api/compositions/{self.composition.id}/',
{'name': 'Updated'},
format='json',
)
self.assertIn(response.status_code, (
status.HTTP_401_UNAUTHORIZED,
status.HTTP_403_FORBIDDEN,
))
class CampaignAPITests(APITestCase):
def setUp(self):
User = get_user_model()
self.admin = User.objects.create_user(
username='admin_test',
password='test_admin_pass_123',
is_staff=True,
)
self.token = Token.objects.create(user=self.admin)
now = timezone.now()
self.active = Campaign.objects.create(
name='Active Campaign',
@@ -231,12 +261,27 @@ class CampaignAPITests(APITestCase):
end_time=now - timedelta(days=5),
)
def test_list_campaigns(self):
def test_list_campaigns_public(self):
response = self.client.get('/api/campaigns/')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['count'], 3)
def test_create_campaign(self):
def test_create_campaign_requires_admin(self):
now = timezone.now()
payload = {
'name': 'New Campaign',
'description': 'Desc',
'start_time': (now + timedelta(days=1)).isoformat(),
'end_time': (now + timedelta(days=3)).isoformat(),
}
response = self.client.post('/api/campaigns/', payload, format='json')
self.assertIn(response.status_code, (
status.HTTP_401_UNAUTHORIZED,
status.HTTP_403_FORBIDDEN,
))
def test_create_campaign_as_admin(self):
self.client.credentials(HTTP_AUTHORIZATION=f'Token {self.token.key}')
now = timezone.now()
payload = {
'name': 'New Campaign',
@@ -247,19 +292,16 @@ class CampaignAPITests(APITestCase):
response = self.client.post('/api/campaigns/', payload, format='json')
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
def test_retrieve_campaign(self):
def test_retrieve_campaign_public(self):
response = self.client.get(f'/api/campaigns/{self.active.id}/')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertTrue(response.data['is_active'])
self.assertFalse(response.data['is_upcoming'])
self.assertFalse(response.data['is_ended'])
def test_active_campaigns(self):
response = self.client.get('/api/campaigns/active/')
self.assertEqual(response.status_code, status.HTTP_200_OK)
names = [item['name'] for item in response.data]
self.assertIn('Active Campaign', names)
self.assertNotIn('Upcoming Campaign', names)
def test_upcoming_campaigns(self):
response = self.client.get('/api/campaigns/upcoming/')
@@ -273,17 +315,6 @@ class CampaignAPITests(APITestCase):
names = [item['name'] for item in response.data]
self.assertIn('Ended Campaign', names)
def test_create_campaign_invalid_dates(self):
now = timezone.now()
payload = {
'name': 'Bad Dates',
'description': 'Desc',
'start_time': (now + timedelta(days=3)).isoformat(),
'end_time': (now + timedelta(days=1)).isoformat(),
}
response = self.client.post('/api/campaigns/', payload, format='json')
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
class AdminLoginAPITests(APITestCase):
def setUp(self):
@@ -311,7 +342,6 @@ class AdminLoginAPITests(APITestCase):
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertIn('token', response.data)
self.assertEqual(response.data['user']['username'], self.admin_username)
self.assertTrue(response.data['user']['is_staff'])
def test_admin_login_wrong_password(self):
response = self.client.post(
@@ -341,34 +371,6 @@ class AdminLoginAPITests(APITestCase):
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['username'], self.admin_username)
def test_admin_me_requires_auth(self):
response = self.client.get('/api/admin/me/')
self.assertIn(response.status_code, (
status.HTTP_401_UNAUTHORIZED,
status.HTTP_403_FORBIDDEN,
))
def test_token_allows_protected_update(self):
contact = ContactUs.objects.create(
name='Old Name',
email_or_phone='a@b.com',
description='msg',
category='سایر',
)
login = self.client.post(
'/api/admin/login/',
{'username': self.admin_username, 'password': self.admin_password},
format='json',
)
self.client.credentials(HTTP_AUTHORIZATION=f'Token {login.data["token"]}')
response = self.client.patch(
f'/api/contact-us/{contact.id}/',
{'name': 'New Name'},
format='json',
)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['name'], 'New Name')
def test_admin_logout(self):
login = self.client.post(
'/api/admin/login/',
@@ -379,8 +381,6 @@ class AdminLoginAPITests(APITestCase):
self.client.credentials(HTTP_AUTHORIZATION=f'Token {token}')
response = self.client.post('/api/admin/logout/')
self.assertEqual(response.status_code, status.HTTP_200_OK)
# Token should no longer work
response = self.client.get('/api/admin/me/')
self.assertIn(response.status_code, (
status.HTTP_401_UNAUTHORIZED,

125
api/views.py
View File

@@ -1,11 +1,10 @@
from rest_framework import viewsets, status
from rest_framework.decorators import action, api_view, permission_classes
from rest_framework.response import Response
from rest_framework.permissions import AllowAny, IsAuthenticated
from rest_framework.permissions import AllowAny, IsAuthenticated, IsAdminUser
from rest_framework.parsers import MultiPartParser, FormParser, JSONParser
from rest_framework.authtoken.models import Token
from django.shortcuts import get_object_or_404
from django.db.models import Q
from django.utils import timezone
from django.utils.dateparse import parse_datetime
from .models import ContactUs, Composition, Campaign, CompositionImage
@@ -13,7 +12,6 @@ from .serializers import (
ContactUsSerializer,
CompositionSerializer,
CampaignSerializer,
CompositionImageSerializer,
AdminLoginSerializer,
)
@@ -43,7 +41,7 @@ def admin_login(request):
@api_view(['POST'])
@permission_classes([IsAuthenticated])
@permission_classes([IsAdminUser])
def admin_logout(request):
"""Delete the current admin auth token (logout)."""
Token.objects.filter(user=request.user).delete()
@@ -51,7 +49,7 @@ def admin_logout(request):
@api_view(['GET'])
@permission_classes([IsAuthenticated])
@permission_classes([IsAdminUser])
def admin_me(request):
"""Return the currently authenticated admin user."""
user = request.user
@@ -66,24 +64,47 @@ def admin_me(request):
class ContactUsViewSet(viewsets.ModelViewSet):
"""
ViewSet for ContactUs model.
Provides CRUD operations for contact form submissions.
Contact Us permissions:
- Public: create a message, and view own messages via /mine/
- Admin: list all, retrieve, update (including admin_response), delete
"""
queryset = ContactUs.objects.all()
serializer_class = ContactUsSerializer
permission_classes = [AllowAny] # Allow anyone to submit contact forms
permission_classes = [IsAdminUser]
def get_permissions(self):
"""
Override to allow GET (list, retrieve) and POST (create) for anyone.
"""
if self.action in ['list', 'retrieve', 'create', 'by_category']:
if self.action in ['create', 'mine']:
return [AllowAny()]
return [IsAuthenticated()]
return [IsAdminUser()]
def get_serializer(self, *args, **kwargs):
serializer = super().get_serializer(*args, **kwargs)
# Public create cannot set admin_response
if self.action == 'create' and not (
self.request.user and self.request.user.is_staff
):
serializer.fields['admin_response'].read_only = True
return serializer
@action(detail=False, methods=['get'])
def mine(self, request):
"""
Public: list contacts for a given email_or_phone (own submissions),
including admin_response.
"""
email_or_phone = (request.query_params.get('email_or_phone') or '').strip()
if not email_or_phone:
return Response(
{'error': 'email_or_phone query parameter is required.'},
status=status.HTTP_400_BAD_REQUEST,
)
contacts = self.queryset.filter(email_or_phone__iexact=email_or_phone)
serializer = self.get_serializer(contacts, many=True)
return Response(serializer.data)
@action(detail=False, methods=['get'])
def by_category(self, request):
"""Get contacts filtered by category."""
"""Admin: get contacts filtered by category."""
category = request.query_params.get('category', None)
if category:
contacts = self.queryset.filter(category=category)
@@ -97,35 +118,28 @@ class ContactUsViewSet(viewsets.ModelViewSet):
class CompositionViewSet(viewsets.ModelViewSet):
"""
ViewSet for Composition model.
Supports uploading one or more images at create/update time via the
multipart ``uploaded_images`` field, with an optional ``main_image_index``
to flag the main image. Extra actions allow adding images, choosing the
main image, and deleting an image after creation.
Compositions (مقالات):
- Public: list, retrieve, by-created-at
- Admin: create, update, delete, manage images
"""
queryset = Composition.objects.prefetch_related('images').all()
serializer_class = CompositionSerializer
permission_classes = [AllowAny] # Public read access
permission_classes = [IsAdminUser]
parser_classes = [MultiPartParser, FormParser, JSONParser]
def get_permissions(self):
"""
Override to allow GET (list, retrieve) and POST (create) for anyone.
"""
if self.action in ['list', 'retrieve', 'create', 'by_created_at']:
if self.action in ['list', 'retrieve', 'by_created_at']:
return [AllowAny()]
return [IsAuthenticated()]
return [IsAdminUser()]
def get_serializer_context(self):
"""Add request to serializer context for image URL generation."""
context = super().get_serializer_context()
context['request'] = self.request
return context
@action(detail=False, methods=['get'], url_path='by-created-at')
def by_created_at(self, request):
"""Get compositions filtered by created_at date range."""
"""Public: compositions filtered by created_at date range."""
from_param = request.query_params.get('from')
to_param = request.query_params.get('to')
@@ -161,17 +175,16 @@ class CompositionViewSet(viewsets.ModelViewSet):
serializer = self.get_serializer(compositions, many=True)
return Response(serializer.data)
def _composition_response(self, composition):
"""Return a fresh composition payload with up-to-date images."""
composition = Composition.objects.prefetch_related('images').get(
pk=composition.pk
)
return self.get_serializer(composition).data
@action(detail=True, methods=['post'], url_path='add-images')
def add_images(self, request, pk=None):
"""Add one or more images to an existing composition."""
"""Admin: add one or more images to an existing composition."""
composition = self.get_object()
serializer = self.get_serializer(
composition, data=request.data, partial=True
@@ -179,10 +192,10 @@ class CompositionViewSet(viewsets.ModelViewSet):
serializer.is_valid(raise_exception=True)
composition = serializer.save()
return Response(self._composition_response(composition))
@action(detail=True, methods=['post'], url_path='set-main-image')
def set_main_image(self, request, pk=None):
"""Flag one of the composition's images as the main image."""
"""Admin: flag one image as the main image."""
composition = self.get_object()
image_id = request.data.get('image_id')
if image_id is None:
@@ -194,16 +207,16 @@ class CompositionViewSet(viewsets.ModelViewSet):
CompositionImage, pk=image_id, composition=composition
)
image.is_main = True
image.save() # model.save() unsets is_main on the other images
image.save()
return Response(self._composition_response(composition))
@action(
detail=True,
methods=['delete'],
url_path='images/(?P<image_id>[^/.]+)'
)
def delete_image(self, request, pk=None, image_id=None):
"""Delete a single image from the composition."""
"""Admin: delete a single image from the composition."""
composition = self.get_object()
image = get_object_or_404(
CompositionImage, pk=image_id, composition=composition
@@ -214,30 +227,26 @@ class CompositionViewSet(viewsets.ModelViewSet):
class CampaignViewSet(viewsets.ModelViewSet):
"""
ViewSet for Campaign model.
Provides CRUD operations for campaigns.
Campaigns:
- Public: list, retrieve, active, upcoming, ended
- Admin: create, update, delete
"""
queryset = Campaign.objects.all()
serializer_class = CampaignSerializer
permission_classes = [AllowAny] # Public read access
permission_classes = [IsAdminUser]
def get_permissions(self):
"""
Override to allow GET (list, retrieve) and POST (create) for anyone.
"""
if self.action in ['list', 'retrieve', 'create', 'active', 'upcoming', 'ended']:
if self.action in ['list', 'retrieve', 'active', 'upcoming', 'ended']:
return [AllowAny()]
return [IsAuthenticated()]
return [IsAdminUser()]
def get_serializer_context(self):
"""Add request to serializer context for image URL generation."""
context = super().get_serializer_context()
context['request'] = self.request
return context
@action(detail=False, methods=['get'])
def active(self, request):
"""Get all currently active campaigns."""
now = timezone.now()
active_campaigns = self.queryset.filter(
start_time__lte=now,
@@ -245,18 +254,16 @@ class CampaignViewSet(viewsets.ModelViewSet):
)
serializer = self.get_serializer(active_campaigns, many=True)
return Response(serializer.data)
@action(detail=False, methods=['get'])
def upcoming(self, request):
"""Get all upcoming campaigns."""
now = timezone.now()
upcoming_campaigns = self.queryset.filter(start_time__gt=now)
serializer = self.get_serializer(upcoming_campaigns, many=True)
return Response(serializer.data)
@action(detail=False, methods=['get'])
def ended(self, request):
"""Get all ended campaigns."""
now = timezone.now()
ended_campaigns = self.queryset.filter(end_time__lt=now)
serializer = self.get_serializer(ended_campaigns, many=True)