aboutsummaryrefslogtreecommitdiff
path: root/.venv/lib/python3.12/site-packages/core/base/providers/email.py
blob: 73f88162b638083e5dbcb567cfea18807a4f2b75 (about) (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
import logging
import os
from abc import ABC, abstractmethod
from typing import Optional

from .base import Provider, ProviderConfig


class EmailConfig(ProviderConfig):
    smtp_server: Optional[str] = None
    smtp_port: Optional[int] = None
    smtp_username: Optional[str] = None
    smtp_password: Optional[str] = None
    from_email: Optional[str] = None
    use_tls: Optional[bool] = True
    sendgrid_api_key: Optional[str] = None
    mailersend_api_key: Optional[str] = None
    verify_email_template_id: Optional[str] = None
    reset_password_template_id: Optional[str] = None
    password_changed_template_id: Optional[str] = None
    frontend_url: Optional[str] = None
    sender_name: Optional[str] = None

    @property
    def supported_providers(self) -> list[str]:
        return [
            "smtp",
            "console",
            "sendgrid",
            "mailersend",
        ]  # Could add more providers like AWS SES, SendGrid etc.

    def validate_config(self) -> None:
        if (
            self.provider == "sendgrid"
            and not self.sendgrid_api_key
            and not os.getenv("SENDGRID_API_KEY")
        ):
            raise ValueError(
                "SendGrid API key is required when using SendGrid provider"
            )

        if (
            self.provider == "mailersend"
            and not self.mailersend_api_key
            and not os.getenv("MAILERSEND_API_KEY")
        ):
            raise ValueError(
                "MailerSend API key is required when using MailerSend provider"
            )


logger = logging.getLogger(__name__)


class EmailProvider(Provider, ABC):
    def __init__(self, config: EmailConfig):
        if not isinstance(config, EmailConfig):
            raise ValueError(
                "EmailProvider must be initialized with an EmailConfig"
            )
        super().__init__(config)
        self.config: EmailConfig = config

    @abstractmethod
    async def send_email(
        self,
        to_email: str,
        subject: str,
        body: str,
        html_body: Optional[str] = None,
        *args,
        **kwargs,
    ) -> None:
        pass

    @abstractmethod
    async def send_verification_email(
        self, to_email: str, verification_code: str, *args, **kwargs
    ) -> None:
        pass

    @abstractmethod
    async def send_password_reset_email(
        self, to_email: str, reset_token: str, *args, **kwargs
    ) -> None:
        pass

    @abstractmethod
    async def send_password_changed_email(
        self,
        to_email: str,
        *args,
        **kwargs,
    ) -> None:
        pass