utils.py 3.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112
  1. import logging
  2. from datetime import datetime, timedelta
  3. from pathlib import Path
  4. from typing import Any, Dict, Optional
  5. import emails
  6. from emails.template import JinjaTemplate
  7. from jose import jwt
  8. from app.core.config import settings
  9. import random, string
  10. def send_email(
  11. email_to: str,
  12. subject_template: str = "",
  13. html_template: str = "",
  14. environment: Dict[str, Any] = {},
  15. ) -> None:
  16. assert settings.EMAILS_ENABLED, "no provided configuration for email variables"
  17. message = emails.Message(
  18. subject=JinjaTemplate(subject_template),
  19. html=JinjaTemplate(html_template),
  20. mail_from=(settings.EMAILS_FROM_NAME, settings.EMAILS_FROM_EMAIL),
  21. )
  22. smtp_options = {"host": settings.SMTP_HOST, "port": settings.SMTP_PORT}
  23. if settings.SMTP_TLS:
  24. smtp_options["tls"] = True
  25. if settings.SMTP_USER:
  26. smtp_options["user"] = settings.SMTP_USER
  27. if settings.SMTP_PASSWORD:
  28. smtp_options["password"] = settings.SMTP_PASSWORD
  29. response = message.send(to=email_to, render=environment, smtp=smtp_options)
  30. logging.info(f"send email result: {response}")
  31. def send_test_email(email_to: str) -> None:
  32. project_name = settings.PROJECT_NAME
  33. subject = f"{project_name} - Test email"
  34. with open(Path(settings.EMAIL_TEMPLATES_DIR) / "test_email.html") as f:
  35. template_str = f.read()
  36. send_email(
  37. email_to=email_to,
  38. subject_template=subject,
  39. html_template=template_str,
  40. environment={"project_name": settings.PROJECT_NAME, "email": email_to},
  41. )
  42. def send_reset_password_email(email_to: str, email: str, token: str) -> None:
  43. project_name = settings.PROJECT_NAME
  44. subject = f"{project_name} - Password recovery for user {email}"
  45. with open(Path(settings.EMAIL_TEMPLATES_DIR) / "reset_password.html") as f:
  46. template_str = f.read()
  47. server_host = settings.SERVER_HOST
  48. link = f"{server_host}/reset-password?token={token}"
  49. send_email(
  50. email_to=email_to,
  51. subject_template=subject,
  52. html_template=template_str,
  53. environment={
  54. "project_name": settings.PROJECT_NAME,
  55. "username": email,
  56. "email": email_to,
  57. "valid_hours": settings.EMAIL_RESET_TOKEN_EXPIRE_HOURS,
  58. "link": link,
  59. },
  60. )
  61. def send_new_account_email(email_to: str, username: str, password: str) -> None:
  62. project_name = settings.PROJECT_NAME
  63. subject = f"{project_name} - New account for user {username}"
  64. with open(Path(settings.EMAIL_TEMPLATES_DIR) / "new_account.html") as f:
  65. template_str = f.read()
  66. link = settings.SERVER_HOST
  67. send_email(
  68. email_to=email_to,
  69. subject_template=subject,
  70. html_template=template_str,
  71. environment={
  72. "project_name": settings.PROJECT_NAME,
  73. "username": username,
  74. "password": password,
  75. "email": email_to,
  76. "link": link,
  77. },
  78. )
  79. def generate_password_reset_token(email: str) -> str:
  80. delta = timedelta(hours=settings.EMAIL_RESET_TOKEN_EXPIRE_HOURS)
  81. now = datetime.utcnow()
  82. expires = now + delta
  83. exp = expires.timestamp()
  84. encoded_jwt = jwt.encode(
  85. {"exp": exp, "nbf": now, "sub": email}, settings.SECRET_KEY, algorithm="HS256",
  86. )
  87. return encoded_jwt
  88. def verify_password_reset_token(token: str) -> Optional[str]:
  89. try:
  90. decoded_token = jwt.decode(token, settings.SECRET_KEY, algorithms=["HS256"])
  91. return decoded_token["email"]
  92. except jwt.JWTError:
  93. return None
  94. def random_name(n):
  95. randlst = [random.choice(string.ascii_letters + string.digits) for i in range(n)]
  96. return ''.join(randlst)