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 NotImplementedErrorFile: 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.
"""
pass3. 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 integrationsLast updated