trading_agent.attachments.attachment_store

Keeps the files the user sends with messages, in each desk's uploads folder.

Each upload gets its own folder, named by its identifier, under the India date it arrived:

desks//uploads/2026-10-17/3f9a1c2b7d4e/ report.pdf the file as it was sent text.txt for a document, its text, labelled as data from outside trading-agent attachment.json what the store knows about it

A picture is checked to be a real PNG, JPEG, WebP or GIF by its contents. A document is turned into text once, when it arrives, so every model reads the same text and nothing has to be parsed again. The text file starts and ends with the same marker lines that label web pages, because a document can come from anywhere and must not be able to give the desk instructions. Uploads are kept: the desk can come back to a document weeks later.

Typical usage example:

store = AttachmentStore(layout) attachment = store.save('main', 'report.pdf', data) same = store.find('main', attachment.attachment_id)

MAXIMUM_BYTES module-attribute

MAXIMUM_BYTES = 20 * 1024 * 1024

IMAGE_EXTENSIONS module-attribute

IMAGE_EXTENSIONS = ('.png', '.jpg', '.jpeg', '.webp', '.gif')

AttachmentRefusedError

Bases: ValueError

An upload cannot be accepted: it is empty, too large, of a type not taken, or damaged.

AttachmentNotFoundError

Bases: LookupError

No upload has that identifier on that desk.

AttachmentStore

AttachmentStore(layout, image_preparer=None, readers=None, untrusted_text=None, clock=None)

Every desk's uploads.

Creates the store.

Parameters:

Name	Type	Description	Default
layout	WorkspaceLayout	Where the desk folders are.	required
image_preparer	ImagePreparer | None	Checks pictures; None for a new one.	None
readers	Sequence[TextReader] | None	The document readers; None for PDF, Word, Excel and plain text.	None
untrusted_text	UntrustedText | None	Labels document text as data; None for a new one.	None
clock	Callable[[], datetime] | None	Returns the current time; None uses the system clock. Tests replace it.	None

Source code in src/trading_agent/attachments/attachment_store.py

def __init__(
    self,
    layout: workspace_layout.WorkspaceLayout,
    image_preparer: image_preparer_module.ImagePreparer | None = None,
    readers: Sequence[text_reader.TextReader] | None = None,
    untrusted_text: untrusted_text_module.UntrustedText | None = None,
    clock: Callable[[], datetime.datetime] | None = None,
):
    """Creates the store.

    Args:
        layout (workspace_layout.WorkspaceLayout): Where the desk folders are.
        image_preparer (image_preparer_module.ImagePreparer | None): Checks pictures; None for a new one.
        readers (Sequence[text_reader.TextReader] | None): The document readers; None for PDF, Word, Excel and plain text.
        untrusted_text (untrusted_text_module.UntrustedText | None): Labels document text as data; None for a new one.
        clock (Callable[[], datetime.datetime] | None): Returns the current time; None uses the system clock. Tests replace it.
    """
    self._layout = layout
    self._image_preparer = (
        image_preparer or image_preparer_module.ImagePreparer()
    )
    if readers is None:
        readers = [
            pdf_text_reader.PdfTextReader(),
            word_text_reader.WordTextReader(),
            excel_text_reader.ExcelTextReader(),
            plain_text_reader.PlainTextReader(),
        ]
    self._readers = list(readers)
    self._untrusted_text = (
        untrusted_text or untrusted_text_module.UntrustedText()
    )
    self._clock = clock or self._real_now

accepted_extensions

accepted_extensions()

Every file ending the store takes.

Returns:

Type	Description
list[str]	The picture endings, then the document endings.

Source code in src/trading_agent/attachments/attachment_store.py

def accepted_extensions(self) -> list[str]:
    """Every file ending the store takes.

    Returns:
        list[str]: The picture endings, then the document endings.
    """
    extensions = list(IMAGE_EXTENSIONS)
    for reader in self._readers:
        extensions.extend(reader.extensions)
    return extensions

save

save(desk_name, file_name, data)

Stores one upload, and reads a document's text.

Parameters:

Name	Type	Description	Default
desk_name	str	The desk it is for.	required
file_name	str	The file's name as the user sent it.	required
data	bytes	The file's contents.	required

Returns:

Type	Description
Attachment	The stored upload.

Raises:

Type	Description
AttachmentRefusedError	The file is empty, larger than 20 MB, of a type not taken, or a damaged picture.

Source code in src/trading_agent/attachments/attachment_store.py

def save(
    self,
    desk_name: str,
    file_name: str,
    data: bytes,
) -> attachment_module.Attachment:
    """Stores one upload, and reads a document's text.

    Args:
        desk_name (str): The desk it is for.
        file_name (str): The file's name as the user sent it.
        data (bytes): The file's contents.

    Returns:
        attachment_module.Attachment: The stored upload.

    Raises:
        AttachmentRefusedError: The file is empty, larger than 20 MB, of a type not taken, or a damaged picture.
    """
    self._check_size(file_name, data)
    safe_name = self._safe_name(file_name)
    reader = self._reader_for(safe_name)
    is_image = Path(safe_name).suffix.lower() in IMAGE_EXTENSIONS
    if not is_image and reader is None:
        raise AttachmentRefusedError(
            f'{file_name} is not a type the desks can use. Send a picture or one of: {", ".join(self.accepted_extensions())}.'
        )
    now = self._clock()
    attachment_id = uuid.uuid4().hex[:12]
    folder = (
        self._layout.uploads_directory(desk_name)
        / f'{now.astimezone(_INDIA):%Y-%m-%d}'
        / attachment_id
    )
    folder.mkdir(parents=True)
    stored = folder / safe_name
    stored.write_bytes(data)
    if is_image:
        attachment = self._image_attachment(
            desk_name, file_name, attachment_id, stored, len(data), now
        )
    else:
        attachment = self._document_attachment(
            desk_name,
            file_name,
            attachment_id,
            stored,
            reader,
            now,
        )
    (folder / _RECORD_FILE_NAME).write_text(
        json.dumps(attachment.to_json_dictionary(), indent=2),
        encoding='utf-8',
    )
    return attachment

find

find(desk_name, attachment_id)

Finds one of a desk's uploads.

Parameters:

Name	Type	Description	Default
desk_name	str	The desk.	required
attachment_id	str	The upload's identifier.	required

Returns:

Type	Description
Attachment	The upload.

Raises:

Type	Description
AttachmentNotFoundError	The desk has no upload with that identifier.

Source code in src/trading_agent/attachments/attachment_store.py

def find(
    self,
    desk_name: str,
    attachment_id: str,
) -> attachment_module.Attachment:
    """Finds one of a desk's uploads.

    Args:
        desk_name (str): The desk.
        attachment_id (str): The upload's identifier.

    Returns:
        attachment_module.Attachment: The upload.

    Raises:
        AttachmentNotFoundError: The desk has no upload with that identifier.
    """
    if not _IDENTIFIER_PATTERN.match(attachment_id):
        raise AttachmentNotFoundError(
            f'{attachment_id!r} is not an upload identifier'
        )
    uploads = self._layout.uploads_directory(desk_name)
    desk_directory = self._layout.desk_directory(desk_name)
    for record in uploads.glob(f'*/{attachment_id}/{_RECORD_FILE_NAME}'):
        dictionary = json.loads(record.read_text(encoding='utf-8'))
        dictionary['path'] = str(desk_directory / dictionary['desk_path'])
        return attachment_module.Attachment.from_json_dictionary(dictionary)
    raise AttachmentNotFoundError(
        f'The {desk_name} desk has no upload {attachment_id}'
    )