← full-stack-fastapi-template  /  backend/app/utils.py

1
import logging
2
from dataclasses import dataclass
3
from datetime import datetime, timedelta, timezone
4
from pathlib import Path
5
from typing import Any
6
7
import emails  # type: ignore[import-untyped]
8
import jwt
9
from jinja2 import Template
10
from jwt.exceptions import InvalidTokenError
11
12
from app.core import security
13
from app.core.config import settings
14
15
logging.basicConfig(level=logging.INFO)
16
logger = logging.getLogger(__name__)
17
18
19
@dataclass
20
class EmailData:
21
    html_content: str
22
    subject: str
23
24
25
def render_email_template(*, template_name: str, context: dict[str, Any]) -> str:
26
    template_str = (
27
        Path(__file__).parent / "email-templates" / "build" / template_name
28
    ).read_text()
29
    html_content = Template(template_str).render(context)
30
    return html_content
31
32
33
def send_email(
34
    *,
35
    email_to: str,
36
    subject: str = "",
37
    html_content: str = "",
38
) -> None:
39
    assert settings.emails_enabled, "no provided configuration for email variables"
40
    message = emails.Message(
41
        subject=subject,
42
        html=html_content,
43
        mail_from=(settings.EMAILS_FROM_NAME, settings.EMAILS_FROM_EMAIL),
44
    )
45
    smtp_options = {"host": settings.SMTP_HOST, "port": settings.SMTP_PORT}
46
    if settings.SMTP_TLS:
47
        smtp_options["tls"] = True
48
    elif settings.SMTP_SSL:
49
        smtp_options["ssl"] = True
50
    if settings.SMTP_USER:
51
        smtp_options["user"] = settings.SMTP_USER
52
    if settings.SMTP_PASSWORD:
53
        smtp_options["password"] = settings.SMTP_PASSWORD
54
    response = message.send(to=email_to, smtp=smtp_options)
55
    logger.info(f"send email result: {response}")
56
57
58
def generate_test_email(email_to: str) -> EmailData:
59
    project_name = settings.PROJECT_NAME
60
    subject = f"{project_name} - Test email"
61
    html_content = render_email_template(
62
        template_name="test_email.html",
63
        context={"project_name": settings.PROJECT_NAME, "email": email_to},
64
    )
65
    return EmailData(html_content=html_content, subject=subject)
66
67
68
def generate_reset_password_email(email_to: str, email: str, token: str) -> EmailData:
69
    project_name = settings.PROJECT_NAME
70
    subject = f"{project_name} - Password recovery for user {email}"
71
    link = f"{settings.FRONTEND_HOST}/reset-password?token={token}"
72
    html_content = render_email_template(
73
        template_name="reset_password.html",
74
        context={
75
            "project_name": settings.PROJECT_NAME,
76
            "username": email,
77
            "email": email_to,
78
            "valid_hours": settings.EMAIL_RESET_TOKEN_EXPIRE_HOURS,
79
            "link": link,
80
        },
81
    )
82
    return EmailData(html_content=html_content, subject=subject)
83
84
85
def generate_new_account_email(
86
    email_to: str, username: str, password: str
87
) -> EmailData:
88
    project_name = settings.PROJECT_NAME
89
    subject = f"{project_name} - New account for user {username}"
90
    html_content = render_email_template(
91
        template_name="new_account.html",
92
        context={
93
            "project_name": settings.PROJECT_NAME,
94
            "username": username,
95
            "password": password,
96
            "email": email_to,
97
            "link": settings.FRONTEND_HOST,
98
        },
99
    )
100
    return EmailData(html_content=html_content, subject=subject)
101
102
103
def generate_password_reset_token(email: str) -> str:
104
    delta = timedelta(hours=settings.EMAIL_RESET_TOKEN_EXPIRE_HOURS)
105
    now = datetime.now(timezone.utc)
106
    expires = now + delta
107
    exp = expires.timestamp()
108
    encoded_jwt = jwt.encode(
109
        {"exp": exp, "nbf": now, "sub": email},
110
        settings.SECRET_KEY,
111
        algorithm=security.ALGORITHM,
112
    )
113
    return encoded_jwt
114
115
116
def verify_password_reset_token(token: str) -> str | None:
117
    try:
118
        decoded_token = jwt.decode(
119
            token, settings.SECRET_KEY, algorithms=[security.ALGORITHM]
120
        )
121
        return str(decoded_token["sub"])
122
    except InvalidTokenError:
123
        return None
124