Coverage for postrfp / mail / postmark.py: 81%

1""" 

2Postmark email delivery implementation. 

3 

4This module implements the MailerProtocol through module-level functions. 

5The module itself can be used wherever MailerProtocol is expected. 

6""" 

7 

8import logging 

9from typing import Union 

10 

11import requests 

12 

13from postrfp import conf 

14 

15from postrfp.mail.schemas import TemplateEmailModel, SimpleEmailModel 

16 

17 

18log = logging.getLogger(__name__) 

19 

20REQUEST_TIMEOUT = 5 # seconds 

21 

22 

23def api_headers(key: str) -> dict[str, str]: 

24 return { 

25 "X-Postmark-Server-Token": key, 

26 "Accept": "application/json", 

27 "Content-Type": "application/json", 

28 } 

29 

30 

31def _handle_response(to_addr: str, response: requests.Response) -> str: 

32 if not response.ok: 

33 log.warning("Mail delivery to %s failed: %s", to_addr, response.text) 

34 response.raise_for_status() 

35 return response.json()["MessageID"] 

36 

37 

38def send_simple_email(model: SimpleEmailModel, headers: dict[str, str]) -> str: 

39 """ 

40 Send a simple email using Postmark API. 

41 NB: This function modifies the 'To' field to use an override address, this is for safety in 

42 development. 

43 """ 

44 # Convert Pydantic model to dict 

45 model_dict = model.model_dump() 

46 model_dict["To"] = conf.CONF.email_to_override 

47 resp = requests.post( 

48 "https://api.postmarkapp.com/email", 

49 json=model_dict, 

50 headers=headers, 

51 timeout=REQUEST_TIMEOUT, 

52 ) 

53 return _handle_response(model_dict["To"], resp) 

54 

55 

56def send_template_email(model: TemplateEmailModel, headers: dict[str, str]) -> str: 

57 """ 

58 Send a template-based email using Postmark API. 

59 

60 NB: This function modifies the 'To' field to use an override address, this is for safety in 

61 development. 

62 """ 

63 model_dict = model.model_dump() 

64 model_dict["To"] = conf.CONF.email_to_override 

65 resp = requests.post( 

66 "https://api.postmarkapp.com/email/withTemplate", 

67 json=model_dict, 

68 headers=headers, 

69 timeout=REQUEST_TIMEOUT, 

70 ) 

71 return _handle_response(model_dict["To"], resp) 

72 

73 

74def send_email(model: Union[TemplateEmailModel, SimpleEmailModel]) -> str: 

75 """ 

76 Unified email sending interface for Postmark API. 

77 

78 Determines whether to use template-based or simple email sending 

79 based on the type of the Pydantic model. 

80 

81 Args: 

82 model: Email data Pydantic model - either TemplateEmailModel for event 

83 notifications or SimpleEmailModel for basic messages. 

84 

85 Returns: 

86 str: Postmark MessageID 

87 

88 Raises: 

89 HTTPError: If Postmark API request fails 

90 KeyError: If required email fields are missing 

91 """ 

92 headers = api_headers(conf.CONF.postmark_api_key) 

93 

94 if isinstance(model, TemplateEmailModel): 

95 # Use template-based email for event notifications 

96 return send_template_email(model, headers) 

97 elif isinstance(model, SimpleEmailModel): 

98 # Use simple email for basic messages 

99 return send_simple_email(model, headers) 

100 else: 

101 raise TypeError(f"Unsupported email model type: {type(model)}") 

102 

103 

104if __name__ == "__main__": # pragma: no cover 

105 from postrfp.mail.schemas import SimpleEmailModel 

106 from postrfp.shared.init.sysconfig import configure_postrfp 

107 

108 configure_postrfp() 

109 headers = api_headers("POSTMARK_API_TEST") 

110 headers = api_headers(conf.CONF.postmark_api_key) 

111 

112 model = SimpleEmailModel( 

113 To="patrick.dobbs@supplierselect.com", 

114 From="notifications@supplierselect.com", 

115 Subject="Test blackhole", 

116 TextBody="Just a test", 

117 Tag="Testing", 

118 ) 

119 resp = send_simple_email(model, headers) 

120 print(resp)