Store a DM attachment using an existing DB connection/transaction.
(conn, dm_id: int, file_storage)
| 12650 | if estimated > usage.free or (usage.total and ((usage.used + estimated) / usage.total) >= 0.95): |
| 12651 | raise OSError("Not enough temporary storage to create ZIP") |
| 12652 | fd, zip_path = tempfile.mkstemp(prefix="vault-download-", suffix=".zip", dir=TMP_UPLOAD_DIR); os.close(fd) |
| 12653 | root = user_root(owner) |
| 12654 | with zipfile.ZipFile(zip_path, "w", compression=zipfile.ZIP_DEFLATED, allowZip64=True) as archive: |
| 12655 | for rel in relpaths: |
| 12656 | rel = safe_relpath(rel); target = abs_user_path(owner, rel) |
| 12657 | if not os.path.exists(target): continue |
| 12658 | if os.path.isfile(target): |
| 12659 | arcname = rel or os.path.basename(target) |
| 12660 | if is_encrypted_file(target): |
| 12661 | with archive.open(arcname, "w") as out: |
| 12662 | for block in aesgcm_decrypt_generator(target): out.write(block) |
| 12663 | else: archive.write(target, arcname=arcname) |
| 12664 | else: |
| 12665 | base_parent = os.path.dirname(target) |
| 12666 | for current, dirs, files_ in os.walk(target, followlinks=False): |
| 12667 | dirs[:] = [d for d in dirs if not os.path.islink(os.path.join(current, d))] |
| 12668 | for name in files_: |
| 12669 | fp = os.path.join(current, name) |
| 12670 | if os.path.islink(fp): continue |
| 12671 | arcname = os.path.relpath(fp, base_parent).replace(os.sep, "/") |
| 12672 | if is_encrypted_file(fp): |
| 12673 | with archive.open(arcname, "w") as out: |
| 12674 | for block in aesgcm_decrypt_generator(fp): out.write(block) |
| 12675 | else: archive.write(fp, arcname=arcname) |
no test coverage detected