| """ | |
| 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()) | |