Add email templates

Author: drmacsika

__pycache__/main.cpython-39.pyc

@@ -1,13 +1,65 @@
-from fastapi import APIRouter
+from typing import Any, List
 
-from accounts.models import UserIn_Pydantic, UserOut_Pydantic
+from core.dependencies import (get_current_active_superuser,
+                               get_current_active_user, get_session)
+from fastapi import APIRouter, Body, Depends
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from accounts.crud import user
+from accounts.models import User
+from accounts.schemas import UserCreate, UserUpdate
 
 router = APIRouter()
 
 
+@router.get("/users/", response_model=List[User])
+async def get_all_users(
+    db: AsyncSession = Depends(get_session),
+    limit: int = 100, offset: int = 0,
+    current_user: User = Depends(get_current_active_superuser)) -> Any:
+    """Return all users."""
+    return await user.get_multiple(db=db, offset=offset, limit=limit)
+    
+
+@router.post("/users/", response_model=User, status_code=201)
+async def create_user(
+    user_in: UserCreate,
+    db: AsyncSession = Depends(get_session),
+    current_user: User = Depends(get_current_active_superuser)) -> Any:
+    """
+    Create new user.
+    """
+    return await user.create(obj_in=user_in, db=db)
 
+    
 
-@router.post("/signup/", response_model=UserOut_Pydantic)
-async def signup(user: UserIn_Pydantic):
+@router.put("/me", response_model=User)
+def update_user_me(
+    *,
+    db: AsyncSession = Depends(get_session),
+    password: str = Body(None),
+    full_name: str = Body(None),
+    email: EmailStr = Body(None),
+    current_user: User = Depends(get_current_active_user),
+) -> Any:
+    """
+    Update own user.
+    """
+    current_user_data = jsonable_encoder(current_user)
+    user_in = schemas.UserUpdate(**current_user_data)
+    if password is not None:
+        user_in.password = password
+    if full_name is not None:
+        user_in.full_name = full_name
+    if email is not None:
+        user_in.email = email
+    user = crud.user.update(db, db_obj=current_user, obj_in=user_in)
+    return user
+    
+    
 
-    return {"User": "Name"}
+    
+    
+    
+    
+

accounts/main.py

@@ -1,13 +1,14 @@
 import secrets
 from functools import lru_cache
+from pathlib import Path
 from typing import Any, Dict, List, Optional, Union
 
 from pydantic import (AnyHttpUrl, BaseSettings, EmailStr, HttpUrl, PostgresDsn,
                       validator)
 
 
 class Settings(BaseSettings):
-    PROJECT_NAME: str = "Archangel Macsika"
+    PROJECT_NAME: str = "Tech Blog"
     API_V1_STR: str = "/api/v1"
     SECRET_KEY: str = secrets.token_urlsafe(32)
     # 60 minutes * 24 hours * 8 days = 8 days
@@ -57,12 +58,13 @@ class Settings(BaseSettings):
     EMAILS_FROM_EMAIL: Optional[EmailStr] = None
     EMAILS_FROM_NAME: Optional[str] = None
 
-    @validator("EMAILS_FROM_NAME")
+    @validator("EMAILS_FROM_NAME", allow_reuse=True)
     def get_project_name(cls, v: Optional[str], values: Dict[str, Any]) -> str:
         if not v:
             return values["PROJECT_NAME"]
         return v
 
+    
     EMAIL_RESET_TOKEN_EXPIRE_HOURS: int = 48
     EMAIL_TEMPLATES_DIR: str = "/templates/build"
     EMAILS_ENABLED: bool = False
@@ -76,8 +78,8 @@ def get_emails_enabled(cls, v: bool, values: Dict[str, Any]) -> bool:
         )
 
     EMAIL_TEST_USER: EmailStr = "test@example.com"  # type: ignore
-    FIRST_SUPERUSER: EmailStr
-    FIRST_SUPERUSER_PASSWORD: str
+    FIRST_SUPERUSER: EmailStr = "admin@example.com"
+    FIRST_SUPERUSER_PASSWORD: str = "admin"
     USERS_OPEN_REGISTRATION: bool = False
 
 

blog/__pycache__/main.cpython-39.pyc

@@ -1,13 +1,17 @@
 import base64
+import logging
 import math
 import random
 import re
 import string
 import unicodedata
 from datetime import datetime, timedelta
-from typing import Any, Union
+from pathlib import Path
+from typing import Any, Dict, Optional, Union
 
+import emails
 import pyotp
+from emails.template import JinjaTemplate
 from fastapi import Depends, HTTPException
 from jose import jwt
 from passlib.context import CryptContext
@@ -44,6 +48,108 @@ def get_password_hash(password: str) -> str:
     return pwd_context.hash(password)
 
 
+def send_email(
+    email_to: str,
+    subject_template: str = "",
+    html_template: str = "",
+    environment: Dict[str, Any] = {},
+) -> None:
+    """Send an email"""
+    assert settings.EMAILS_ENABLED, "no provided configuration for email variables"
+    message = emails.Message(
+        subject=JinjaTemplate(subject_template),
+        html=JinjaTemplate(html_template),
+        mail_from=(settings.EMAILS_FROM_NAME, settings.EMAILS_FROM_EMAIL),
+    )
+    smtp_options = {"host": settings.SMTP_HOST, "port": settings.SMTP_PORT}
+    if settings.SMTP_TLS:
+        smtp_options["tls"] = True
+    if settings.SMTP_USER:
+        smtp_options["user"] = settings.SMTP_USER
+    if settings.SMTP_PASSWORD:
+        smtp_options["password"] = settings.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) -> None:
+    """Send test emails"""
+    project_name = settings.PROJECT_NAME
+    subject = f"{project_name} - Test email"
+    with open(Path(settings.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": settings.PROJECT_NAME, "email": email_to},
+    )
+
+
+def send_reset_password_email(email_to: str, email: str, token: str) -> None:
+    """Send email for password reset"""
+    project_name = settings.PROJECT_NAME
+    subject = f"{project_name} - Password recovery for user {email}"
+    with open(Path(settings.EMAIL_TEMPLATES_DIR) / "reset_password.html") as f:
+        template_str = f.read()
+    server_host = settings.SERVER_HOST
+    link = f"{server_host}/reset-password?token={token}"
+    send_email(
+        email_to=email_to,
+        subject_template=subject,
+        html_template=template_str,
+        environment={
+            "project_name": settings.PROJECT_NAME,
+            "username": email,
+            "email": email_to,
+            "valid_hours": settings.EMAIL_RESET_TOKEN_EXPIRE_HOURS,
+            "link": link,
+        },
+    )
+
+
+def send_new_account_email(email_to: str, username: str, password: str) -> None:
+    """Send email for new user account registration"""
+    project_name = settings.PROJECT_NAME
+    subject = f"{project_name} - New account for user {username}"
+    with open(Path(settings.EMAIL_TEMPLATES_DIR) / "new_account.html") as f:
+        template_str = f.read()
+    link = settings.SERVER_HOST
+    send_email(
+        email_to=email_to,
+        subject_template=subject,
+        html_template=template_str,
+        environment={
+            "project_name": settings.PROJECT_NAME,
+            "username": username,
+            "password": password,
+            "email": email_to,
+            "link": link,
+        },
+    )
+
+
+def generate_password_reset_token(email: str) -> str:
+    """Generate new password reset token"""
+    delta = timedelta(hours=settings.EMAIL_RESET_TOKEN_EXPIRE_HOURS)
+    now = datetime.utcnow()
+    expires = now + delta
+    exp = expires.timestamp()
+    encoded_jwt = jwt.encode(
+        {"exp": exp, "nbf": now, "sub": email}, settings.SECRET_KEY, algorithm="HS256",
+    )
+    return encoded_jwt
+
+
+def verify_password_reset_token(token: str) -> Optional[str]:
+    """Verify the password reset token"""
+    try:
+        decoded_token = jwt.decode(token, settings.SECRET_KEY, algorithms=["HS256"])
+        return decoded_token["email"]
+    except jwt.JWTError:
+        return None
+
+
 async def check_existing_row_by_slug(cls, slug: str, db: AsyncSession, status_code: int = None, msg: str = None, **kwargs) -> Any:
     """
     Precheck function for checking the existence of a row using a slug
@@ -64,17 +170,22 @@ async def check_existing_row_by_slug(cls, slug: str, db: AsyncSession, status_co
 
 
 def generate_hotp(email: str, settings: settings = Depends(settings)):
+    """
+    If token genaration is otp based, this generates an 6-digit OTP.
+    """
     keygen = email+str(datetime.date(datetime.now()))+settings.SECRET_KEY
     key = base64.b32encode(keygen.encode())
     return pyotp.HOTP(key)
 
 
 def create_otp(email: str, counter: int):
+    """This creates an OTP."""
     hotp = generate_hotp(email)
     return hotp.at(counter)
 
 
 def verify_otp(email: str, token: str, counter: int):
+    """This verifies an otp."""
     hotp = generate_hotp(email)
     return hotp.verify(token, counter) # => True