"""Сохранение вложений Streamlit на диск не должно удваивать их в памяти.
UploadedFile — это BytesIO: Streamlit уже держит всё содержимое вложения в
памяти. getvalue() делает ещё одну полную копию каждого файла перед записью
на диск; при 230 вложениях в контейнере на 1 ГБ это лишняя пиковая нагрузка.
Копировать нужно потоково, через файловый интерфейс, не трогая getvalue().
"""
from __future__ import annotations
import io
from pathlib import Path
from main import _save_attachments
class _UploadWithoutGetValue(io.BytesIO):
"""Вложение, у которого есть только файловый интерфейс и имя."""
def __init__(self, name: str, payload: bytes) -> None:
super().__init__(payload)
self.name = name
def getvalue(self) -> bytes: # type: ignore[override]
raise AssertionError("getvalue() копирует всё вложение в память")
def test_save_attachments_streams_without_getvalue(tmp_path: Path) -> None:
uploads = [
_UploadWithoutGetValue("a.xml", b""),
_UploadWithoutGetValue("a.xml", b"" * 1000),
]
paths = _save_attachments(uploads, tmp_path)
assert [path.name for path in paths] == ["a.xml", "a.xml"]
assert paths[0] != paths[1]
assert paths[0].read_bytes() == b""
assert paths[1].read_bytes() == b"" * 1000
def test_save_attachments_copies_from_start_even_if_upload_was_read(
tmp_path: Path,
) -> None:
upload = _UploadWithoutGetValue("f.xml", b"0123456789")
upload.read(4)
(path,) = _save_attachments([upload], tmp_path)
assert path.read_bytes() == b"0123456789"