192 lines
9.4 KiB
Python
192 lines
9.4 KiB
Python
"""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
|
||
]})
|