SealSkin
Server Referencerouters

uploads

Chunked uploads, scoped to the uploading user.

Uploads live under <upload_dir>/<username>/<upload uuid>/ and every consumer must pass the authenticated username, so one user can never read or complete another user's upload.

attributelogger
= logging.getLogger(__name__)
attributerouter
= APIRouter(prefix='/api/upload', dependencies=[Depends(verify_token)], route_class=EncryptedRoute)
funcvalidate_upload_id(upload_id) -> str

Ensure an upload id is a lower-case UUID.

paramupload_idstr

Client-supplied id.

Returns

str

The id unchanged.

funcuser_upload_root(username) -> str

Return the upload directory of a user (created on demand).

paramusernamestr

Returns

str
funcupload_path(username, upload_id) -> str

Return the directory of one upload after validating ownership inputs.

paramusernamestr
paramupload_idstr

Returns

str
funcreassemble_file(username, upload_id, total_chunks) -> str

Concatenate the chunks of an upload into a temporary file.

paramusernamestr

Owner of the upload.

paramupload_idstr

Upload id.

paramtotal_chunksint

Expected number of chunks.

Returns

str

Path of the reassembled temporary file (caller removes it).

funcupload_initiate(decrypted_body=Depends(get_decrypted_request_body), user=Depends(verify_token)) -> dict[str, str]

Start a chunked upload and return its id.

paramdecrypted_bodydict[str, Any]
= Depends(get_decrypted_request_body)
paramuserdict[str, Any]
= Depends(verify_token)

Returns

dict[str, str]
funcupload_chunk(decrypted_body=Depends(get_decrypted_request_body), user=Depends(verify_token)) -> dict[str, Any]

Store one chunk of an upload.

paramdecrypted_bodydict[str, Any]
= Depends(get_decrypted_request_body)
paramuserdict[str, Any]
= Depends(verify_token)

Returns

dict[str, typing.Any]
funcupload_to_storage(decrypted_body=Depends(get_decrypted_request_body), user=Depends(verify_persistent_storage_enabled)) -> dict[str, str]

Finalise an upload into the user's shared files directory.

paramdecrypted_bodydict[str, Any]
= Depends(get_decrypted_request_body)
paramuserdict[str, Any]
= Depends(verify_persistent_storage_enabled)

Returns

dict[str, str]