Coverage for documents/mail.py: 29%

38 statements  

« prev     ^ index     » next       coverage.py v7.15.2, created at 2026-10-10 09:07 +0000

1from __future__ import annotations 

2 

3from dataclasses import dataclass 

4from email import message_from_bytes 

5from pathlib import Path 

6 

7from django.conf import settings 

8from django.core.mail import EmailMessage 

9from filelock import FileLock 

10 

11 

12@dataclass(frozen=True) 

13class EmailAttachment: 

14 path: Path 

15 mime_type: str 

16 friendly_name: str 

17 

18 

19def send_email( 

20 subject: str, 

21 body: str, 

22 to: list[str], 

23 attachments: list[EmailAttachment], 

24) -> int: 

25 """ 

26 Send an email with attachments. 

27 

28 Args: 

29 subject: Email subject 

30 body: Email body text 

31 to: List of recipient email addresses 

32 attachments: List of attachments 

33 

34 Returns: 

35 Number of emails sent 

36 

37 TODO: re-evaluate this pending https://code.djangoproject.com/ticket/35581 / https://github.com/django/django/pull/18966 

38 """ 

39 if "\r" in subject or "\n" in subject: 

40 subject = " ".join(line.strip(" \t") for line in subject.splitlines()) 

41 

42 email = EmailMessage( 

43 subject=subject, 

44 body=body, 

45 to=to, 

46 ) 

47 

48 used_filenames: set[str] = set() 

49 

50 # Something could be renaming the file concurrently so it can't be attached 

51 with FileLock(settings.MEDIA_LOCK): 

52 for attachment in attachments: 

53 filename = _get_unique_filename( 

54 attachment.friendly_name, 

55 used_filenames, 

56 ) 

57 used_filenames.add(filename) 

58 

59 with attachment.path.open("rb") as f: 

60 content = f.read() 

61 if attachment.mime_type == "message/rfc822": 

62 # See https://forum.djangoproject.com/t/using-emailmessage-with-an-attached-email-file-crashes-due-to-non-ascii/37981 

63 content = message_from_bytes(content) 

64 

65 email.attach( 

66 filename=filename, 

67 content=content, 

68 mimetype=attachment.mime_type, 

69 ) 

70 

71 return email.send() 

72 

73 

74def _get_unique_filename(friendly_name: str, used_names: set[str]) -> str: 

75 """ 

76 Constructs a unique friendly filename for the given document, append a counter if needed. 

77 """ 

78 if friendly_name not in used_names: 

79 return friendly_name 

80 

81 stem = Path(friendly_name).stem 

82 suffix = "".join(Path(friendly_name).suffixes) 

83 

84 counter = 1 

85 while True: 

86 filename = f"{stem}_{counter:02}{suffix}" 

87 if filename not in used_names: 

88 return filename 

89 counter += 1