spark / cbh /api /security /utils.py
brestok's picture
fix
e2a8a40
"""
Security utilities module.
"""
import asyncio
from fastapi import HTTPException
from jinja2 import Environment, FileSystemLoader, select_autoescape
from cbh.api.account.models import AccountModel, AccountShorten
from cbh.api.security.models import VerificationCodeModel
from cbh.core.config import settings
async def send_password_reset_email(
code: str, account_obj: AccountShorten
) -> None:
"""
Send a password reset email.
"""
templates_path = settings.BASE_DIR / "templates"
env = Environment(
loader=FileSystemLoader(templates_path),
autoescape=select_autoescape(["html", "xml"]),
)
template = env.get_template("resetPassword.html")
link = f"{settings.Audience}/reset?code={code}"
template_content = template.render(
link=link,
name=account_obj.name,
)
await settings.EMAIL_CLIENT.send_email(
account_obj.email,
"You requested a password reset in Arena",
template_content,
)
def _prepare_joins_str(org_accounts: list[AccountShorten]):
if not org_accounts:
return None
if len(org_accounts) == 1:
return f"{org_accounts[0].name} has joined"
elif len(org_accounts) == 2:
return f"{org_accounts[0].name} and {org_accounts[1].name} have joined"
return f"{org_accounts[0].name} and {len(org_accounts) - 1} others have joined"
async def send_invite_email(
admin_account: AccountModel,
code: VerificationCodeModel,
org_accounts: list[AccountShorten],
email: str,
message: str | None = None,
) -> None:
"""
Send an invite email.
"""
templates_path = settings.BASE_DIR / "cbh" / "templates" / "emails"
env = Environment(
loader=FileSystemLoader(templates_path),
autoescape=select_autoescape(["html", "xml"]),
)
template = env.get_template("inviteToOrg.html")
icons = [
org_account.pictureUrl for org_account in org_accounts if org_account.pictureUrl
][:3]
joins_str = _prepare_joins_str(org_accounts)
template_content = template.render(
admin_name=admin_account.name,
organization_name=admin_account.organization.name,
link=f"{settings.Audience}/signup?code={code.id}",
icons=icons if icons else None,
joins_str=joins_str,
audience_link=f"{admin_account.organization.slug}.{settings.Domain}",
message=message,
)
await settings.EMAIL_CLIENT.send_email(
email,
f"{admin_account.name.title()} invited you to train with them in Arena",
template_content,
)
async def delete_account():
await settings.DB_CLIENT.accounts.delete_one(
{"email": "maksimshymanouski@gmail.com"}
)
if __name__ == "__main__":
asyncio.run(delete_account())