SealSkin

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)
attributeMAX_CHUNK_BYTES
= 8 * 1024 * 1024
attributeMAX_OPEN_UPLOADS
= 16
attributeUPLOAD_TTL_SECONDS
= 24 * 3600
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).

func_open_uploads(username) -> list[str]

The user's unfinished uploads, after clearing those idle past UPLOAD_TTL_SECONDS.

paramusernamestr

Returns

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

Start a chunked upload and return its id.

One that would not fit the user's storage limit, or past MAX_OPEN_UPLOADS open, is refused.

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]