Use pydantic for validations

Use enum wrapper for options
Update README.md, requirements.txt and Dockerfile
Release alpha version

Dockerfile

@@ -2,10 +2,9 @@ FROM python:3.9-slim
 
 RUN mkdir /opt/temp
 COPY . /opt/temp
+WORKDIR /opt/temp
 
 RUN /usr/local/bin/python3 -m pip install --upgrade pip
-RUN pip install --user gmail-connector
-
-WORKDIR /opt/temp
+RUN pip install --user .
 
 ENTRYPOINT ["/usr/local/bin/python", "./test_runner.py"]

README.md

@@ -27,20 +27,29 @@ pip install gmail-connector
 ## Env Vars
 Environment variables can be loaded from any `.env` file.
 ```bash
-# For authentication
 GMAIL_USER='[email protected]',
 GMAIL_PASS='<ACCOUNT_PASSWORD>'
-
-# For outbound SMS
-PHONE='1234567890'
 ```
 
-*Optionally `.env` files can also be scanned for:*
+<details>
+<summary><strong>Env variable customization</strong></summary>
+
+To load a custom `.env` file, set the filename as the env var `env_file` before importing `gmailconnector`
 ```python
+import os
+os.environ['env_file'] = 'custom'  # to load a custom .env file
 import gmailconnector as gc
-
-gc.load_env(scan=True)
 ```
+To avoid using env variables, arguments can be loaded during object instantiation.
+```python
+import gmailconnector as gc
+kwargs = dict(gmail_user='EMAIL_ADDRESS',
+              gmail_pass='PASSWORD',
+              encryption=gc.Encryption.SSL,
+              timeout=5)
+email_obj = gc.SendEmail(**kwargs)
+```
+</details>
 
 ## Usage
 ### [Send SMS](https://github.com/thevickypedia/gmail-connector/blob/master/gmailconnector/send_sms.py)

gmailconnector/__init__.py

@@ -1,11 +1,7 @@
 """Place holder for package."""
 
-import os
-from typing import Union
-
-import dotenv
-
-from .models.config import Encryption, SMSGateway  # noqa: F401
+from .models.config import (EgressConfig, Encryption,  # noqa: F401
+                            IngressConfig, SMSGateway)
 from .models.options import Category, Condition, Folder  # noqa: F401
 from .models.responder import Response  # noqa: F401
 from .read_email import ReadEmail  # noqa: F401
@@ -15,18 +11,4 @@
 from .validator.address import EmailAddress  # noqa: F401
 from .validator.validate_email import validate_email  # noqa: F401
 
-version = "0.9.1"
-
-
-def load_env(filename: Union[str, os.PathLike] = ".env", scan: bool = False) -> None:
-    """Load .env files."""
-    if scan:
-        for file in os.listdir():
-            if os.path.isfile(file) and file.endswith(".env"):
-                dotenv.load_dotenv(dotenv_path=file, verbose=False)
-    else:
-        if os.path.isfile(filename):
-            dotenv.load_dotenv(dotenv_path=filename, verbose=False)
-
-
-load_env()
+version = "1.0a"

gmailconnector/lib/requirements.txt

@@ -1,4 +1,5 @@
-idna>=3.3
-python-dotenv>=0.21.0
-pytz>=2021.3
-dnspython==2.3.0
+idna==3.*
+pytz==2023.*
+dnspython==2.4.2
+pydantic[email]==2.4.*
+pydantic_settings==2.0.*

gmailconnector/models/config.py

@@ -1,15 +1,30 @@
+import os
 from enum import Enum
+from typing import Union
+
+from pydantic import BaseModel, EmailStr, Field
+from pydantic_settings import BaseSettings
+
+from .options import Folder
 
 
 class Encryption(str, Enum):
-    """Enum wrapper for TLS and SSL encryption."""
+    """Enum wrapper for TLS and SSL encryption.
+
+    >>> Encryption
+
+    """
 
     TLS: str = "TLS"
     SSL: str = "SSL"
 
 
-class SMSGateway:
-    """Enum wrapper for different SMS gateways."""
+class SMSGatewayModel(BaseModel):
+    """Wrapper for SMS gateways.
+
+    >>> SMSGatewayModel
+
+    """
 
     att: str = "mms.att.net"
     tmobile: str = "tmomail.net"
@@ -18,3 +33,50 @@ class SMSGateway:
     cricket: str = "sms.cricketwireless.net"
     uscellular: str = "email.uscc.net"
     all: tuple = (att, tmobile, verizon, boost, cricket, uscellular)
+
+
+SMSGateway = SMSGatewayModel()
+
+
+class EgressConfig(BaseSettings):
+    """Configure arguments for ``SendEmail``/``SendSMS`` and validate using ``pydantic`` to share across modules.
+
+    >>> EgressConfig
+
+    """
+
+    gmail_user: EmailStr
+    gmail_pass: str
+    recipient: Union[EmailStr, None] = None
+    phone: Union[str, None] = Field(None, pattern="\\d{10}$")
+    gmail_host: str = "smtp.gmail.com"
+    encryption: Encryption = Encryption.TLS
+    timeout: int = 10
+
+    class Config:
+        """Environment variables configuration."""
+
+        env_prefix = ""
+        env_file = os.environ.get("env_file", os.environ.get("ENV_FILE", ".env"))
+        extra = "allow"
+
+
+class IngressConfig(BaseSettings):
+    """Configure arguments for ``ReadEmail`` and validate using ``pydantic`` to share across modules.
+
+    >>> IngressConfig
+
+    """
+
+    gmail_user: EmailStr
+    gmail_pass: str
+    folder: Folder = Folder.inbox
+    gmail_host: str = "imap.gmail.com"
+    timeout: Union[int, float] = 10
+
+    class Config:
+        """Environment variables configuration."""
+
+        env_prefix = ""
+        env_file = os.environ.get("env_file", os.environ.get("ENV_FILE", ".env"))
+        extra = "allow"

gmailconnector/models/options.py

@@ -1,6 +1,7 @@
 """Options that can be included while reading emails."""
 
 import datetime
+from enum import Enum
 from typing import Union
 
 
@@ -30,7 +31,7 @@ def subject(subject: str):
         return 'SUBJECT "%s"' % subject
 
 
-class Folder:
+class Folder(str, Enum):
     """Wrapper for folders to choose emails from."""
 
     inbox: str = "inbox"

gmailconnector/read_email.py

@@ -1,15 +1,16 @@
 import email
 import imaplib
-import os
 import socket
 from collections.abc import Generator
 from datetime import datetime, timedelta, timezone
 from email.header import decode_header, make_header
 from typing import Iterable, Union
 
 import pytz
+from typing_extensions import Unpack
 
-from .models.options import Category, Condition, Folder
+from .models.config import IngressConfig
+from .models.options import Category, Condition
 from .models.responder import Email, Response
 
 
@@ -22,13 +23,14 @@ class ReadEmail:
 
     LOCAL_TIMEZONE = datetime.now(timezone.utc).astimezone().tzinfo
 
-    def __init__(self, gmail_user: str = None, gmail_pass: str = None, folder: Folder.__str__ = Folder.inbox,
-                 gmail_host: str = "imap.gmail.com", timeout: Union[int, float] = 10):
-        """Initiates all the necessary args.
+    def __init__(self, **kwargs: Unpack[IngressConfig]):
+        """Initiates necessary args, creates a connection with Gmail host to read emails from the chosen folder.
 
-        Args:
-            gmail_user: Gmail username to authenticate IMAP lib.
-            gmail_pass: Gmail password to authenticate IMAP lib.
+        kwargs:
+            gmail_user: Gmail username to authenticate SMTP lib.
+            gmail_pass: Gmail password to authenticate SMTP lib.
+            timeout: Connection timeout for SMTP lib.
+            gmail_host: Hostname for gmail's smtp server.
             folder: Folder where the emails have to be read from.
 
         References:
@@ -37,17 +39,11 @@ def __init__(self, gmail_user: str = None, gmail_pass: str = None, folder: Folde
         See Also:
             Uses broad ``Exception`` clause to catch login errors, since the same is raised by ``imaplib``
         """
-        gmail_user = gmail_user or os.environ.get('gmail_user') or os.environ.get('GMAIL_USER')
-        gmail_pass = gmail_pass or os.environ.get('gmail_pass') or os.environ.get('GMAIL_PASS')
-        if not all([gmail_user, gmail_pass]):
-            raise ValueError("'gmail_user' and 'gmail_pass' are mandatory")
-        self.folder = folder
-        self.gmail_user = gmail_user
-        self.gmail_pass = gmail_pass
         self.error = None
         self.mail = None
         self._authenticated = False
-        self.create_ssl_connection(gmail_host=gmail_host, timeout=timeout)
+        self.env = IngressConfig(**kwargs)
+        self.create_ssl_connection(gmail_host=self.env.gmail_host, timeout=self.env.timeout)
 
     def create_ssl_connection(self, gmail_host: str, timeout: Union[int, float]) -> None:
         """Creates an SSL connection to gmail's SSL server."""
@@ -71,9 +67,9 @@ def authenticate(self):
                 'body': self.error or "failed to create a connection with gmail's SMTP server"
             })
         try:
-            self.mail.login(user=self.gmail_user, password=self.gmail_pass)
+            self.mail.login(user=self.env.gmail_user, password=self.env.gmail_pass)
             self.mail.list()  # list all the folders within your mailbox (like inbox, sent, drafts, etc)
-            self.mail.select(self.folder)
+            self.mail.select(self.env.folder)
             self._authenticated = True
             return Response(dictionary={
                 'ok': True,
@@ -121,7 +117,8 @@ def instantiate(self,
             return Response(dictionary={
                 'ok': False,
                 'status': 204,
-                'body': f'No {filters.lower()!r} emails found for {self.gmail_user} in {self.folder}',
+                'body': f'No emails found in {self.env.gmail_user} [{self.env.folder}] '
+                        f'for the filter(s) {filters.lower()!r}',
                 'count': num
             })
 
@@ -172,12 +169,15 @@ def get_info(self, response_part: tuple, dt_flag: bool) -> Email:
                 receive = local_time.strftime("on %A, %B %d, at %I:%M %p")
         else:
             receive = local_time
-        body = None
         if original_email.get_content_type() == "text/plain":  # ignore attachments and html
             body = original_email.get_payload(decode=True)
             body = body.decode('utf-8')
-        return Email(dictionary=dict(sender=from_[0], sender_email=from_[1].rstrip('>'), subject=sub,
-                                     date_time=receive, body=body))
+        else:
+            body = ""
+            for payload in original_email.get_payload():
+                body += payload.as_string()
+        return Email(dictionary=dict(sender=from_[0], sender_email=from_[1].rstrip('>'),
+                                     subject=sub, date_time=receive, body=body))
 
     def read_mail(self, messages: list or str, humanize_datetime: bool = False) -> Generator[Email]:
         """Yield emails matching the filters' criteria.

gmailconnector/send_email.py

@@ -6,7 +6,9 @@
 from email.mime.text import MIMEText
 from typing import Dict, Union
 
-from .models.config import Encryption
+from typing_extensions import Unpack
+
+from .models.config import EgressConfig, Encryption
 from .models.responder import Response
 from .validator.address import EmailAddress
 
@@ -26,37 +28,24 @@ class SendEmail:
 
     """
 
-    def __init__(self, gmail_user: str = None, gmail_pass: str = None, timeout: Union[int, float] = 10,
-                 gmail_host: str = "smtp.gmail.com", encryption: Encryption.__str__ = Encryption.TLS):
-        """Initiates necessary args, creates a connection with Gmail's SMTP on port 587.
+    def __init__(self, **kwargs: Unpack[EgressConfig]):
+        """Initiates necessary args, creates a connection with Gmail host based on chosen encryption type.
 
-        Args:
+        kwargs:
             gmail_user: Gmail username to authenticate SMTP lib.
             gmail_pass: Gmail password to authenticate SMTP lib.
             timeout: Connection timeout for SMTP lib.
             encryption: Type of encryption to be used.
             gmail_host: Hostname for gmail's smtp server.
         """
-        gmail_user = gmail_user or os.environ.get('gmail_user') or os.environ.get('GMAIL_USER')
-        gmail_pass = gmail_pass or os.environ.get('gmail_pass') or os.environ.get('GMAIL_PASS')
         self.server, self.error = None, None
-        if not all([gmail_user, gmail_pass]):
-            raise ValueError("'gmail_user' and 'gmail_pass' are mandatory")
-        if encryption not in (Encryption.TLS, Encryption.SSL):
-            raise ValueError(
-                'Encryption should either be TLS or SSL'
-            )
-        if gmail_user.endswith('@gmail.com'):
-            self.gmail_user = gmail_user
-        else:
-            self.gmail_user = gmail_user + '@gmail.com'
-        self.gmail_pass = gmail_pass
+        self.env = EgressConfig(**kwargs)
         self._failed_attachments = {"FILE NOT FOUND": [], "FILE SIZE OVER 25 MB": []}
         self._authenticated = False
-        if encryption == Encryption.TLS:
-            self.create_tls_connection(host=gmail_host, timeout=timeout)
+        if self.env.encryption == Encryption.TLS:
+            self.create_tls_connection(host=self.env.gmail_host, timeout=self.env.timeout)
         else:
-            self.create_ssl_connection(host=gmail_host, timeout=timeout)
+            self.create_ssl_connection(host=self.env.gmail_host, timeout=self.env.timeout)
 
     def create_ssl_connection(self, host: str, timeout: Union[int, float]) -> None:
         """Create a connection using SSL encryption."""
@@ -88,7 +77,7 @@ def authenticate(self) -> Response:
                 'body': self.error or "failed to create a connection with gmail's SMTP server"
             })
         try:
-            self.server.login(user=self.gmail_user, password=self.gmail_pass)
+            self.server.login(user=self.env.gmail_user, password=self.env.gmail_pass)
             self._authenticated = True
             return Response(dictionary={
                 'ok': True,
@@ -138,7 +127,7 @@ def multipart_message(self, subject: str, recipient: str or list, sender: str, b
 
         msg = MIMEMultipart()
         msg['Subject'] = subject
-        msg['From'] = f"{sender} <{self.gmail_user}>"
+        msg['From'] = f"{sender} <{self.env.gmail_user}>"
         msg['To'] = ','.join(recipient)
         if cc:
             msg['Cc'] = ','.join(cc)