58 lines
1.9 KiB
Python
58 lines
1.9 KiB
Python
"""Employee invitation emails sent from the platform domain."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.config import get_settings
|
|
from app.models import Account
|
|
from app.services.domain_provisioning import get_domain_by_name
|
|
from app.services.email_templates import BRAND_NAME, build_employee_invitation_email
|
|
from app.services.mail_sender import deliver_email
|
|
from app.services.member_invitations import (
|
|
INVITE_EXPIRY_HOURS,
|
|
INVITE_FROM_LOCAL,
|
|
create_member_invitation,
|
|
)
|
|
|
|
|
|
async def send_employee_invitation(
|
|
session: AsyncSession,
|
|
*,
|
|
member: Account,
|
|
manager: Account,
|
|
) -> None:
|
|
if not member.invitation_email:
|
|
raise ValueError("Invitation email is required.")
|
|
|
|
settings = get_settings()
|
|
platform_domain = await get_domain_by_name(session, settings.platform_domain)
|
|
if platform_domain is None:
|
|
raise ValueError("Platform mail domain is not configured.")
|
|
|
|
raw_token = await create_member_invitation(session, member)
|
|
invite_url = f"https://{settings.app_public_host}/invite/{raw_token}"
|
|
manager_name = f"{manager.first_name} {manager.last_name}".strip()
|
|
company = member.company_domain or settings.platform_domain
|
|
|
|
subject, body_text, body_html = build_employee_invitation_email(
|
|
member_first_name=member.first_name,
|
|
member_email=member.email,
|
|
manager_name=manager_name,
|
|
company_domain=company,
|
|
invite_url=invite_url,
|
|
expires_hours=INVITE_EXPIRY_HOURS,
|
|
)
|
|
|
|
sender = f"{BRAND_NAME} <{INVITE_FROM_LOCAL}@{settings.platform_domain}>"
|
|
await deliver_email(
|
|
sender=sender,
|
|
recipients=[member.invitation_email],
|
|
cc=[],
|
|
bcc=[],
|
|
subject=subject,
|
|
body_text=body_text,
|
|
body_html=body_html,
|
|
domain=platform_domain,
|
|
reply_to=manager.email,
|
|
)
|