[Python] MS-Graph returns 404 and 412 when uploading attachments concurrently

I have the following python script which is sending emails through MS-Graph using (partly) msgraph-sdk in python. The problem I see is that if I try to upload multiple attachments of the email concurrently I get a lot of 404 and 412 errors from ms-graph when trying to upload the attachment chunks. But if I set the semaphore to 1 (which effectively uploads the attachments sequentially) then I get no errors. Is there anything wrong with my script? Does ms-graph support concurrent attachment uploading?

Here is my python code


import asyncio
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 concurrently with semaphore (called attachment_upload_concurrency)
sem = asyncio.Semaphore(4)

async def upload_one(att: EncodedAttachment) -> None:
async with sem:
await self.upload_attachment(email_def.sender_address, draft.id, att)

tasks = [
asyncio.create_task(upload_one(att)) for att in email_def.attachments
]
await asyncio.gather(*tasks) # No return_exceptions

# 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

Continue reading...