"""Regular and resumable upload endpoints (disk-backed).""" import mimetypes import uuid from pathlib import Path from django.conf import settings from django.db import transaction from django.http import Http404 from rest_framework.decorators import api_view, parser_classes from rest_framework.parsers import MultiPartParser, FormParser from rest_framework.response import Response from .models import DriveItem, UploadSession from .services import error, item_info, target_parent, writable_destination, valid_name, readable, writable from .spaces import resolve_space, can_write_space from .permissions import has_space_right from .storage_layout import file_location, disk_path, assert_available, prepare_parent, NameConflict from .quotas import QuotaExceeded, locked_quota_tree, assert_space_capacity @api_view(['POST']) @parser_classes([MultiPartParser,FormParser]) def upload(request): try: space,parent=writable_destination(request.user,request.data.get('space'),request.data.get('parent')) except ValueError as ex: return error(str(ex)) except PermissionError as ex: return error(str(ex),403) incoming=request.FILES.getlist('files') if not incoming: return error('Aucun fichier envoyé.') # Uploads are streamed to a disk temp directory (not buffered in RAM). if any(f.size>settings.DRIVE_MAX_FILE_SIZE for f in incoming): return error('Un fichier dépasse la taille maximum configurée.',413) saved=[] try: with transaction.atomic(): with locked_quota_tree(space): # Include the sum of ALL files in the same request. The lock prevents # another simultaneous request from consuming these bytes meanwhile. assert_space_capacity(space, sum(f.size for f in incoming)) for f in incoming: name=valid_name(f.name) key=str(uuid.uuid4()) location=file_location(space,parent,name) dst=assert_available(space,parent,name) prepare_parent(space,parent) with open(dst,'xb') as out: saved.append(dst) # Cleanup even when a stream fails halfway. for block in f.chunks(chunk_size=2*1024*1024): out.write(block) mime=mimetypes.guess_type(name)[0] or 'application/octet-stream' node=DriveItem.objects.create(owner=request.user,space=space,parent=parent,name=name, storage_key=key,storage_location=location,size=f.size,mime_type=mime) saved[-1] = (dst,node) except (QuotaExceeded,ValueError,OSError) as ex: for entry in saved: path=entry[0] if isinstance(entry,tuple) else entry path.unlink(missing_ok=True) return error(str(ex),413) if isinstance(ex,QuotaExceeded) else error(str(ex),409) if isinstance(ex,(NameConflict,FileExistsError)) else error(f'Import impossible : {ex}') except Exception: for entry in saved: path=entry[0] if isinstance(entry,tuple) else entry path.unlink(missing_ok=True) raise return Response({'items':[item_info(n,request.user) for _,n in saved]},status=201) # Large uploads are chunked in 8 MiB pieces; a refresh does not restart the upload. @api_view(['POST']) def upload_start(request): try: name=valid_name(request.data.get('name')) size=int(request.data.get('size',-1)) space,parent=writable_destination(request.user,request.data.get('space'),request.data.get('parent')) except (TypeError,ValueError) as ex:return error(str(ex)) except PermissionError as ex:return error(str(ex),403) if size<0 or size>settings.DRIVE_MAX_FILE_SIZE:return error('Taille de fichier non autorisée.',413) try: assert_available(space,parent,name) except NameConflict as ex: return error(str(ex),409) with transaction.atomic(): with locked_quota_tree(space): if UploadSession.objects.filter(owner=request.user).count()>=20: return error('Trop d’imports en cours (20 maximum).',429) try:assert_space_capacity(space,size) except QuotaExceeded as ex:return error(str(ex),413) job=UploadSession.objects.create(owner=request.user,space=space,parent=parent,name=name,total_size=size) return Response({'id':str(job.pk),'offset':job.received},status=201) @api_view(['GET']) def upload_status(request,upload_id): try:job=UploadSession.objects.get(pk=upload_id,owner=request.user) except UploadSession.DoesNotExist:raise Http404 return Response({'id':str(job.pk),'offset':job.received,'size':job.total_size}) @api_view(['POST']) @parser_classes([MultiPartParser,FormParser]) def upload_chunk(request,upload_id): part=request.FILES.get('part') if not part or part.size>8*1024*1024:return error('Bloc manquant ou supérieur à 8 Mio.',413) try:offset=int(request.data.get('offset','-1')) except ValueError:return error('Position de bloc invalide.') # Row lock serializes concurrent chunks and prevents repeated chunks corrupting the file. with transaction.atomic(): try:job=UploadSession.objects.select_for_update().get(pk=upload_id,owner=request.user) except UploadSession.DoesNotExist:raise Http404 if offset!=job.received: return Response({'detail':'Décalage de reprise.','offset':job.received},status=409) if offset+part.size>job.total_size:return error('Bloc plus grand que la taille totale.') path=Path(settings.FILE_UPLOAD_TEMP_DIR)/f'drive-{job.pk}.part' path.parent.mkdir(parents=True,exist_ok=True) with open(path,'a+b') as out: # A previous aborted request can leave an uncommitted partial block. out.truncate(job.received) out.seek(job.received) for block in part.chunks(chunk_size=1024*1024):out.write(block) job.received+=part.size job.save(update_fields=['received','updated_at']) return Response({'offset':job.received}) @api_view(['POST']) def upload_finish(request,upload_id): with transaction.atomic(): try:job=UploadSession.objects.select_for_update().get(pk=upload_id,owner=request.user) except UploadSession.DoesNotExist:raise Http404 src=Path(settings.FILE_UPLOAD_TEMP_DIR)/f'drive-{job.pk}.part' if job.received!=job.total_size:return error('Le fichier n’est pas entièrement transmis.') if job.total_size==0: src.touch(exist_ok=True) if not src.exists() or src.stat().st_size!=job.total_size: return error('Le fichier temporaire est incomplet.') if (job.parent and (not readable(job.parent,request.user) or not writable(job.parent,request.user,'add'))) or (not job.parent and not has_space_right(request.user,job.space,'add')): return error('Dossier de destination indisponible.',403) key=str(uuid.uuid4()) location=file_location(job.space,job.parent,job.name) created=False try: dst=assert_available(job.space,job.parent,job.name) prepare_parent(job.space,job.parent) # Exclusive creation: never overwrite a file already on disk. # Copy, rather than rename: the temporary directory can be another volume. with src.open('rb') as source, dst.open('xb') as output: created=True import shutil shutil.copyfileobj(source,output,2*1024*1024) except (NameConflict,FileExistsError) as exc: if created:dst.unlink(missing_ok=True) return error(str(exc),409) except OSError as exc: if created:dst.unlink(missing_ok=True) return error(f'Finalisation impossible : {exc}') mime=mimetypes.guess_type(job.name)[0] or 'application/octet-stream' try: node=DriveItem.objects.create(owner=request.user,space=job.space,parent=job.parent,name=job.name, size=job.total_size,mime_type=mime,storage_key=key,storage_location=location) job.delete() transaction.on_commit(lambda: src.unlink(missing_ok=True)) except Exception: dst.unlink(missing_ok=True) raise return Response(item_info(node,request.user),status=201) @api_view(['DELETE']) def upload_cancel(request,upload_id): # Serialize cancellation with chunk writes and completion. Without the lock # a concurrently arriving chunk could recreate the temp file after delete. with transaction.atomic(): try: job=UploadSession.objects.select_for_update().get(pk=upload_id,owner=request.user) except UploadSession.DoesNotExist: raise Http404 path=Path(settings.FILE_UPLOAD_TEMP_DIR)/f'drive-{job.pk}.part' job.delete() transaction.on_commit(lambda: path.unlink(missing_ok=True)) return Response(status=204) @api_view(['GET']) def upload_pending(request): """Allow a user to find/cancel their own dangling uploads after a reload. Session metadata belonging to any other account is never disclosed. """ sessions = UploadSession.objects.filter(owner=request.user).order_by('-updated_at')[:20] return Response({'sessions': [ {'id': str(job.pk), 'name': job.name, 'size': job.total_size, 'received': job.received, 'space_id': job.space_id} for job in sessions ]})