mirror of
https://github.com/grillazz/fastapi-sqlalchemy-asyncpg.git
synced 2025-08-26 16:40:40 +03:00
refactor and black
This commit is contained in:
parent
189158d31f
commit
6ba6d23047
@ -10,6 +10,7 @@ class SMTPConfig(BaseModel):
|
|||||||
port: int = os.getenv("EMAIL_PORT", 587)
|
port: int = os.getenv("EMAIL_PORT", 587)
|
||||||
username: str = os.getenv("EMAIL_HOST_USER", "smtp_user")
|
username: str = os.getenv("EMAIL_HOST_USER", "smtp_user")
|
||||||
password: str = os.getenv("EMAIL_HOST_PASSWORD", "smtp_password")
|
password: str = os.getenv("EMAIL_HOST_PASSWORD", "smtp_password")
|
||||||
|
template_path: str = os.getenv("EMAIL_TEMPLATE_PATH", "templates")
|
||||||
|
|
||||||
|
|
||||||
class Settings(BaseSettings):
|
class Settings(BaseSettings):
|
||||||
|
@ -45,4 +45,3 @@ class SchedulerMiddleware:
|
|||||||
await self.app(scope, receive, send)
|
await self.app(scope, receive, send)
|
||||||
else:
|
else:
|
||||||
await self.app(scope, receive, send)
|
await self.app(scope, receive, send)
|
||||||
|
|
||||||
|
@ -1,3 +1,4 @@
|
|||||||
|
from attrs import define, field
|
||||||
import smtplib
|
import smtplib
|
||||||
from email.mime.multipart import MIMEMultipart
|
from email.mime.multipart import MIMEMultipart
|
||||||
from email.mime.text import MIMEText
|
from email.mime.text import MIMEText
|
||||||
@ -15,23 +16,35 @@ from app.utils.singleton import SingletonMetaNoArgs
|
|||||||
logger = AppLogger().get_logger()
|
logger = AppLogger().get_logger()
|
||||||
|
|
||||||
|
|
||||||
|
@define
|
||||||
class SMTPEmailService(metaclass=SingletonMetaNoArgs):
|
class SMTPEmailService(metaclass=SingletonMetaNoArgs):
|
||||||
def __init__(self):
|
# SMTP configuration
|
||||||
self.server = smtplib.SMTP(
|
server_host: str = field(default=global_settings.smtp.server)
|
||||||
global_settings.smtp.server, global_settings.smtp.port
|
server_port: int = field(default=global_settings.smtp.port)
|
||||||
)
|
username: str = field(default=global_settings.smtp.username)
|
||||||
self.server.starttls()
|
password: str = field(default=global_settings.smtp.password)
|
||||||
self.server.login(global_settings.smtp.username, global_settings.smtp.password)
|
|
||||||
self.templates = Jinja2Templates("templates")
|
|
||||||
|
|
||||||
def send_email(
|
# Dependencies
|
||||||
|
templates: Jinja2Templates = field(
|
||||||
|
factory=lambda: Jinja2Templates(global_settings.templates_dir)
|
||||||
|
)
|
||||||
|
server: smtplib.SMTP = field(init=False) # Deferred initialization in post-init
|
||||||
|
|
||||||
|
def __attrs_post_init__(self):
|
||||||
|
"""Initialize the SMTP server connection after object creation."""
|
||||||
|
self.server = smtplib.SMTP(self.server_host, self.server_port)
|
||||||
|
self.server.starttls()
|
||||||
|
self.server.login(self.username, self.password)
|
||||||
|
|
||||||
|
def _prepare_email(
|
||||||
self,
|
self,
|
||||||
sender: EmailStr,
|
sender: EmailStr,
|
||||||
recipients: list[EmailStr],
|
recipients: list[EmailStr],
|
||||||
subject: str,
|
subject: str,
|
||||||
body_text: str = "",
|
body_text: str,
|
||||||
body_html=None,
|
body_html: str,
|
||||||
):
|
) -> MIMEMultipart:
|
||||||
|
"""Prepare the email message."""
|
||||||
msg = MIMEMultipart()
|
msg = MIMEMultipart()
|
||||||
msg["From"] = sender
|
msg["From"] = sender
|
||||||
msg["To"] = ",".join(recipients)
|
msg["To"] = ",".join(recipients)
|
||||||
@ -39,16 +52,29 @@ class SMTPEmailService(metaclass=SingletonMetaNoArgs):
|
|||||||
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"))
|
||||||
|
return msg
|
||||||
|
|
||||||
|
def send_email(
|
||||||
|
self,
|
||||||
|
sender: EmailStr,
|
||||||
|
recipients: list[EmailStr],
|
||||||
|
subject: str,
|
||||||
|
body_text: str = "",
|
||||||
|
body_html: str = None,
|
||||||
|
):
|
||||||
|
"""Send a regular email (plain text or 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())
|
||||||
|
|
||||||
def send_template_email(
|
def send_template_email(
|
||||||
self,
|
self,
|
||||||
recipients: list[EmailStr],
|
recipients: list[EmailStr],
|
||||||
subject: str,
|
subject: str,
|
||||||
template: str = None,
|
template: str,
|
||||||
context: dict = None,
|
context: dict,
|
||||||
sender: EmailStr = global_settings.smtp.from_email,
|
sender: EmailStr,
|
||||||
):
|
):
|
||||||
|
"""Send an email using a template with the provided context."""
|
||||||
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)
|
||||||
self.send_email(sender, recipients, subject, body_html=body_html)
|
self.send_email(sender, recipients, subject, body_html=body_html)
|
||||||
|
Loading…
x
Reference in New Issue
Block a user