✨ Update all for Postgres and new techniques

{{cookiecutter.project_slug}}/backend/app/app/utils.py

@@ -1,96 +1,86 @@
 import logging
 from datetime import datetime, timedelta
 from pathlib import Path
-from typing import Union
+from typing import Optional
 
 import emails
 import jwt
 from emails.template import JinjaTemplate
 from jwt.exceptions import InvalidTokenError
 
-from app.core.config import (
-    EMAIL_RESET_TOKEN_EXPIRE_HOURS,
-    EMAIL_TEMPLATES_DIR,
-    EMAILS_ENABLED,
-    EMAILS_FROM_EMAIL,
-    EMAILS_FROM_NAME,
-    PROJECT_NAME,
-    SECRET_KEY,
-    SERVER_HOST,
-    SMTP_HOST,
-    SMTP_PASSWORD,
-    SMTP_PORT,
-    SMTP_TLS,
-    SMTP_USER,
-)
+from app.core import config
 
 password_reset_jwt_subject = "preset"
 
 
 def send_email(email_to: str, subject_template="", html_template="", environment={}):
-    assert EMAILS_ENABLED, "no provided configuration for email variables"
+    assert config.EMAILS_ENABLED, "no provided configuration for email variables"
     message = emails.Message(
         subject=JinjaTemplate(subject_template),
         html=JinjaTemplate(html_template),
-        mail_from=(EMAILS_FROM_NAME, EMAILS_FROM_EMAIL),
+        mail_from=(config.EMAILS_FROM_NAME, config.EMAILS_FROM_EMAIL),
     )
-    smtp_options = {"host": SMTP_HOST, "port": SMTP_PORT}
-    if SMTP_TLS:
+    smtp_options = {"host": config.SMTP_HOST, "port": config.SMTP_PORT}
+    if config.SMTP_TLS:
         smtp_options["tls"] = True
-    if SMTP_USER:
-        smtp_options["user"] = SMTP_USER
-    if SMTP_PASSWORD:
-        smtp_options["password"] = SMTP_PASSWORD
+    if config.SMTP_USER:
+        smtp_options["user"] = config.SMTP_USER
+    if config.SMTP_PASSWORD:
+        smtp_options["password"] = config.SMTP_PASSWORD
     response = message.send(to=email_to, render=environment, smtp=smtp_options)
     logging.info(f"send email result: {response}")
 
 
 def send_test_email(email_to: str):
-    subject = f"{PROJECT_NAME} - Test email"
-    with open(Path(EMAIL_TEMPLATES_DIR) / "test_email.html") as f:
+    project_name = config.PROJECT_NAME
+    subject = f"{project_name} - Test email"
+    with open(Path(config.EMAIL_TEMPLATES_DIR) / "test_email.html") as f:
         template_str = f.read()
     send_email(
         email_to=email_to,
         subject_template=subject,
         html_template=template_str,
-        environment={"project_name": PROJECT_NAME, "email": email_to},
+        environment={"project_name": config.PROJECT_NAME, "email": email_to},
     )
 
 
-def send_reset_password_email(email_to: str, username: str, token: str):
-    subject = f"{PROJECT_NAME} - Password recovery for user {username}"
-    with open(Path(EMAIL_TEMPLATES_DIR) / "reset_password.html") as f:
+def send_reset_password_email(email_to: str, email: str, token: str):
+    project_name = config.PROJECT_NAME
+    subject = f"{project_name} - Password recovery for user {email}"
+    with open(Path(config.EMAIL_TEMPLATES_DIR) / "reset_password.html") as f:
         template_str = f.read()
     if hasattr(token, "decode"):
         use_token = token.decode()
     else:
         use_token = token
-    link = f"{SERVER_HOST}/reset-password?token={use_token}"
+    server_host = config.SERVER_HOST
+    link = f"{server_host}/reset-password?token={use_token}"
     send_email(
         email_to=email_to,
         subject_template=subject,
         html_template=template_str,
         environment={
-            "project_name": PROJECT_NAME,
-            "username": username,
+            "project_name": config.PROJECT_NAME,
+            "username": email,
             "email": email_to,
-            "valid_hours": EMAIL_RESET_TOKEN_EXPIRE_HOURS,
+            "valid_hours": config.EMAIL_RESET_TOKEN_EXPIRE_HOURS,
             "link": link,
         },
     )
 
 
 def send_new_account_email(email_to: str, username: str, password: str):
-    subject = f"{PROJECT_NAME} - New acccount for user {username}"
-    with open(Path(EMAIL_TEMPLATES_DIR) / "new_account.html") as f:
+    project_name = config.PROJECT_NAME
+    subject = f"{project_name} - New acccount for user {username}"
+    with open(Path(config.EMAIL_TEMPLATES_DIR) / "new_account.html") as f:
         template_str = f.read()
-    link = f"{SERVER_HOST}"
+    link = config.SERVER_HOST
     send_email(
         email_to=email_to,
         subject_template=subject,
         html_template=template_str,
         environment={
-            "project_name": PROJECT_NAME,
+            "project_name": config.PROJECT_NAME,
             "username": username,
             "password": password,
             "email": email_to,
@@ -99,28 +89,23 @@ def send_new_account_email(email_to: str, username: str, password: str):
     )
 
 
-def generate_password_reset_token(username):
-    delta = timedelta(hours=EMAIL_RESET_TOKEN_EXPIRE_HOURS)
+def generate_password_reset_token(email):
+    delta = timedelta(hours=config.EMAIL_RESET_TOKEN_EXPIRE_HOURS)
     now = datetime.utcnow()
     expires = now + delta
     exp = expires.timestamp()
     encoded_jwt = jwt.encode(
-        {
-            "exp": exp,
-            "nbf": now,
-            "sub": password_reset_jwt_subject,
-            "username": username,
-        },
-        SECRET_KEY,
+        {"exp": exp, "nbf": now, "sub": password_reset_jwt_subject, "email": email},
+        config.SECRET_KEY,
         algorithm="HS256",
     )
     return encoded_jwt
 
 
-def verify_password_reset_token(token) -> Union[str, bool]:
+def verify_password_reset_token(token) -> Optional[str]:
     try:
-        decoded_token = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
+        decoded_token = jwt.decode(token, config.SECRET_KEY, algorithms=["HS256"])
         assert decoded_token["sub"] == password_reset_jwt_subject
-        return decoded_token["username"]
+        return decoded_token["email"]
     except InvalidTokenError:
-        return False
+        return None