mirror of
https://github.com/grillazz/fastapi-sqlalchemy-asyncpg.git
synced 2025-08-26 16:40:40 +03:00
docz and logz
This commit is contained in:
parent
6ba6d23047
commit
abb6eae23a
@ -18,6 +18,24 @@ logger = AppLogger().get_logger()
|
|||||||
|
|
||||||
@define
|
@define
|
||||||
class SMTPEmailService(metaclass=SingletonMetaNoArgs):
|
class SMTPEmailService(metaclass=SingletonMetaNoArgs):
|
||||||
|
"""
|
||||||
|
SMTPEmailService provides a reusable interface to send emails via an SMTP server.
|
||||||
|
|
||||||
|
This service supports plaintext and HTML emails, and also allows
|
||||||
|
sending template-based emails using the Jinja2 template engine.
|
||||||
|
|
||||||
|
It is implemented as a singleton to ensure that only one SMTP connection is maintained
|
||||||
|
throughout the application lifecycle, optimizing resource usage.
|
||||||
|
|
||||||
|
Attributes:
|
||||||
|
server_host (str): SMTP server hostname or IP address.
|
||||||
|
server_port (int): Port number for the SMTP connection.
|
||||||
|
username (str): SMTP username for authentication.
|
||||||
|
password (str): SMTP password for authentication.
|
||||||
|
templates (Jinja2Templates): Jinja2Templates instance for loading and rendering email templates.
|
||||||
|
server (smtplib.SMTP): An SMTP object for sending emails, initialized after object creation.
|
||||||
|
"""
|
||||||
|
|
||||||
# SMTP configuration
|
# SMTP configuration
|
||||||
server_host: str = field(default=global_settings.smtp.server)
|
server_host: str = field(default=global_settings.smtp.server)
|
||||||
server_port: int = field(default=global_settings.smtp.port)
|
server_port: int = field(default=global_settings.smtp.port)
|
||||||
@ -26,15 +44,21 @@ class SMTPEmailService(metaclass=SingletonMetaNoArgs):
|
|||||||
|
|
||||||
# Dependencies
|
# Dependencies
|
||||||
templates: Jinja2Templates = field(
|
templates: Jinja2Templates = field(
|
||||||
factory=lambda: Jinja2Templates(global_settings.templates_dir)
|
factory=lambda: Jinja2Templates(global_settings.smtp.template_path)
|
||||||
)
|
)
|
||||||
server: smtplib.SMTP = field(init=False) # Deferred initialization in post-init
|
server: smtplib.SMTP = field(init=False) # Deferred initialization in post-init
|
||||||
|
|
||||||
def __attrs_post_init__(self):
|
def __attrs_post_init__(self):
|
||||||
"""Initialize the SMTP server connection after object creation."""
|
"""
|
||||||
|
Initializes the SMTP server connection after the object is created.
|
||||||
|
|
||||||
|
This method sets up a secure connection to the SMTP server, including STARTTLS encryption
|
||||||
|
and logs in using the provided credentials.
|
||||||
|
"""
|
||||||
self.server = smtplib.SMTP(self.server_host, self.server_port)
|
self.server = smtplib.SMTP(self.server_host, self.server_port)
|
||||||
self.server.starttls()
|
self.server.starttls() # Upgrade the connection to secure TLS
|
||||||
self.server.login(self.username, self.password)
|
self.server.login(self.username, self.password)
|
||||||
|
logger.info("SMTPEmailService initialized successfully and connected to SMTP server.")
|
||||||
|
|
||||||
def _prepare_email(
|
def _prepare_email(
|
||||||
self,
|
self,
|
||||||
@ -44,14 +68,28 @@ class SMTPEmailService(metaclass=SingletonMetaNoArgs):
|
|||||||
body_text: str,
|
body_text: str,
|
||||||
body_html: str,
|
body_html: str,
|
||||||
) -> MIMEMultipart:
|
) -> MIMEMultipart:
|
||||||
"""Prepare the email message."""
|
"""
|
||||||
|
Prepares a MIME email message with the given plaintext and HTML content.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
sender (EmailStr): The email address of the sender.
|
||||||
|
recipients (list[EmailStr]): A list of recipient email addresses.
|
||||||
|
subject (str): The subject line of the email.
|
||||||
|
body_text (str): The plaintext content of the email.
|
||||||
|
body_html (str): The HTML content of the email (optional).
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
MIMEMultipart: A MIME email object ready to be sent.
|
||||||
|
"""
|
||||||
msg = MIMEMultipart()
|
msg = MIMEMultipart()
|
||||||
msg["From"] = sender
|
msg["From"] = sender
|
||||||
msg["To"] = ",".join(recipients)
|
msg["To"] = ",".join(recipients)
|
||||||
msg["Subject"] = subject
|
msg["Subject"] = subject
|
||||||
|
# Add plain text and HTML content (if provided)
|
||||||
msg.attach(MIMEText(body_text, "plain"))
|
msg.attach(MIMEText(body_text, "plain"))
|
||||||
if body_html:
|
if body_html:
|
||||||
msg.attach(MIMEText(body_html, "html"))
|
msg.attach(MIMEText(body_html, "html"))
|
||||||
|
logger.debug(f"Prepared email from {sender} to {recipients}.")
|
||||||
return msg
|
return msg
|
||||||
|
|
||||||
def send_email(
|
def send_email(
|
||||||
@ -62,9 +100,29 @@ class SMTPEmailService(metaclass=SingletonMetaNoArgs):
|
|||||||
body_text: str = "",
|
body_text: str = "",
|
||||||
body_html: str = None,
|
body_html: str = None,
|
||||||
):
|
):
|
||||||
"""Send a regular email (plain text or HTML)."""
|
"""
|
||||||
|
Sends an email to the specified recipients.
|
||||||
|
|
||||||
|
Supports plaintext and HTML email content. This method constructs
|
||||||
|
the email message using `_prepare_email` and sends it using the SMTP server.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
sender (EmailStr): The email address of the sender.
|
||||||
|
recipients (list[EmailStr]): A list of recipient email addresses.
|
||||||
|
subject (str): The subject line of the email.
|
||||||
|
body_text (str): The plaintext content of the email.
|
||||||
|
body_html (str): The HTML content of the email (optional).
|
||||||
|
|
||||||
|
Raises:
|
||||||
|
smtplib.SMTPException: If the email cannot be sent.
|
||||||
|
"""
|
||||||
|
try:
|
||||||
msg = self._prepare_email(sender, recipients, subject, body_text, body_html)
|
msg = self._prepare_email(sender, recipients, subject, body_text, body_html)
|
||||||
self.server.sendmail(sender, recipients, msg.as_string())
|
self.server.sendmail(sender, recipients, msg.as_string())
|
||||||
|
logger.info(f"Email sent successfully to {recipients} from {sender}.")
|
||||||
|
except smtplib.SMTPException as e:
|
||||||
|
logger.error("Failed to send email", exc_info=e)
|
||||||
|
raise
|
||||||
|
|
||||||
def send_template_email(
|
def send_template_email(
|
||||||
self,
|
self,
|
||||||
@ -74,7 +132,28 @@ class SMTPEmailService(metaclass=SingletonMetaNoArgs):
|
|||||||
context: dict,
|
context: dict,
|
||||||
sender: EmailStr,
|
sender: EmailStr,
|
||||||
):
|
):
|
||||||
"""Send an email using a template with the provided context."""
|
"""
|
||||||
|
Sends an email using a Jinja2 template.
|
||||||
|
|
||||||
|
This method renders the template with the provided context and sends it
|
||||||
|
to the specified recipients.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
recipients (list[EmailStr]): A list of recipient email addresses.
|
||||||
|
subject (str): The subject line of the email.
|
||||||
|
template (str): The name of the template file in the templates directory.
|
||||||
|
context (dict): A dictionary of values to render the template with.
|
||||||
|
sender (EmailStr): The email address of the sender.
|
||||||
|
|
||||||
|
Raises:
|
||||||
|
jinja2.TemplateNotFound: If the specified template is not found.
|
||||||
|
smtplib.SMTPException: If the email cannot be sent.
|
||||||
|
"""
|
||||||
|
try:
|
||||||
template_str = self.templates.get_template(template)
|
template_str = self.templates.get_template(template)
|
||||||
body_html = template_str.render(context)
|
body_html = template_str.render(context) # Render the HTML using context variables
|
||||||
self.send_email(sender, recipients, subject, body_html=body_html)
|
self.send_email(sender, recipients, subject, body_html=body_html)
|
||||||
|
logger.info(f"Template email sent successfully to {recipients} using template {template}.")
|
||||||
|
except Exception as e:
|
||||||
|
logger.error("Failed to send template email", exc_info=e)
|
||||||
|
raise
|
Loading…
x
Reference in New Issue
Block a user