|
| 1 | +from __future__ import annotations |
| 2 | + |
| 3 | +import smtplib |
| 4 | +from email.mime.text import MIMEText |
| 5 | + |
| 6 | +from patchwork.common.utils.utils import mustache_render |
| 7 | +from patchwork.step import Step |
| 8 | +from patchwork.steps.SendEmail.typed import SendEmailInputs, SendEmailOutputs |
| 9 | + |
| 10 | + |
| 11 | +class SendEmail(Step, input_class=SendEmailInputs, output_class=SendEmailOutputs): |
| 12 | + def __init__(self, inputs): |
| 13 | + super().__init__(inputs) |
| 14 | + self.email_template_value = inputs.get("email_template_value", dict()) |
| 15 | + self.subject = inputs.get("subject", "Patchwork Execution Email") |
| 16 | + self.body = inputs.get("body", "Patchwork Execution Email") |
| 17 | + self.sender_email = inputs["sender_email"] |
| 18 | + self.recipient_email = inputs["recipient_email"] |
| 19 | + self.password = inputs["sender_email_password"] |
| 20 | + self.smtp_host = inputs.get("smtp_host", "smtp.gmail.com") |
| 21 | + self.smtp_port = int(inputs.get("smtp_port", 465)) |
| 22 | + |
| 23 | + |
| 24 | + def run(self) -> dict: |
| 25 | + msg = MIMEText(mustache_render(self.body, self.email_template_value)) |
| 26 | + msg['Subject'] = mustache_render(self.subject, self.email_template_value) |
| 27 | + msg['From'] = self.sender_email |
| 28 | + msg['To'] = self.recipient_email |
| 29 | + |
| 30 | + # TODO: support smtp without ssl |
| 31 | + with smtplib.SMTP_SSL(self.smtp_host, self.smtp_port) as smtp_server: |
| 32 | + smtp_server.login(self.sender_email, self.password) |
| 33 | + smtp_server.sendmail(self.sender_email, self.recipient_email, msg.as_string()) |
| 34 | + |
| 35 | + return dict() |
0 commit comments