webshop/app/email.py

import os
from pathlib import Path
from typing import NoReturn

from fastapi import BackgroundTasks
from fastapi_mail import ConnectionConfig, FastMail, MessageSchema, MessageType
from jinja2 import Environment, FileSystemLoader

from app.parser import read_file


def prepare_email(
    git_repo: str, document_match: list[str], block_types: list[str], order
) -> dict[str]:
    """Prepare email message."""
    filename = "checkout/email"
    doc = read_file(
        git_repo,
        filename,
        document_match,
        block_types,
        "",
    )

    # -- setup jinja template and expand data in it
    environment = Environment(loader=FileSystemLoader(f"{Path(git_repo) / 'checkout'}"))
    template = environment.get_template("email.md")
    body = template.render(order)

    # clean up data
    block_marker = "\n---\n"
    blocks = body.split(block_marker)
    blocks = [block.strip() for block in blocks]

    email_body = "\n\n---\n\n".join(blocks[1:])

    return {
        "sender": doc.meta.sender,
        "subject": doc.meta.title,
        "body": email_body,
    }


async def send_email(
    settings: dict[str, list[str]],
    order: dict[str],
    background_tasks: BackgroundTasks | None = None,
) -> NoReturn | tuple[FastMail, MessageSchema]:
    """Send email to given customer. Async process."""

    email_data = prepare_email(
        settings["git_repo"], settings["document_match"], settings["block_types"], order
    )

    conf = ConnectionConfig(
        MAIL_USERNAME=os.getenv("MAIL_USERNAME"),
        MAIL_PASSWORD=os.getenv("MAIL_PASSWORD"),
        MAIL_FROM=os.getenv("MAIL_FROM"),
        MAIL_FROM_NAME=email_data["sender"],
        MAIL_PORT=os.getenv("MAIL_PORT"),
        MAIL_SERVER=os.getenv("MAIL_SERVER"),
        MAIL_STARTTLS=os.getenv("MAIL_STARTTLS"),
        MAIL_SSL_TLS=os.getenv("MAIL_SSL_TLS"),
        USE_CREDENTIALS=os.getenv("MAIL_USE_CREDENTIALS"),
        VALIDATE_CERTS=os.getenv("MAIL_VALIDATE_CERTS"),
    )

    recipients = []
    if background_tasks:
        recipients.append(order['email'])
    else:
        recipients.append(os.getenv("MAIL_TO_TEST"))

    message = MessageSchema(
        subject=email_data["subject"],
        recipients=recipients,
        body=email_data["body"],
        subtype=MessageType.plain,
        bcc=[os.getenv("MAIL_FROM")]
    )

    fm = FastMail(conf)

    if background_tasks:
        background_tasks.add_task(fm.send_message, message)
    else:
        return (fm, message)