171 simple and fast smtp client

app/config.py

@@ -1,17 +1,26 @@
 import os
 
-from pydantic import PostgresDsn, RedisDsn, computed_field
+from pydantic import PostgresDsn, RedisDsn, computed_field, BaseModel
 from pydantic_core import MultiHostUrl
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
 
+class SMTPConfig(BaseModel):
+    server: str = os.getenv("EMAIL_HOST", "smtp_server")
+    port: int = os.getenv("EMAIL_PORT", 587)
+    username: str = os.getenv("EMAIL_HOST_USER", "smtp_user")
+    password: str = os.getenv("EMAIL_HOST_PASSWORD", "smtp_password")
+
+
 class Settings(BaseSettings):
     model_config = SettingsConfigDict(
         env_file=".env", env_ignore_empty=True, extra="ignore"
     )
     jwt_algorithm: str = os.getenv("JWT_ALGORITHM")
     jwt_expire: int = os.getenv("JWT_EXPIRE")
 
+    smtp: SMTPConfig = SMTPConfig()
+
     REDIS_HOST: str
     REDIS_PORT: int
     REDIS_DB: str

app/services/smtp.py

@@ -0,0 +1,54 @@
+import smtplib
+from email.mime.multipart import MIMEMultipart
+from email.mime.text import MIMEText
+
+from app.config import settings as global_settings
+
+from fastapi.templating import Jinja2Templates
+
+from pydantic import EmailStr
+
+from app.utils.logging import AppLogger
+from app.utils.singleton import SingletonMetaNoArgs
+
+
+logger = AppLogger().get_logger()
+
+
+class SMTPEmailService(metaclass=SingletonMetaNoArgs):
+    def __init__(self):
+        self.server = smtplib.SMTP(
+            global_settings.smtp.server, global_settings.smtp.port
+        )
+        self.server.starttls()
+        self.server.login(global_settings.smtp.username, global_settings.smtp.password)
+        self.templates = Jinja2Templates("templates")
+
+    def send_email(
+        self,
+        sender: EmailStr,
+        recipients: list[EmailStr],
+        subject: str,
+        body_text: str = "",
+        body_html=None,
+    ):
+        msg = MIMEMultipart()
+        msg["From"] = sender
+        msg["To"] = ",".join(recipients)
+        msg["Subject"] = subject
+        msg.attach(MIMEText(body_text, "plain"))
+        if body_html:
+            msg.attach(MIMEText(body_html, "html"))
+        self.server.sendmail(sender, recipients, msg.as_string())
+
+    def send_template_email(
+        self,
+        recipients: list[EmailStr],
+        subject: str,
+        template: str = None,
+        context: dict = None,
+        sender: EmailStr = global_settings.smtp.from_email,
+    ):
+        template_str = self.templates.get_template(template)
+        body_html = template_str.render(context)
+        self.send_email(sender, recipients, subject, body_html=body_html)

app/utils/singleton.py

@@ -16,3 +16,21 @@ def __call__(cls, *args, **kwargs):
                 instance = super().__call__(*args, **kwargs)
                 cls._instances[cls] = instance
         return cls._instances[cls]
+
+
+class SingletonMetaNoArgs(type):
+    """
+    Singleton metaclass for classes without parameters on constructor,
+    for compatibility with FastApi Depends() function.
+    """
+
+    _instances = {}
+
+    _lock: Lock = Lock()
+
+    def __call__(cls):
+        with cls._lock:
+            if cls not in cls._instances:
+                instance = super().__call__()
+                cls._instances[cls] = instance
+        return cls._instances[cls]