Spaces:
Running
Running
File size: 1,824 Bytes
4ef118d | 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 | """
Abstract base class for email providers.
Defines the interface that all email provider implementations must follow,
making it easy to add new providers (Outlook, Yahoo, etc.) in the future.
"""
from __future__ import annotations
from abc import ABC, abstractmethod
from dataclasses import dataclass
from datetime import datetime
@dataclass
class EmailMessage:
"""Normalized email message structure returned by all providers."""
message_id: str # Provider-specific unique ID (prevents duplicate processing)
subject: str
sender: str # "Name <email@example.com>" format
received_at: datetime
body_text: str # Plain text body (truncated to ~2000 chars for summarization)
class BaseEmailProvider(ABC):
"""
Abstract base class for email providers.
Each provider must implement fetch_new_emails() and mark_as_read().
"""
@abstractmethod
def fetch_new_emails(self, max_results: int = 5) -> list[tuple[str, EmailMessage]]:
"""
Fetch recent unread emails from the provider.
Args:
max_results: Maximum number of emails to return per poll cycle.
Returns:
List of (imap_id, EmailMessage) tuples, ordered newest first.
imap_id is the provider-specific ID needed for mark_as_read().
"""
...
@abstractmethod
def mark_as_read(self, imap_id: str) -> bool:
"""
Mark an email as read on the provider's server.
Args:
imap_id: The provider-specific message ID returned by fetch_new_emails.
Returns:
True if successful, False otherwise.
"""
...
@abstractmethod
def get_provider_name(self) -> str:
"""Return the provider identifier string (e.g. 'gmail')."""
...
|