extensions-website/extensions/views/api.py

import logging

from rest_framework.permissions import AllowAny
from rest_framework.response import Response
from rest_framework import serializers, status
from rest_framework.views import APIView
from rest_framework.permissions import IsAuthenticated
from drf_spectacular.utils import OpenApiParameter, extend_schema
from django.core.exceptions import ValidationError
from django.db import transaction

from common.compare import is_in_version_range, version
from extensions.models import Extension, Platform, Version
from extensions.utils import clean_json_dictionary_from_optional_fields
from extensions.views.manage import NewVersionView
from files.forms import FileFormSkipAgreed


from constants.base import (
    EXTENSION_TYPE_SLUGS_SINGULAR,
)

log = logging.getLogger(__name__)


class ListedExtensionsSerializer(serializers.ModelSerializer):
    error_messages = {
        "invalid_blender_version": "Invalid blender_version: use full semantic versioning like "
        "4.2.0.",
    }

    class Meta:
        model = Extension
        fields = ()

    UNKNOWN_PLATFORM = 'unknown-platform-value'

    def __init__(self, *args, **kwargs):
        self.request = kwargs.pop('request', None)
        self.blender_version = kwargs.pop('blender_version', None)
        self.platform = kwargs.pop('platform', None)
        self._validate()
        super().__init__(*args, **kwargs)

    def _validate(self):
        if self.blender_version:
            try:
                version(self.blender_version)
            except ValidationError:
                self.fail('invalid_blender_version')
        if self.platform:
            # FIXME change to an in-memory lookup?
            try:
                Platform.objects.get(slug=self.platform)
            except Platform.DoesNotExist:
                self.platform = self.UNKNOWN_PLATFORM

    def to_representation(self, instance):
        matching_version = None
        # avoid triggering additional db queries, reuse the prefetched queryset
        versions = [
            v
            for v in instance.versions.all()
            if v.file and v.file.status in instance.valid_file_statuses
        ]
        if not versions:
            return None
        versions = sorted(versions, key=lambda v: v.date_created, reverse=True)
        for v in versions:
            if self.blender_version and not is_in_version_range(
                self.blender_version,
                v.blender_version_min,
                v.blender_version_max,
            ):
                continue
            platform_slugs = set(p.slug for p in v.platforms.all())
            # empty platforms field matches any platform filter
            # UNKNOWN_PLATFORM matches only empty platforms field
            if self.platform and (platform_slugs and self.platform not in platform_slugs):
                continue
            matching_version = v
            break

        if not matching_version:
            return None

        data = {
            'id': instance.extension_id,
            'schema_version': matching_version.schema_version,
            'name': instance.name,
            'version': matching_version.version,
            'tagline': matching_version.tagline,
            'archive_hash': matching_version.file.original_hash,
            'archive_size': matching_version.file.size_bytes,
            'archive_url': self.request.build_absolute_uri(
                matching_version.download_url(append_repository=False)
            ),
            'type': EXTENSION_TYPE_SLUGS_SINGULAR.get(instance.type),
            'blender_version_min': matching_version.blender_version_min,
            'blender_version_max': matching_version.blender_version_max,
            'website': self.request.build_absolute_uri(instance.get_absolute_url()),
            # avoid triggering additional db queries, reuse the prefetched queryset
            'maintainer': instance.team and instance.team.name or str(instance.authors.all()[0]),
            'license': [license_iter.slug for license_iter in matching_version.licenses.all()],
            'permissions': {
                permission.slug: permission.help
                for permission in matching_version.permissions.all()
            },
            'platforms': [platform.slug for platform in matching_version.platforms.all()],
            # TODO: handle copyright
            'tags': [str(tag) for tag in matching_version.tags.all()],
        }

        return clean_json_dictionary_from_optional_fields(data)


class ExtensionsAPIView(APIView):
    permission_classes = [AllowAny]
    serializer_class = ListedExtensionsSerializer

    @extend_schema(
        parameters=[
            OpenApiParameter(
                name="blender_version",
                description=("Blender version to check for compatibility"),
                type=str,
            ),
            OpenApiParameter(
                name="platform",
                description=("Platform to check for compatibility"),
                type=str,
            ),
        ]
    )
    def get(self, request):
        blender_version = request.GET.get('blender_version')
        platform = request.GET.get('platform')
        qs = Extension.objects.listed.prefetch_related(
            'authors',
            'team',
            'versions',
            'versions__file',
            'versions__licenses',
            'versions__permissions',
            'versions__platforms',
            'versions__tags',
        ).all()
        serializer = self.serializer_class(
            qs,
            blender_version=blender_version,
            platform=platform,
            request=request,
            many=True,
        )
        data = [e for e in serializer.data if e is not None]
        return Response(
            {
                # TODO implement extension blocking by moderators
                'blocklist': [],
                'data': data,
                'version': 'v1',
            }
        )


class ExtensionVersionSerializer(serializers.Serializer):
    version_file = serializers.FileField()
    release_notes = serializers.CharField(max_length=1024, required=False)


class UploadExtensionVersionView(APIView):
    permission_classes = [IsAuthenticated]

    @extend_schema(
        request=ExtensionVersionSerializer,
        responses={201: 'Extension version uploaded successfully!'},
    )
    def post(self, request, extension_id, *args, **kwargs):
        serializer = ExtensionVersionSerializer(data=request.data)
        if not serializer.is_valid():
            return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)

        user = request.user
        version_file = serializer.validated_data['version_file']
        release_notes = serializer.validated_data.get('release_notes', '')

        extension = Extension.objects.filter(extension_id=extension_id).first()
        if not extension:
            return Response(
                {
                    'message': f'Extension "{extension_id}" not found',
                },
                status=status.HTTP_404_NOT_FOUND,
            )

        if not extension.has_maintainer(user):
            return Response(
                {
                    'message': f'Extension "{extension_id}" not maintained by user "{user}"',
                },
                status=status.HTTP_403_FORBIDDEN,
            )

        # Create a NewVersionView instance to handle file creation
        new_version_view = NewVersionView(request=request, extension=extension)

        # Pass the version_file to the form
        form = new_version_view.get_form(FileFormSkipAgreed)
        form.fields['source'].initial = version_file

        if not form.is_valid():
            return Response({'message': form.errors}, status=status.HTTP_400_BAD_REQUEST)

        with transaction.atomic():
            # Create the file instance
            file_instance = form.save(commit=False)
            file_instance.user = user
            file_instance.save()

            # Create the version from the file
            version = Version.objects.update_or_create(
                extension=extension,
                file=file_instance,
                release_notes=release_notes,
                **file_instance.parsed_version_fields,
            )[0]

        return Response(
            {
                'message': 'Extension version uploaded successfully!',
                'extension_id': extension_id,
                'version_file': version_file.name,
                'release_notes': version.release_notes,
            },
            status=status.HTTP_201_CREATED,
        )