MS-Graph returns 404 or 412 or Timeouts when uploading large attachment chunks

I have the following python script which is sending emails through MS-Graph using msgraph-sdk in python.
I use the ms-graph-sdk to create the draft email and the upload sessions for the attachments.

The problem I experience is with chunk uploads for large attachments. The python ms-graph sdk does not natively support uploading the attachment chunks so I am using an http client (aiohttp) to upload the chunks using the uploadUrl I receive from the upload session response.

When my service is running on my Azure env I get Timeout errors quite frequently when uploading attachment chunks. My upload requests are sequential. I have a retry mechanism in place which 99% of the times manages to send the email eventually but there are occassions where the email sending is failing after 3 retry attempts.
I have also seen more rarely some 404 or 400 responses when trying to upload an attachment chunk which doesn't make sense since the uploadUrl is provided by MS-Graph itself.

Here is my python code

import aiohttp

from mylib import EmailSender
from mylib import (
    EmailDefinition,
    EmailContent,
    EncodedAttachment,
)
from azure.identity import ClientSecretCredential
from msgraph.graph_service_client import GraphServiceClient
from mylib import get_logger
from msgraph.generated.models.message import Message
from msgraph.generated.models.item_body import ItemBody
from msgraph.generated.models.body_type import BodyType
from msgraph.generated.models.recipient import Recipient
from msgraph.generated.models.email_address import EmailAddress
from msgraph.generated.models.attachment_type import AttachmentType
from msgraph.generated.models.attachment_item import AttachmentItem
from msgraph.generated.users.item.messages.item.attachments.create_upload_session.create_upload_session_post_request_body import (
    CreateUploadSessionPostRequestBody,
)

logger = get_logger(__name__)

CHUNK_SIZE = 3 * 1024 * 1024  # 3MB
CHUNK_UPLOAD_TIMEOUT_SECONDS = 20


class EmailSenderMSG(EmailSender):
    def __init__(self, cfg):
        _credential = ClientSecretCredential(
            tenant_id=cfg.tenant_id,
            client_id=cfg.client_id,
            client_secret=cfg.client_secret.get_secret_value(),
        )
        self._graph_client = GraphServiceClient(
            credentials=_credential, scopes=[cfg.scope]
        )

    async def _send_large_email(self, email_def: EmailDefinition):
        # Implementation reference https://learn.microsoft.com/en-us/graph/outlook-large-attachments?tabs=python
        # Step 1: Create Draft
        message = self.create_base_message(email_def)

        draft = self._graph_client.users.by_user_id(email_def.sender_address).messages.post(message)

        # Step 2: Upload attachments sequentially
        for att in email_def.attachments:
            await self.upload_attachment(email_def.sender_address, draft.id, att)
            
        # Step 4: Send the draft
        logger.info(f"Finalising draft message with id {draft.id}")
        await (self._graph_client.users.by_user_id(email_def.sender_address)
               .messages.by_message_id(draft.id).send.post())

    async def upload_attachment(
            self, sender_address: str, draft_message_id: str, attachment: EncodedAttachment) -> None:
        attachment_size = len(attachment.content)
        upload_session_body = CreateUploadSessionPostRequestBody(
            attachment_item=AttachmentItem(
                attachment_type=AttachmentType.File,
                name=attachment.name,
                size=attachment_size,
                content_type=attachment.content_type,
            )
        )

        logger.info(
            f"Creating upload session for attachment '{attachment.name}' of size {attachment_size} bytes. Request body: {upload_session_body}"
        )
        upload_session = await (
            self._graph_client.users.by_user_id(sender_address).messages.by_message_id(draft_message_id)
            .attachments.create_upload_session.post(upload_session_body))
        logger.debug(f"Upload session for attachment '{attachment.name}' : {upload_session}")

        # Step 3: Upload in chunks (simple, no retries for now)
        await self.upload_chunks(attachment, attachment_size, upload_session)

    @staticmethod
    async def upload_chunks(attachment, attachment_size, upload_session):
        async with aiohttp.ClientSession() as session:
            for i in range(0, attachment_size, CHUNK_SIZE):
                chunk = attachment.content[i: i + CHUNK_SIZE]
                start = i
                end = i + len(chunk) - 1
                total = attachment_size

                # msgraph-dsk is not supporting attachment chunk uploading, so we use aiohttp directly
                headers = {
                    "Content-Length": str(len(chunk)),
                    "Content-Range": f"bytes {start}-{end}/{total}",
                    "Content-Type": "application/octet-stream",
                }

                logger.debug(
                    f"Uploading chunk of attachment '{attachment.name}' with url {upload_session.upload_url} and headers {headers}"
                )
                await EmailSenderMSG.upload_single_chunk(
                    session,
                    upload_session.upload_url,
                    chunk,
                    headers,
                    attachment.name,
                )

    @staticmethod
    async def upload_single_chunk(
            session: aiohttp.ClientSession, url: str, chunk: bytes, headers: dict[str, str], attachment_name: str
    ) -> None:
        async with session.put(
                url,
                data=chunk,
                headers=headers,
                timeout=aiohttp.ClientTimeout(total=CHUNK_UPLOAD_TIMEOUT_SECONDS),
        ) as response:
            try:
                response.raise_for_status()
            except aiohttp.ClientResponseError as e:
                logger.warning(
                    f"ClientResponseError: {e.status}, attachment '{attachment_name}', {e.message}, Headers: {e.headers}",
                    exc_info=True,
                )
                raise

    @staticmethod
    def _create_item_body(email_content: EmailContent) -> ItemBody:
        if email_content.html:
            return ItemBody(content_type=BodyType.Html, content=email_content.html)
        elif email_content.plain_text:
            return ItemBody(
                content_type=BodyType.Text, content=email_content.plain_text
            )

        raise ValueError("Email content must have either 'html' or 'plain_text' set.")

    @staticmethod
    def create_base_message(email_def):
        message = Message(
            subject=email_def.content.subject,
            body=EmailSenderMSG._create_item_body(email_def.content),
            to_recipients=[
                Recipient(email_address=EmailAddress(address=addr))
                for addr in email_def.recipients.to
            ],
            cc_recipients=[
                Recipient(email_address=EmailAddress(address=addr))
                for addr in email_def.recipients.cc
            ],
            bcc_recipients=[
                Recipient(email_address=EmailAddress(address=addr))
                for addr in email_def.recipients.bcc
            ],
        )
        return message

thanks