v0.32.0: add email sending

So we have the same name for everyhing

README.md

@@ -10,6 +10,7 @@ This is a pip package that can be installed into any project and covers the foll
 - logging update with exception logs
 - requests wrapper for easier auth pass on access
 - dict fingerprinting
+- sending email
 - jmespath search
 - json helpers for conten replace and output
 - dump outputs for data for debugging
@@ -26,6 +27,7 @@ This is a pip package that can be installed into any project and covers the foll
 - debug_handling: various debug helpers like data dumper, timer, utilization, etc
 - db_handling: SQLite interface class
 - encyption_handling: symmetric encryption
+- email_handling: simple email sending
 - file_handling: crc handling for file content and file names, progress bar
 - json_handling: jmespath support and json date support, replace content in dict with json paths
 - iterator_handling: list and dictionary handling support (search, fingerprinting, etc)

pyproject.toml

@@ -1,7 +1,7 @@
 # MARK: Project info
 [project]
 name = "corelibs"
-version = "0.31.0"
+version = "0.32.0"
 description = "Collection of utils for Python scripts"
 readme = "README.md"
 requires-python = ">=3.13"

src/corelibs/email_handling/send_email.py

@@ -0,0 +1,199 @@
+"""
+Send email wrapper
+"""
+
+import smtplib
+from email.message import EmailMessage
+from typing import TYPE_CHECKING, Any
+if TYPE_CHECKING:
+    from corelibs.logging_handling.log import Logger
+
+
+class SendEmail:
+    """
+    send emails based on a template to a list of receivers
+    """
+
+    def __init__(
+        self,
+        log: "Logger",
+        settings: dict[str, Any],
+        template: dict[str, str],
+        from_email: str,
+        combined_send: bool = True,
+        receivers: list[str] | None = None,
+        data: list[dict[str, str]] | None = None,
+    ):
+        """
+        init send email class
+
+        Args:
+            template (dict): Dictionary with body and subject
+            from_email (str): from email as "Name" <email>
+            combined_send (bool): True for sending as one set for all receivers
+            receivers (list): list of emails to send to
+            data (dict): data to replace in template
+            args (Namespace): _description_
+        """
+        self.log = log
+        self.settings = settings
+        # internal settings
+        self.template = template
+        self.from_email = from_email
+        self.combined_send = combined_send
+        self.receivers = receivers
+        self.data = data
+
+    def send_email(
+        self,
+        data: list[dict[str, str]] | None,
+        receivers: list[str] | None,
+        template: dict[str, str] | None = None,
+        from_email: str | None = None,
+        combined_send: bool | None = None,
+        test_only: bool | None = None
+    ):
+        """
+        build email and send
+
+        Arguments:
+            data {list[dict[str, str]] | None} -- _description_
+            receivers {list[str] | None} -- _description_
+            combined_send {bool | None} -- _description_
+
+        Keyword Arguments:
+            template {dict[str, str] | None} -- _description_ (default: {None})
+            from_email {str | None} -- _description_ (default: {None})
+
+        Raises:
+            ValueError: _description_
+            ValueError: _description_
+        """
+        if data is None and self.data is not None:
+            data = self.data
+        if data is None:
+            raise ValueError("No replace data set, cannot send email")
+        if receivers is None and self.receivers is not None:
+            receivers = self.receivers
+        if receivers is None:
+            raise ValueError("No receivers list set, cannot send email")
+        if combined_send is None:
+            combined_send = self.combined_send
+        if test_only is not None:
+            self.settings['test'] = test_only
+
+        if template is None:
+            template = self.template
+        if from_email is None:
+            from_email = self.from_email
+
+        if not template['subject'] or not template['body']:
+            raise ValueError("Both Subject and Body must be set")
+
+        self.log.debug(
+            "[EMAIL]:\n"
+            f"Subject: {template['subject']}\n"
+            f"Body: {template['body']}\n"
+            f"From: {from_email}\n"
+            f"Combined send: {combined_send}\n"
+            f"Receivers: {receivers}\n"
+            f"Replace data: {data}"
+        )
+
+        # send email
+        self.send_email_list(
+            self.prepare_email_content(
+                from_email, template, data
+            ),
+            receivers,
+            combined_send,
+            test_only
+        )
+
+    def prepare_email_content(
+        self,
+        from_email: str,
+        template: dict[str, str],
+        data: list[dict[str, str]],
+    ) -> list[EmailMessage]:
+        """
+        prepare email for sending
+
+        Args:
+            template (dict): template data for this email
+            data (dict): data to replace in email
+
+        Returns:
+            list: Email Message Objects as list
+        """
+        _subject = ""
+        _body = ""
+        msg: list[EmailMessage] = []
+        for replace in data:
+            _subject = template["subject"]
+            _body = template["body"]
+            for key, value in replace.items():
+                _subject = _subject.replace(f"{{{{{key}}}}}", value)
+                _body = _body.replace(f"{{{{{key}}}}}", value)
+            # create a simple email and add subhect, from email
+            msg_email = EmailMessage()
+            # msg.set_content(_body, charset='utf-8', cte='quoted-printable')
+            msg_email.set_content(_body, charset="utf-8")
+            msg_email["Subject"] = _subject
+            msg_email["From"] = from_email
+            # push to array for sening
+            msg.append(msg_email)
+        return msg
+
+    def send_email_list(
+        self,
+        email: list[EmailMessage], receivers: list[str],
+        combined_send: bool | None = None,
+        test_only: bool | None = None
+    ):
+        """
+        send email to receivers list
+
+        Args:
+            email (list): Email Message object with set obdy, subject, from as list
+            receivers (array): email receivers list as array
+            combined_send (bool): True for sending as one set for all receivers
+        """
+
+        if test_only is not None:
+            self.settings['test'] = test_only
+
+        # localhost (postfix does the rest)
+        smtp = None
+        smtp_host = self.settings.get('smtp_host', "localhost")
+        try:
+            smtp = smtplib.SMTP(smtp_host)
+        except ConnectionRefusedError as e:
+            self.log.error("Could not open SMTP connection to: %s, %s", smtp_host, e)
+        # loop over messages and then over recievers
+        for msg in email:
+            if combined_send is True:
+                msg["To"] = ", ".join(receivers)
+                if not self.settings.get('test'):
+                    if smtp is not None:
+                        smtp.send_message(msg, msg["From"], receivers)
+                else:
+                    self.log.info(f"[EMAIL] Test, not sending email\n{msg}")
+            else:
+                for receiver in receivers:
+                    # send to
+                    self.log.debug(f"===> Send to: {receiver}")
+                    if "To" in msg:
+                        msg.replace_header("To", receiver)
+                    else:
+                        msg["To"] = receiver
+                    if not self.settings.get('test'):
+                        if smtp is not None:
+                            smtp.send_message(msg)
+                    else:
+                        self.log.info(f"[EMAIL] Test, not sending email\n{msg}")
+        # close smtp
+        if smtp is not None:
+            smtp.quit()
+
+# __END__

test-run/iterator_handling/dict_helpers.py

@@ -5,7 +5,7 @@ Iterator helper testing
 from typing import Any
 from corelibs.debug_handling.dump_data import dump_data
 from corelibs.iterator_handling.dict_mask import mask
-from corelibs.iterator_handling.dict_helper import set_entry
+from corelibs.iterator_handling.dict_helpers import set_entry
 
 
 def __mask():

tests/unit/iterator_handling/test_dict_helpers.py

@@ -6,7 +6,7 @@ iterator_handling.dict_helper tests
 
 from typing import Any
 import pytest
-from corelibs.iterator_handling.dict_helper import (
+from corelibs.iterator_handling.dict_helpers import (
     delete_keys_from_set,
     build_dict,
     set_entry,

uv.lock

@@ -108,7 +108,7 @@ wheels = [
 
 [[package]]
 name = "corelibs"
-version = "0.31.0"
+version = "0.32.0"
 source = { editable = "." }
 dependencies = [
     { name = "cryptography" },