import smtplib from datetime import datetime from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText from email.utils import formatdate from email.utils import make_msgid from onyx.configs.app_configs import EMAIL_CONFIGURED from onyx.configs.app_configs import EMAIL_FROM from onyx.configs.app_configs import SMTP_PASS from onyx.configs.app_configs import SMTP_PORT from onyx.configs.app_configs import SMTP_SERVER from onyx.configs.app_configs import SMTP_USER from onyx.configs.app_configs import WEB_DOMAIN from onyx.configs.constants import AuthType from onyx.configs.constants import TENANT_ID_COOKIE_NAME from onyx.db.models import User HTML_EMAIL_TEMPLATE = """\ {title} """ def build_html_email( heading: str, message: str, cta_text: str | None = None, cta_link: str | None = None ) -> str: if cta_text and cta_link: cta_block = f'{cta_text}' else: cta_block = "" return HTML_EMAIL_TEMPLATE.format( title=heading, heading=heading, message=message, cta_block=cta_block, year=datetime.now().year, ) def send_email( user_email: str, subject: str, html_body: str, text_body: str, mail_from: str = EMAIL_FROM, ) -> None: if not EMAIL_CONFIGURED: raise ValueError("Email is not configured.") msg = MIMEMultipart("alternative") msg["Subject"] = subject msg["To"] = user_email msg["From"] = mail_from msg["Date"] = formatdate(localtime=True) msg["Message-ID"] = make_msgid(domain="onyx.app") part_text = MIMEText(text_body, "plain") part_html = MIMEText(html_body, "html") msg.attach(part_text) msg.attach(part_html) try: with smtplib.SMTP(SMTP_SERVER, SMTP_PORT) as s: s.starttls() s.login(SMTP_USER, SMTP_PASS) s.send_message(msg) except Exception as e: raise e def send_subscription_cancellation_email(user_email: str) -> None: # Example usage of the reusable HTML subject = "Your Onyx Subscription Has Been Canceled" heading = "Subscription Canceled" message = ( "

We're sorry to see you go.

" "

Your subscription has been canceled and will end on your next billing date.

" "

If you change your mind, you can always come back!

" ) cta_text = "Renew Subscription" cta_link = "https://www.onyx.app/pricing" html_content = build_html_email(heading, message, cta_text, cta_link) text_content = ( "We're sorry to see you go.\n" "Your subscription has been canceled and will end on your next billing date.\n" "If you change your mind, visit https://www.onyx.app/pricing" ) send_email(user_email, subject, html_content, text_content) def send_user_email_invite( user_email: str, current_user: User, auth_type: AuthType ) -> None: subject = "Invitation to Join Onyx Organization" heading = "You've Been Invited!" # the exact action taken by the user, and thus the message, depends on the auth type message = f"

You have been invited by {current_user.email} to join an organization on Onyx.

" if auth_type == AuthType.CLOUD: message += ( "

To join the organization, please click the button below to set a password " "or login with Google and complete your registration.

" ) elif auth_type == AuthType.BASIC: message += ( "

To join the organization, please click the button below to set a password " "and complete your registration.

" ) elif auth_type == AuthType.GOOGLE_OAUTH: message += ( "

To join the organization, please click the button below to login with Google " "and complete your registration.

" ) elif auth_type == AuthType.OIDC or auth_type == AuthType.SAML: message += ( "

To join the organization, please click the button below to" " complete your registration.

" ) else: raise ValueError(f"Invalid auth type: {auth_type}") cta_text = "Join Organization" cta_link = f"{WEB_DOMAIN}/auth/signup?email={user_email}" html_content = build_html_email(heading, message, cta_text, cta_link) # text content is the fallback for clients that don't support HTML # not as critical, so not having special cases for each auth type text_content = ( f"You have been invited by {current_user.email} to join an organization on Onyx.\n" "To join the organization, please visit the following link:\n" f"{WEB_DOMAIN}/auth/signup?email={user_email}\n" ) if auth_type == AuthType.CLOUD: text_content += "You'll be asked to set a password or login with Google to complete your registration." send_email(user_email, subject, html_content, text_content) def send_forgot_password_email( user_email: str, token: str, mail_from: str = EMAIL_FROM, tenant_id: str | None = None, ) -> None: # Builds a forgot password email with or without fancy HTML subject = "Onyx Forgot Password" link = f"{WEB_DOMAIN}/auth/reset-password?token={token}" if tenant_id: link += f"&{TENANT_ID_COOKIE_NAME}={tenant_id}" message = f"

Click the following link to reset your password:

{link}

" html_content = build_html_email("Reset Your Password", message) text_content = f"Click the following link to reset your password: {link}" send_email(user_email, subject, html_content, text_content, mail_from) def send_user_verification_email( user_email: str, token: str, mail_from: str = EMAIL_FROM, ) -> None: # Builds a verification email subject = "Onyx Email Verification" link = f"{WEB_DOMAIN}/auth/verify-email?token={token}" message = ( f"

Click the following link to verify your email address:

{link}

" ) html_content = build_html_email("Verify Your Email", message) text_content = f"Click the following link to verify your email address: {link}" send_email(user_email, subject, html_content, text_content, mail_from)