mirror of
https://github.com/seanmorley15/AdventureLog.git
synced 2025-08-02 19:55:18 +02:00
feat: Enhance collection sharing and location management features
- Implemented unsharing functionality in CollectionViewSet, including removal of user-owned locations from collections. - Refactored ContentImageViewSet to support multiple content types and improved permission checks for image uploads. - Added user ownership checks in LocationViewSet for delete operations. - Enhanced collection management in the frontend to display both owned and shared collections separately. - Updated Immich integration to handle access control based on location visibility and user permissions. - Improved UI components to show creator information and manage collection links more effectively. - Added loading states and error handling in collection fetching logic.
This commit is contained in:
Sean Morley
parent
7f80dad94b
commit
ba162175fe
19 changed files with 641 additions and 245 deletions
|
|
@ -2,51 +2,97 @@ from rest_framework import viewsets, status
|
|||
from rest_framework.decorators import action
|
||||
from rest_framework.permissions import IsAuthenticated
|
||||
from rest_framework.response import Response
|
||||
from django.contrib.contenttypes.models import ContentType
|
||||
from adventures.models import Location, ContentAttachment
|
||||
from adventures.serializers import AttachmentSerializer
|
||||
from adventures.permissions import IsOwnerOrSharedWithFullAccess
|
||||
|
||||
|
||||
class AttachmentViewSet(viewsets.ModelViewSet):
|
||||
serializer_class = AttachmentSerializer
|
||||
permission_classes = [IsAuthenticated]
|
||||
permission_classes = [IsOwnerOrSharedWithFullAccess]
|
||||
|
||||
def get_queryset(self):
|
||||
return ContentAttachment.objects.filter(user=self.request.user)
|
||||
|
||||
|
||||
def create(self, request, *args, **kwargs):
|
||||
if not request.user.is_authenticated:
|
||||
return Response({"error": "User is not authenticated"}, status=status.HTTP_401_UNAUTHORIZED)
|
||||
location_id = request.data.get('location')
|
||||
try:
|
||||
location = Location.objects.get(id=location_id)
|
||||
except Location.DoesNotExist:
|
||||
return Response({"error": "Location not found"}, status=status.HTTP_404_NOT_FOUND)
|
||||
|
||||
if location.user != request.user:
|
||||
# Check if the location has any collections
|
||||
if location.collections.exists():
|
||||
# Check if the user is in the shared_with list of any of the location's collections
|
||||
user_has_access = False
|
||||
for collection in location.collections.all():
|
||||
if collection.shared_with.filter(id=request.user.id).exists():
|
||||
user_has_access = True
|
||||
break
|
||||
|
||||
if not user_has_access:
|
||||
return Response({"error": "User does not have permission to access this location"}, status=status.HTTP_403_FORBIDDEN)
|
||||
else:
|
||||
return Response({"error": "User does not own this location"}, status=status.HTTP_403_FORBIDDEN)
|
||||
|
||||
return super().create(request, *args, **kwargs)
|
||||
|
||||
def perform_create(self, serializer):
|
||||
location_id = self.request.data.get('location')
|
||||
location = Location.objects.get(id=location_id)
|
||||
# Get the content object details
|
||||
content_type_id = request.data.get('content_type')
|
||||
object_id = request.data.get('object_id')
|
||||
|
||||
# For backward compatibility, also check for 'location' parameter
|
||||
location_id = request.data.get('location')
|
||||
|
||||
if location_id and not (content_type_id and object_id):
|
||||
# Handle legacy location-specific requests
|
||||
try:
|
||||
location = Location.objects.get(id=location_id)
|
||||
content_type = ContentType.objects.get_for_model(Location)
|
||||
content_type_id = content_type.id
|
||||
object_id = location_id
|
||||
except Location.DoesNotExist:
|
||||
return Response({"error": "Location not found"}, status=status.HTTP_404_NOT_FOUND)
|
||||
|
||||
if not (content_type_id and object_id):
|
||||
return Response({"error": "content_type and object_id are required"}, status=status.HTTP_400_BAD_REQUEST)
|
||||
|
||||
# If the location belongs to collections, set the owner to the collection owner
|
||||
if location.collections.exists():
|
||||
# Get the first collection's owner (assuming all collections have the same owner)
|
||||
collection = location.collections.first()
|
||||
serializer.save(user=collection.user)
|
||||
else:
|
||||
# Otherwise, set the owner to the request user
|
||||
serializer.save(user=self.request.user)
|
||||
try:
|
||||
content_type = ContentType.objects.get(id=content_type_id)
|
||||
model_class = content_type.model_class()
|
||||
content_object = model_class.objects.get(id=object_id)
|
||||
except (ContentType.DoesNotExist, model_class.DoesNotExist):
|
||||
return Response({"error": "Content object not found"}, status=status.HTTP_404_NOT_FOUND)
|
||||
|
||||
return super().create(request, *args, **kwargs)
|
||||
|
||||
def perform_create(self, serializer):
|
||||
content_type_id = self.request.data.get('content_type')
|
||||
object_id = self.request.data.get('object_id')
|
||||
|
||||
# Handle legacy location parameter
|
||||
location_id = self.request.data.get('location')
|
||||
if location_id and not (content_type_id and object_id):
|
||||
content_type = ContentType.objects.get_for_model(Location)
|
||||
content_type_id = content_type.id
|
||||
object_id = location_id
|
||||
|
||||
content_type = ContentType.objects.get(id=content_type_id)
|
||||
model_class = content_type.model_class()
|
||||
content_object = model_class.objects.get(id=object_id)
|
||||
|
||||
# Determine the appropriate user to assign
|
||||
attachment_user = self._get_attachment_user(content_object)
|
||||
|
||||
serializer.save(
|
||||
user=attachment_user,
|
||||
content_type=content_type,
|
||||
object_id=object_id
|
||||
)
|
||||
|
||||
def _get_attachment_user(self, content_object):
|
||||
"""
|
||||
Determine which user should own the attachment based on the content object.
|
||||
This preserves the original logic for shared collections.
|
||||
"""
|
||||
# Handle Location objects
|
||||
if isinstance(content_object, Location):
|
||||
if content_object.collections.exists():
|
||||
# Get the first collection's owner (assuming all collections have the same owner)
|
||||
collection = content_object.collections.first()
|
||||
return collection.user
|
||||
else:
|
||||
return self.request.user
|
||||
|
||||
# Handle other content types with collections
|
||||
elif hasattr(content_object, 'collection') and content_object.collection:
|
||||
return content_object.collection.user
|
||||
|
||||
# Handle content objects with a user field
|
||||
elif hasattr(content_object, 'user'):
|
||||
return content_object.user
|
||||
|
||||
# Default to request user
|
||||
return self.request.user
|
||||
|
|
@ -183,9 +183,12 @@ class CollectionViewSet(viewsets.ModelViewSet):
|
|||
def unshare(self, request, pk=None, uuid=None):
|
||||
if not request.user.is_authenticated:
|
||||
return Response({"error": "User is not authenticated"}, status=400)
|
||||
|
||||
collection = self.get_object()
|
||||
|
||||
if not uuid:
|
||||
return Response({"error": "User UUID is required"}, status=400)
|
||||
|
||||
try:
|
||||
user = User.objects.get(uuid=uuid, public_profile=True)
|
||||
except User.DoesNotExist:
|
||||
|
|
@ -197,9 +200,25 @@ class CollectionViewSet(viewsets.ModelViewSet):
|
|||
if not collection.shared_with.filter(id=user.id).exists():
|
||||
return Response({"error": "Collection is not shared with this user"}, status=400)
|
||||
|
||||
# Remove user from shared_with
|
||||
collection.shared_with.remove(user)
|
||||
|
||||
# Handle locations owned by the unshared user that are in this collection
|
||||
# These locations should be removed from the collection since they lose access
|
||||
locations_to_remove = collection.locations.filter(user=user)
|
||||
removed_count = locations_to_remove.count()
|
||||
|
||||
if locations_to_remove.exists():
|
||||
# Remove these locations from the collection
|
||||
collection.locations.remove(*locations_to_remove)
|
||||
|
||||
collection.save()
|
||||
return Response({"success": f"Unshared with {user.username}"})
|
||||
|
||||
success_message = f"Unshared with {user.username}"
|
||||
if removed_count > 0:
|
||||
success_message += f" and removed {removed_count} location(s) they owned from the collection"
|
||||
|
||||
return Response({"success": success_message})
|
||||
|
||||
def get_queryset(self):
|
||||
if self.action == 'destroy':
|
||||
|
|
|
|||
|
|
@ -4,13 +4,14 @@ from rest_framework.permissions import IsAuthenticated
|
|||
from rest_framework.response import Response
|
||||
from django.db.models import Q
|
||||
from django.core.files.base import ContentFile
|
||||
from adventures.models import Location, ContentImage
|
||||
from django.contrib.contenttypes.models import ContentType
|
||||
from adventures.models import Location, Transportation, Note, Lodging, Visit, ContentImage
|
||||
from adventures.serializers import ContentImageSerializer
|
||||
from integrations.models import ImmichIntegration
|
||||
import uuid
|
||||
import requests
|
||||
|
||||
class AdventureImageViewSet(viewsets.ModelViewSet):
|
||||
class ContentImageViewSet(viewsets.ModelViewSet):
|
||||
serializer_class = ContentImageSerializer
|
||||
permission_classes = [IsAuthenticated]
|
||||
|
||||
|
|
@ -20,23 +21,28 @@ class AdventureImageViewSet(viewsets.ModelViewSet):
|
|||
|
||||
@action(detail=True, methods=['post'])
|
||||
def toggle_primary(self, request, *args, **kwargs):
|
||||
# Makes the image the primary image for the location, if there is already a primary image linked to the location, it is set to false and the new image is set to true. make sure that the permission is set to the owner of the location
|
||||
if not request.user.is_authenticated:
|
||||
return Response({"error": "User is not authenticated"}, status=status.HTTP_401_UNAUTHORIZED)
|
||||
|
||||
instance = self.get_object()
|
||||
location = instance.location
|
||||
if location.user != request.user:
|
||||
return Response({"error": "User does not own this location"}, status=status.HTTP_403_FORBIDDEN)
|
||||
content_object = instance.content_object
|
||||
|
||||
# Check ownership based on content type
|
||||
if hasattr(content_object, 'user') and content_object.user != request.user:
|
||||
return Response({"error": "User does not own this content"}, status=status.HTTP_403_FORBIDDEN)
|
||||
|
||||
# Check if the image is already the primary image
|
||||
if instance.is_primary:
|
||||
return Response({"error": "Image is already the primary image"}, status=status.HTTP_400_BAD_REQUEST)
|
||||
|
||||
# Set the current primary image to false
|
||||
ContentImage.objects.filter(location=location, is_primary=True).update(is_primary=False)
|
||||
# Set other images of the same content object to not primary
|
||||
ContentImage.objects.filter(
|
||||
content_type=instance.content_type,
|
||||
object_id=instance.object_id,
|
||||
is_primary=True
|
||||
).update(is_primary=False)
|
||||
|
||||
# Set the new image to true
|
||||
# Set the new image to primary
|
||||
instance.is_primary = True
|
||||
instance.save()
|
||||
return Response({"success": "Image set as primary image"})
|
||||
|
|
@ -44,29 +50,67 @@ class AdventureImageViewSet(viewsets.ModelViewSet):
|
|||
def create(self, request, *args, **kwargs):
|
||||
if not request.user.is_authenticated:
|
||||
return Response({"error": "User is not authenticated"}, status=status.HTTP_401_UNAUTHORIZED)
|
||||
location_id = request.data.get('location')
|
||||
try:
|
||||
location = Location.objects.get(id=location_id)
|
||||
except Location.DoesNotExist:
|
||||
return Response({"error": "location not found"}, status=status.HTTP_404_NOT_FOUND)
|
||||
|
||||
if location.user != request.user:
|
||||
# Check if the location has any collections
|
||||
if location.collections.exists():
|
||||
# Check if the user is in the shared_with list of any of the location's collections
|
||||
user_has_access = False
|
||||
for collection in location.collections.all():
|
||||
if collection.shared_with.filter(id=request.user.id).exists():
|
||||
user_has_access = True
|
||||
break
|
||||
|
||||
if not user_has_access:
|
||||
return Response({"error": "User does not have permission to access this location"}, status=status.HTTP_403_FORBIDDEN)
|
||||
else:
|
||||
return Response({"error": "User does not own this location"}, status=status.HTTP_403_FORBIDDEN)
|
||||
# Get content type and object ID from request
|
||||
content_type_name = request.data.get('content_type')
|
||||
object_id = request.data.get('object_id')
|
||||
|
||||
if not content_type_name or not object_id:
|
||||
return Response({
|
||||
"error": "content_type and object_id are required"
|
||||
}, status=status.HTTP_400_BAD_REQUEST)
|
||||
|
||||
# Map content type names to model classes
|
||||
content_type_map = {
|
||||
'location': Location,
|
||||
'transportation': Transportation,
|
||||
'note': Note,
|
||||
'lodging': Lodging,
|
||||
'visit': Visit,
|
||||
}
|
||||
|
||||
if content_type_name not in content_type_map:
|
||||
return Response({
|
||||
"error": f"Invalid content_type. Must be one of: {', '.join(content_type_map.keys())}"
|
||||
}, status=status.HTTP_400_BAD_REQUEST)
|
||||
|
||||
# Get the content type and object
|
||||
try:
|
||||
content_type = ContentType.objects.get_for_model(content_type_map[content_type_name])
|
||||
content_object = content_type_map[content_type_name].objects.get(id=object_id)
|
||||
except (ValueError, content_type_map[content_type_name].DoesNotExist):
|
||||
return Response({
|
||||
"error": f"{content_type_name} not found"
|
||||
}, status=status.HTTP_404_NOT_FOUND)
|
||||
|
||||
# Check permissions based on content type
|
||||
if hasattr(content_object, 'user'):
|
||||
if content_object.user != request.user:
|
||||
# For Location, check if user has shared access
|
||||
if content_type_name == 'location':
|
||||
if content_object.collections.exists():
|
||||
user_has_access = False
|
||||
for collection in content_object.collections.all():
|
||||
if collection.shared_with.filter(id=request.user.id).exists() or collection.user == request.user:
|
||||
user_has_access = True
|
||||
break
|
||||
|
||||
if not user_has_access:
|
||||
return Response({
|
||||
"error": "User does not have permission to access this content"
|
||||
}, status=status.HTTP_403_FORBIDDEN)
|
||||
else:
|
||||
return Response({
|
||||
"error": "User does not own this content"
|
||||
}, status=status.HTTP_403_FORBIDDEN)
|
||||
else:
|
||||
return Response({
|
||||
"error": "User does not own this content"
|
||||
}, status=status.HTTP_403_FORBIDDEN)
|
||||
|
||||
# Handle Immich ID for shared users by downloading the image
|
||||
if (request.user != location.user and
|
||||
if (hasattr(content_object, 'user') and
|
||||
request.user != content_object.user and
|
||||
'immich_id' in request.data and
|
||||
request.data.get('immich_id')):
|
||||
|
||||
|
|
@ -91,8 +135,8 @@ class AdventureImageViewSet(viewsets.ModelViewSet):
|
|||
immich_response.raise_for_status()
|
||||
|
||||
# Create a temporary file with the downloaded content
|
||||
content_type = immich_response.headers.get('Content-Type', 'image/jpeg')
|
||||
if not content_type.startswith('image/'):
|
||||
content_type_header = immich_response.headers.get('Content-Type', 'image/jpeg')
|
||||
if not content_type_header.startswith('image/'):
|
||||
return Response({
|
||||
"error": "Invalid content type returned from Immich server.",
|
||||
"code": "invalid_content_type"
|
||||
|
|
@ -105,7 +149,7 @@ class AdventureImageViewSet(viewsets.ModelViewSet):
|
|||
'image/webp': '.webp',
|
||||
'image/gif': '.gif'
|
||||
}
|
||||
file_ext = ext_map.get(content_type, '.jpg')
|
||||
file_ext = ext_map.get(content_type_header, '.jpg')
|
||||
filename = f"immich_{immich_id}{file_ext}"
|
||||
|
||||
# Create a Django ContentFile from the downloaded image
|
||||
|
|
@ -115,14 +159,20 @@ class AdventureImageViewSet(viewsets.ModelViewSet):
|
|||
request_data = request.data.copy()
|
||||
request_data.pop('immich_id', None) # Remove immich_id
|
||||
request_data['image'] = image_file # Add the image file
|
||||
request_data['content_type'] = content_type.id
|
||||
request_data['object_id'] = object_id
|
||||
|
||||
# Create the serializer with the modified data
|
||||
serializer = self.get_serializer(data=request_data)
|
||||
serializer.is_valid(raise_exception=True)
|
||||
|
||||
# Save with the downloaded image
|
||||
location = serializer.validated_data['location']
|
||||
serializer.save(user=location.user, image=image_file)
|
||||
serializer.save(
|
||||
user=content_object.user if hasattr(content_object, 'user') else request.user,
|
||||
image=image_file,
|
||||
content_type=content_type,
|
||||
object_id=object_id
|
||||
)
|
||||
|
||||
return Response(serializer.data, status=status.HTTP_201_CREATED)
|
||||
|
||||
|
|
@ -137,34 +187,47 @@ class AdventureImageViewSet(viewsets.ModelViewSet):
|
|||
"code": "immich_processing_error"
|
||||
}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
|
||||
|
||||
return super().create(request, *args, **kwargs)
|
||||
# Add content type and object ID to request data
|
||||
request_data = request.data.copy()
|
||||
request_data['content_type'] = content_type.id
|
||||
request_data['object_id'] = object_id
|
||||
|
||||
# Create serializer with modified data
|
||||
serializer = self.get_serializer(data=request_data)
|
||||
serializer.is_valid(raise_exception=True)
|
||||
|
||||
# Save the image
|
||||
serializer.save(
|
||||
user=content_object.user if hasattr(content_object, 'user') else request.user,
|
||||
content_type=content_type,
|
||||
object_id=object_id
|
||||
)
|
||||
|
||||
return Response(serializer.data, status=status.HTTP_201_CREATED)
|
||||
|
||||
def update(self, request, *args, **kwargs):
|
||||
if not request.user.is_authenticated:
|
||||
return Response({"error": "User is not authenticated"}, status=status.HTTP_401_UNAUTHORIZED)
|
||||
|
||||
location_id = request.data.get('location')
|
||||
try:
|
||||
location = Location.objects.get(id=location_id)
|
||||
except Location.DoesNotExist:
|
||||
return Response({"error": "location not found"}, status=status.HTTP_404_NOT_FOUND)
|
||||
instance = self.get_object()
|
||||
content_object = instance.content_object
|
||||
|
||||
if location.user != request.user:
|
||||
return Response({"error": "User does not own this location"}, status=status.HTTP_403_FORBIDDEN)
|
||||
# Check ownership
|
||||
if hasattr(content_object, 'user') and content_object.user != request.user:
|
||||
return Response({"error": "User does not own this content"}, status=status.HTTP_403_FORBIDDEN)
|
||||
|
||||
return super().update(request, *args, **kwargs)
|
||||
|
||||
def perform_destroy(self, instance):
|
||||
return super().perform_destroy(instance)
|
||||
|
||||
def destroy(self, request, *args, **kwargs):
|
||||
if not request.user.is_authenticated:
|
||||
return Response({"error": "User is not authenticated"}, status=status.HTTP_401_UNAUTHORIZED)
|
||||
|
||||
instance = self.get_object()
|
||||
location = instance.location
|
||||
if location.user != request.user:
|
||||
return Response({"error": "User does not own this location"}, status=status.HTTP_403_FORBIDDEN)
|
||||
content_object = instance.content_object
|
||||
|
||||
# Check ownership
|
||||
if hasattr(content_object, 'user') and content_object.user != request.user:
|
||||
return Response({"error": "User does not own this content"}, status=status.HTTP_403_FORBIDDEN)
|
||||
|
||||
return super().destroy(request, *args, **kwargs)
|
||||
|
||||
|
|
@ -173,41 +236,33 @@ class AdventureImageViewSet(viewsets.ModelViewSet):
|
|||
return Response({"error": "User is not authenticated"}, status=status.HTTP_401_UNAUTHORIZED)
|
||||
|
||||
instance = self.get_object()
|
||||
location = instance.location
|
||||
if location.user != request.user:
|
||||
return Response({"error": "User does not own this location"}, status=status.HTTP_403_FORBIDDEN)
|
||||
content_object = instance.content_object
|
||||
|
||||
# Check ownership
|
||||
if hasattr(content_object, 'user') and content_object.user != request.user:
|
||||
return Response({"error": "User does not own this content"}, status=status.HTTP_403_FORBIDDEN)
|
||||
|
||||
return super().partial_update(request, *args, **kwargs)
|
||||
|
||||
@action(detail=False, methods=['GET'], url_path='(?P<location_id>[0-9a-f-]+)')
|
||||
def location_images(self, request, location_id=None, *args, **kwargs):
|
||||
if not request.user.is_authenticated:
|
||||
return Response({"error": "User is not authenticated"}, status=status.HTTP_401_UNAUTHORIZED)
|
||||
|
||||
try:
|
||||
location_uuid = uuid.UUID(location_id)
|
||||
except ValueError:
|
||||
return Response({"error": "Invalid location ID"}, status=status.HTTP_400_BAD_REQUEST)
|
||||
|
||||
# Updated queryset to include images from locations the user owns OR has shared access to
|
||||
queryset = ContentImage.objects.filter(
|
||||
Q(location__id=location_uuid) & (
|
||||
Q(location__user=request.user) | # User owns the location
|
||||
Q(location__collections__shared_with=request.user) # User has shared access via collection
|
||||
)
|
||||
).distinct()
|
||||
|
||||
serializer = self.get_serializer(queryset, many=True, context={'request': request})
|
||||
return Response(serializer.data)
|
||||
|
||||
|
||||
def get_queryset(self):
|
||||
# Updated to include images from locations the user owns OR has shared access to
|
||||
return ContentImage.objects.filter(
|
||||
Q(location__user=self.request.user) | # User owns the location
|
||||
Q(location__collections__shared_with=self.request.user) # User has shared access via collection
|
||||
"""Get all images the user has access to"""
|
||||
if not self.request.user.is_authenticated:
|
||||
return ContentImage.objects.none()
|
||||
|
||||
# Get content type for Location to handle shared access
|
||||
location_content_type = ContentType.objects.get_for_model(Location)
|
||||
|
||||
# Build queryset with proper permissions
|
||||
queryset = ContentImage.objects.filter(
|
||||
Q(content_object__user=self.request.user) | # User owns the content
|
||||
Q(content_type=location_content_type, content_object__collections__shared_with=self.request.user) # Shared locations
|
||||
).distinct()
|
||||
|
||||
return queryset
|
||||
|
||||
def perform_create(self, serializer):
|
||||
# Always set the image owner to the location owner, not the current user
|
||||
location = serializer.validated_data['location']
|
||||
serializer.save(user=location.user)
|
||||
# The content_type and object_id are already set in the create method
|
||||
# Just ensure the user is set correctly
|
||||
pass
|
||||
|
|
@ -141,6 +141,16 @@ class LocationViewSet(viewsets.ModelViewSet):
|
|||
|
||||
self.perform_update(serializer)
|
||||
return Response(serializer.data)
|
||||
|
||||
def destroy(self, request, *args, **kwargs):
|
||||
"""Only allow the owner to delete a location."""
|
||||
instance = self.get_object()
|
||||
|
||||
# Check if the user is the owner
|
||||
if instance.user != request.user:
|
||||
raise PermissionDenied("Only the owner can delete this location.")
|
||||
|
||||
return super().destroy(request, *args, **kwargs)
|
||||
|
||||
# ==================== CUSTOM ACTIONS ====================
|
||||
|
||||
|
|
@ -221,37 +231,50 @@ class LocationViewSet(viewsets.ModelViewSet):
|
|||
|
||||
# ==================== HELPER METHODS ====================
|
||||
|
||||
def _validate_collection_permissions(self, collections):
|
||||
"""Validate user has permission to use all provided collections. Only the owner or shared users can use collections."""
|
||||
for collection in collections:
|
||||
if not (collection.user == self.request.user or
|
||||
collection.shared_with.filter(uuid=self.request.user.uuid).exists()):
|
||||
raise PermissionDenied(
|
||||
f"You do not have permission to use collection '{collection.name}'."
|
||||
)
|
||||
|
||||
def _validate_collection_update_permissions(self, instance, new_collections):
|
||||
"""Validate permissions for collection updates (add/remove)."""
|
||||
# Check permissions for new collections being added
|
||||
for collection in new_collections:
|
||||
if (collection.user != self.request.user and
|
||||
not collection.shared_with.filter(uuid=self.request.user.uuid).exists()):
|
||||
raise PermissionDenied(
|
||||
f"You do not have permission to use collection '{collection.name}'."
|
||||
)
|
||||
|
||||
# Check permissions for collections being removed
|
||||
"""Validate collection permissions for updates, allowing collection owners to unlink locations."""
|
||||
current_collections = set(instance.collections.all())
|
||||
new_collections_set = set(new_collections)
|
||||
|
||||
# Collections being added
|
||||
collections_to_add = new_collections_set - current_collections
|
||||
|
||||
# Collections being removed
|
||||
collections_to_remove = current_collections - new_collections_set
|
||||
|
||||
|
||||
# Validate permissions for collections being added
|
||||
for collection in collections_to_add:
|
||||
# Standard validation for adding collections
|
||||
if collection.user != self.request.user:
|
||||
# Check if user has shared access to the collection
|
||||
if not collection.shared_with.filter(uuid=self.request.user.uuid).exists():
|
||||
raise PermissionDenied(
|
||||
f"You don't have permission to add location to collection '{collection.name}'"
|
||||
)
|
||||
|
||||
# For collections being removed, allow if:
|
||||
# 1. User owns the location, OR
|
||||
# 2. User owns the collection (even if they don't own the location)
|
||||
for collection in collections_to_remove:
|
||||
if (collection.user != self.request.user and
|
||||
not collection.shared_with.filter(uuid=self.request.user.uuid).exists()):
|
||||
raise PermissionDenied(
|
||||
f"You cannot remove the adventure from collection '{collection.name}' "
|
||||
f"as you don't have permission."
|
||||
)
|
||||
user_owns_location = instance.user == self.request.user
|
||||
user_owns_collection = collection.user == self.request.user
|
||||
|
||||
if not (user_owns_location or user_owns_collection):
|
||||
# Check if user has shared access to the collection
|
||||
if not collection.shared_with.filter(uuid=self.request.user.uuid).exists():
|
||||
raise PermissionDenied(
|
||||
f"You don't have permission to remove location from collection '{collection.name}'"
|
||||
)
|
||||
|
||||
def _validate_collection_permissions(self, collections):
|
||||
"""Validate permissions for all collections (used in create)."""
|
||||
for collection in collections:
|
||||
if collection.user != self.request.user:
|
||||
# Check if user has shared access to the collection
|
||||
if not collection.shared_with.filter(uuid=self.request.user.uuid).exists():
|
||||
raise PermissionDenied(
|
||||
f"You don't have permission to add location to collection '{collection.name}'"
|
||||
)
|
||||
|
||||
def _apply_visit_filtering(self, queryset, request):
|
||||
"""Apply visit status filtering to queryset."""
|
||||
|
|
@ -289,7 +312,7 @@ class LocationViewSet(viewsets.ModelViewSet):
|
|||
# Check shared collection access
|
||||
if user.is_authenticated:
|
||||
for collection in adventure.collections.all():
|
||||
if collection.shared_with.filter(uuid=user.uuid).exists():
|
||||
if collection.shared_with.filter(uuid=user.uuid).exists() or collection.user == user:
|
||||
return True
|
||||
|
||||
return False
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue