from collections.abc import Iterable from os.path import join import os from pprint import pprint from brevo_python import (ApiClient, Configuration, SendSmtpEmail, TransactionalEmailsApi) from brevo_python.rest import ApiException from jinja2 import Environment, FileSystemLoader from sqlalchemy.future import select from core.config import BASE_PATH, get_db, settings from core.security.exceptions import GenericError from models.user import PasswordResetToken, UserTypesEnum templates_env = Environment( loader=FileSystemLoader(join(BASE_PATH, 'templates')) ) def render_template(template_name: str, context: dict) -> str: template = templates_env.get_template(template_name) return template.render(context) class Email: def __init__(self, *args, **kwargs): configuration = Configuration() configuration.api_key['api-key'] = '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' configuration.api_key['partner-key'] = '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' self.api_instance = TransactionalEmailsApi(ApiClient(configuration)) self.to = self.validate_email_name_format( kwargs.get('to', []) ) self.bcc = self.validate_email_name_format(kwargs.get('bcc', [])) self.cc = self.validate_email_name_format(kwargs.get('cc', [])) self.html_content = kwargs.get('html_content', '') self.text_content = kwargs.get('text_content', '') self.subject = kwargs.get('subject', '') self.reply_to = kwargs.get( 'reply_to', {"email": "noreply@aatoon.com", "name": "WAC"}) if self.reply_to and not self.validate_email_name_dict(self.reply_to): raise ValueError('Invalid format for reply to') self.sender = kwargs.get( 'sender', {"name": "WAC", "email": "noreply@aatoon.com"}) self.attachment = kwargs.get('attachment', '') self.headers = kwargs.get('headers', {}) self.message_versions = kwargs.get('message_versions', {}) self.tags = kwargs.get('tags', []) self.scheduled_at = kwargs.get('scheduled_at') def validate_email_name_dict(self, value: dict) -> bool: if isinstance(value, dict): return set(value.keys()) == {'email', 'name'} return False def is_email_name_format(self, emails: list) -> bool: if isinstance(emails, Iterable): return all( map( self.validate_email_name_dict, emails ) ) return False def validate_email_name_format(self, emails: list): if self.is_email_name_format(emails): return emails raise ValueError('Invalid format for emails') def send_email_kwargs(self) -> dict: return { 'to': self.to, 'bcc': self.bcc, 'cc': self.cc, 'text_content': self.text_content, 'subject': self.subject, 'reply_to': self.reply_to, 'attachment': self.attachment, 'headers': self.headers, 'tags': self.tags, 'sender': self.sender } def remove_unused_email_kwargs(self, email_kwargs: dict) -> dict: if not email_kwargs.get('attachment'): email_kwargs.pop('attachment', None) if not email_kwargs.get('text_content'): email_kwargs.pop('text_content', None) if not email_kwargs.get('cc'): email_kwargs.pop('cc', None) if not email_kwargs.get('bcc'): email_kwargs.pop('bcc', None) if not email_kwargs.get('headers'): email_kwargs.pop('headers', None) if not email_kwargs.get('tags'): email_kwargs.pop('tags', None) def send_email(self, **kwargs) -> int: to = self.validate_email_name_format( kwargs.get('to', self.to) ) bcc = self.validate_email_name_format(kwargs.get('bcc', self.bcc)) cc = self.validate_email_name_format(kwargs.get('cc', self.cc)) reply_to = kwargs.get('reply_to', self.reply_to) if not self.validate_email_name_dict(reply_to): raise ValueError('Invalid format for reply to') email_kwargs = self.send_email_kwargs() email_kwargs.update(kwargs) email_kwargs.update( to=to, bcc=bcc, cc=cc, reply_to=reply_to ) self.remove_unused_email_kwargs(email_kwargs) send_email_object = SendSmtpEmail(**email_kwargs) try: api_response = self.api_instance.send_transac_email( send_email_object ) return api_response except ApiException as e: print("Exception when calling AccountApi->get_account: %s\n" % e) class SendForgotPasswordEmail(Email): def __init__(self, token_instance: PasswordResetToken): with get_db() as db: self.token_instance = token_instance UserTypeModel = UserTypesEnum( value=token_instance.user_type).model result = db.execute( select(UserTypeModel).where( UserTypeModel.id == token_instance.user_id ) ) self.user_instance = None user_object = result.scalar_one_or_none() if user_object is None: raise GenericError(status_code=400, message='User not found') self.user_instance = user_object recipient_object = { 'email': user_object.email, 'name': user_object.first_name } data = { 'to': [recipient_object], 'subject': 'Forgot password email link' } self.html_template = 'reset-password.html' super().__init__(**data) def send_email(self, **kwargs): html_template = kwargs.get('html_template', self.html_template) url = f'{os.getenv("APP_UI_URL")}/password-reset?token={self.token_instance.token}&user_type={self.token_instance.user_type.value}' kwargs.update( html_content=render_template( html_template, context={'url': url} ) ) return super().send_email(**kwargs) class SendWelcomeEmail(Email): def __init__(self, *args, **kwargs): self.html_template = 'welcome.html' super().__init__(*args, **kwargs) def send_email(self, **kwargs): html_template = kwargs.get('html_template', self.html_template) kwargs.update( html_content=render_template( html_template, context={} ) ) return super().send_email(**kwargs) class SendEnrollmentEmail(Email): def __init__(self, *args, **kwargs): self.html_template = 'order_purchase.html' super().__init__(*args, **kwargs) def send_email(self, **kwargs): html_template = kwargs.get('html_template', self.html_template) context = kwargs.pop('context', {}) kwargs.update( html_content=render_template( html_template, context=context ) ) return super().send_email(**kwargs)