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