Integrations Module

• Standardization: Provides a consistent approach to configuring and managing external systems.

• Immutable Configurations: Uses secure, immutable configuration containers.

• Common Interfaces: Defines standard methods for all integration implementations.

• Adapters: Converts backend DTOs into integration instances.

• Base Classes & Implementations: Offers reusable base functionality and specific service implementations.

1. Integration Configuration

The IntegrationConfig is a frozen dataclass that stores configuration settings for integrations.

File: autoppia_sdk/src/integrations/config.py

pythonCopiarfrom dataclasses import dataclass
from typing import Dict, Any

@dataclass(frozen=True)
class IntegrationConfig:
    """Immutable configuration container for integration instances.
    
    Attributes:
        name (str): The unique identifier or name of the integration.
        category (str): The category or type of integration (e.g., 'database', 'messaging', 'storage', etc.).
        attributes (Dict[str, Any]): A dictionary containing configuration parameters specific to the integration.
    """
    name: str
    category: str
    attributes: Dict[str, Any]

---

2. Integration Interface and Base Class

The IntegrationInterface defines the required methods for integrations, while the Integration base class implements the interface.

File: autoppia_sdk/src/integrations/interface.py

pythonCopiarclass IntegrationInterface:
    """Interface that all integration implementations must adhere to."""
    def send(self, data):
        raise NotImplementedError

    def receive(self):
        raise NotImplementedError

File: autoppia_sdk/src/integrations/implementations/base.py

pythonCopiarfrom autoppia_sdk.src.integrations.interface import IntegrationInterface

class Integration(IntegrationInterface):
    """Base Integration class that implements the IntegrationInterface.
    
    This class serves as a base implementation for all integration types in the Autoppia SDK.
    """
    pass

---

3. SMTP Email Integration

The SDK includes an example implementation for email functionality using SMTP (for sending emails) and IMAP (for receiving emails).

File: autoppia_sdk/src/integrations/implementations/email/smtp_integration.py

pythonCopiarimport email
import imaplib
import smtplib
from email import encoders
from email.mime.base import MIMEBase
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from typing import Dict, List, Optional

from autoppia_sdk.src.integrations.implementations.email.interface import EmailIntegration
from autoppia_sdk.src.integrations.config import IntegrationConfig
from autoppia_sdk.src.integrations.implementations.base import Integration

class SMPTEmailIntegration(EmailIntegration, Integration):
    """SMTP-based email integration for sending and receiving emails.
    
    Attributes:
        integration_config (IntegrationConfig): Configuration object containing email settings.
        smtp_server (str): SMTP server hostname.
        smtp_port (int): SMTP server port.
        imap_server (str): IMAP server hostname.
        imap_port (int): IMAP server port.
        email (str): Email address used for authentication.
        _password (str): Password used for authentication.
    """

    def __init__(self, integration_config: IntegrationConfig):
        self.integration_config = integration_config
        self.smtp_server = integration_config.attributes.get("SMTP Server")
        self.smtp_port = integration_config.attributes.get("SMTP Port")
        self.imap_server = integration_config.attributes.get("IMAP Server")
        self.imap_port = integration_config.attributes.get("IMAP Port")
        self.email = integration_config.attributes.get("email")
        self._password = integration_config.attributes.get("password")

    def send_email(
        self,
        to: str,
        subject: str,
        body: str,
        html_body: str = None,
        files: List[str] = None,
    ) -> Optional[str]:
        try:
            msg = MIMEMultipart()
            msg["From"] = self.email
            msg["To"] = to
            msg["Subject"] = subject

            if html_body:
                msg.attach(MIMEText(html_body, "html"))
            else:
                msg.attach(MIMEText(body, "plain"))

            if files:
                for file in files:
                    part = MIMEBase("application", "octet-stream")
                    with open(file, "rb") as f:
                        part.set_payload(f.read())
                    encoders.encode_base64(part)
                    part.add_header(
                        "Content-Disposition",
                        f"attachment; filename={file.split('/')[-1]}",
                    )
                    msg.attach(part)

            server = smtplib.SMTP_SSL(self.smtp_server, self.smtp_port)
            server.login(self.email, self._password)
            server.send_message(msg)
            server.quit()

            content_snippet = (html_body or body)[:50]
            return f"Email sent successfully from {self.email} to {to}. Message content preview: '{content_snippet}'"
        except Exception as e:
            print(f"An error occurred: {e}")
            return None

    def read_emails(self, num: int = 5) -> Optional[List[Dict[str, str]]]:
        imap_conn = None
        try:
            imap_conn = imaplib.IMAP4_SSL(self.imap_server, self.imap_port)
            imap_conn.login(self.email, self._password)
            imap_conn.select("inbox")

            _, message_numbers = imap_conn.search(None, "ALL")
            start_index = max(0, len(message_numbers[0].split()) - num)
            emails_list = []

            for num in message_numbers[0].split()[start_index:]:
                _, data = imap_conn.fetch(num, "(RFC822)")
                msg = email.message_from_bytes(data[0][1])

                email_data = {
                    "From": msg["From"],
                    "Subject": msg["Subject"],
                    "Body": "",
                }

                if msg.is_multipart():
                    for part in msg.walk():
                        if (
                            part.get_content_type() == "text/plain"
                            and "attachment" not in str(part.get("Content-Disposition"))
                        ):
                            email_data["Body"] = part.get_payload(decode=True).decode()
                            break
                else:
                    email_data["Body"] = (
                        msg.get_payload(decode=True).decode() if msg.get_payload() else ""
                    )

                emails_list.append(email_data)

            return emails_list

        except Exception as e:
            print(f"An error occurred: {e}")
            return None
        finally:
            if imap_conn:
                try:
                    imap_conn.logout()
                except Exception as e:
                    print(f"Error during logout: {e}")

---

4. Integration Adapters

The adapters convert backend configuration DTOs into SDK objects and instantiate the corresponding integration implementations.

File: autoppia_sdk/src/integrations/adapter.py

pythonCopiarfrom autoppia_sdk.src.integrations.config import IntegrationConfig
from autoppia_sdk.src.integrations.implementations.email.smtp_integration import SMPTEmailIntegration
from autoppia_sdk.src.integrations.interface import IntegrationInterface
from autoppia_sdk.src.integrations.implementations.base import Integration

class IntegrationConfigAdapter:
    """
    Adapter class for converting backend integration configuration data to IntegrationConfig objects.
    """
    
    @staticmethod
    def from_autoppia_backend(worker_config_dto):
        # Convert attributes list to dictionary
        attributes = {}
        for attr in worker_config_dto.user_integration_attributes:
            value = attr.value
            if attr.credential_obj:
                value = attr.credential_obj.credential
            attributes[attr.integration_attribute_obj.name] = value

        integration_config = IntegrationConfig(
            worker_config_dto.integration_obj.name,
            worker_config_dto.integration_obj.category,
            attributes
        )
        return integration_config

class IntegrationsAdapter:
    """
    Adapter class for managing and instantiating integration implementations based on backend configuration.
    """

    def __init__(self):
        self.integration_mapping = {
            "email": {
                "Smtp": SMPTEmailIntegration
            }
        }

    def from_autoppia_backend(self, worker_config_dto):
        integrations = {}
        for integration in worker_config_dto.user_integration:
            category = integration.integration_obj.category
            if category not in integrations:
                integrations[category] = {}

            integration_config = IntegrationConfigAdapter.from_autoppia_backend(integration)
            integration_class = self.integration_mapping[integration_config.category][integration_config.name]
            integration_instance = integration_class(integration_config)
            integrations[category][integration_config.name] = integration_instance

        return integrations

---