Implements including attachments in notifications (on supported adapters/backends)

.gitignore

@@ -48,3 +48,4 @@ tmp_email/
 
 # built files
 dist/
+notifications.json

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "vintasend"
-version = "0.1.4"
+version = "1.0.0"
 description = "A flexible package for implementing transactional notifications"
 authors = ["Hugo bessa <hugo@vinta.com.br>"]
 license = "MIT"
@@ -15,6 +15,7 @@ python = ">=3.9,<3.14"
 typing-extensions = "^4.12.2"
 pytest-asyncio = "^0.24.0"
 packaging = "^25.0"
+requests = "^2.32.5"
 
 
 [tool.poetry.group.dev.dependencies]
@@ -164,3 +165,7 @@ omit = [
 [tool.pytest.ini_options]
 python_files = ["test_*.py"]
 addopts = "--dist=loadscope"
+
+[[tool.mypy.overrides]]
+module = ["requests.*"]
+ignore_missing_imports = true

vintasend/services/dataclasses.py

@@ -1,7 +1,106 @@
 import datetime
+import io
+import mimetypes
 import uuid
-from dataclasses import dataclass
-from typing import TypedDict
+from abc import ABC, abstractmethod
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any, BinaryIO, TypedDict
+
+
+# Type alias for supported file inputs (for creating notifications)
+FileAttachment = (
+    BinaryIO |           # File-like object with read()
+    io.BytesIO |         # In-memory bytes
+    io.StringIO |        # In-memory text
+    Path |               # Path object
+    str |                # File path string OR URL
+    bytes                # Raw bytes data
+)
+
+
+class AttachmentFile(ABC):
+    """Abstract interface for accessing stored attachment files"""
+
+    @abstractmethod
+    def read(self) -> bytes:
+        """Read the entire file content"""
+        pass
+
+    @abstractmethod
+    def stream(self) -> BinaryIO:
+        """Get a stream for reading the file"""
+        pass
+
+    @abstractmethod
+    def url(self, expires_in: int = 3600) -> str:
+        """Get a temporary URL for file access"""
+        pass
+
+    @abstractmethod
+    def delete(self) -> None:
+        """Delete the file from storage"""
+        pass
+
+
+@dataclass
+class NotificationAttachment:
+    """Input attachment for creating notifications"""
+    file: FileAttachment
+    filename: str
+    content_type: str | None = None  # Auto-detected if None
+    description: str | None = None
+    is_inline: bool = False
+
+    def __post_init__(self):
+        if self.content_type is None:
+            self.content_type = self._detect_content_type()
+
+    def _detect_content_type(self) -> str:
+        content_type, _ = mimetypes.guess_type(self.filename)
+        return content_type or 'application/octet-stream'
+
+    def is_url(self) -> bool:
+        """Check if file is a URL"""
+        return isinstance(self.file, str) and (
+            self.file.startswith('http://') or
+            self.file.startswith('https://') or
+            self.file.startswith('s3://') or
+            self.file.startswith('gs://') or
+            self.file.startswith('azure://')
+        )
+
+
+@dataclass
+class StoredAttachment:
+    """Represents an attachment stored by the backend"""
+    id: str | uuid.UUID
+    filename: str
+    content_type: str
+    size: int
+    checksum: str
+    created_at: datetime.datetime
+    file: AttachmentFile  # File access - abstracted through AttachmentFile interface
+    description: str | None = None
+    is_inline: bool = False
+    # Backend-specific storage metadata
+    storage_metadata: dict[str, Any] = field(default_factory=dict)
+
+    def get_file_data(self) -> bytes:
+        """Get the raw file data"""
+        return self.file.read()
+
+    def get_file_stream(self) -> BinaryIO:
+        """Get a stream for reading the file (for large files)"""
+        return self.file.stream()
+
+    def get_file_url(self, expires_in: int = 3600) -> str:
+        """Get a temporary URL for file access (if supported by backend)"""
+        return self.file.url(expires_in)
+
+    def delete(self) -> None:
+        """Delete this attachment from storage"""
+        self.file.delete()
 
 
 class NotificationContextDict(dict):
@@ -70,6 +169,7 @@ class Notification:
     context_used: dict | None = None
     adapter_used: str | None = None
     adapter_extra_parameters: dict | None = None
+    attachments: list[StoredAttachment] = field(default_factory=list)
 
 @dataclass
 class OneOffNotification:
@@ -89,6 +189,7 @@ class OneOffNotification:
     context_used: dict | None = None
     adapter_used: str | None = None
     adapter_extra_parameters: dict | None = None
+    attachments: list[StoredAttachment] = field(default_factory=list)
 
 class UpdateNotificationKwargs(TypedDict, total=False):
     title: str
@@ -99,3 +200,4 @@ class UpdateNotificationKwargs(TypedDict, total=False):
     subject_template: str | None
     preheader_template: str | None
     adapter_extra_parameters: dict | None
+    attachments: list[StoredAttachment]

vintasend/services/notification_adapters/stubs/fake_adapter.py

@@ -1,6 +1,6 @@
 import datetime
 import uuid
-from typing import Generic, TypeVar, cast
+from typing import Generic, TypeVar
 
 from vintasend.constants import NotificationTypes
 from vintasend.services.dataclasses import Notification, NotificationContextDict, OneOffNotification
@@ -26,15 +26,29 @@ class FakeEmailAdapter(Generic[B, T], BaseNotificationAdapter[B, T]):
     notification_type = NotificationTypes.EMAIL
     backend: B
     template_renderer: T
-    sent_emails: list[tuple["Notification | OneOffNotification", "NotificationContextDict"]] = []
 
     def __init__(self, *args, **kwargs) -> None:
         super().__init__(*args, **kwargs)
-        self.sent_emails = []
+        self.sent_emails: list[tuple["Notification | OneOffNotification", "NotificationContextDict", list[dict]]] = []
 
     def send(self, notification: "Notification | OneOffNotification", context: "NotificationContextDict") -> None:
         self.template_renderer.render(notification, context)
-        self.sent_emails.append((notification, context))
+
+        # Capture attachment information for testing
+        attachment_info = [
+            {
+                'id': str(attachment.id),
+                'filename': attachment.filename,
+                'content_type': attachment.content_type,
+                'size': attachment.size,
+                'is_inline': attachment.is_inline,
+                'description': attachment.description,
+                'checksum': attachment.checksum,
+            }
+            for attachment in notification.attachments
+        ]
+
+        self.sent_emails.append((notification, context, attachment_info))
 
 
 BAIO = TypeVar("BAIO", bound=AsyncIOBaseNotificationBackend)
@@ -44,15 +58,29 @@ class FakeAsyncIOEmailAdapter(Generic[BAIO, T], AsyncIOBaseNotificationAdapter[B
     notification_type = NotificationTypes.EMAIL
     backend: BAIO
     template_renderer: T
-    sent_emails: list[tuple["Notification | OneOffNotification", "NotificationContextDict"]] = []
 
     def __init__(self, *args, **kwargs) -> None:
         super().__init__(*args, **kwargs)
-        self.sent_emails = []
+        self.sent_emails: list[tuple["Notification | OneOffNotification", "NotificationContextDict", list[dict]]] = []
 
     async def send(self, notification: "Notification | OneOffNotification", context: "NotificationContextDict") -> None:
         self.template_renderer.render(notification, context)
-        self.sent_emails.append((notification, context))
+
+        # Capture attachment information for testing
+        attachment_info = [
+            {
+                'id': str(attachment.id),
+                'filename': attachment.filename,
+                'content_type': attachment.content_type,
+                'size': attachment.size,
+                'is_inline': attachment.is_inline,
+                'description': attachment.description,
+                'checksum': attachment.checksum,
+            }
+            for attachment in notification.attachments
+        ]
+
+        self.sent_emails.append((notification, context, attachment_info))
 
 
 class FakeAsyncEmailAdapter(AsyncBaseNotificationAdapter, Generic[B, T], FakeEmailAdapter[B, T]):
@@ -110,6 +138,7 @@ def notification_from_dict(self, notification_dict: NotificationDict) -> "Notifi
             context_name=notification_dict["context_name"],
             subject_template=notification_dict["subject_template"],
             preheader_template=notification_dict["preheader_template"],
+            attachments=[],  # Default to empty attachments for delayed sending
         )
 
     def one_off_notification_from_dict(self, notification_dict: OneOffNotificationDict) -> "OneOffNotification":
@@ -140,6 +169,7 @@ def one_off_notification_from_dict(self, notification_dict: OneOffNotificationDi
             context_name=notification_dict["context_name"],
             subject_template=notification_dict["subject_template"],
             preheader_template=notification_dict["preheader_template"],
+            attachments=[],  # Default to empty attachments for delayed sending
         )
 
 

vintasend/services/notification_backends/asyncio_base.py

@@ -10,6 +10,7 @@
 if TYPE_CHECKING:
     from vintasend.services.dataclasses import (
         Notification,
+        NotificationAttachment,
         OneOffNotification,
         UpdateNotificationKwargs,
     )
@@ -65,6 +66,7 @@ async def persist_notification(
         subject_template: str,
         preheader_template: str,
         adapter_extra_parameters: dict | None = None,
+        attachments: list["NotificationAttachment"] | None = None,
         lock: asyncio.Lock | None = None
     ) -> "Notification":
         ...
@@ -84,6 +86,7 @@ async def persist_one_off_notification(
         subject_template: str,
         preheader_template: str,
         adapter_extra_parameters: dict | None = None,
+        attachments: list["NotificationAttachment"] | None = None,
         lock: asyncio.Lock | None = None
     ) -> "OneOffNotification":
         ...

vintasend/services/notification_backends/base.py

@@ -10,6 +10,7 @@
 if TYPE_CHECKING:
     from vintasend.services.dataclasses import (
         Notification,
+        NotificationAttachment,
         OneOffNotification,
         UpdateNotificationKwargs,
     )
@@ -69,6 +70,7 @@ def persist_notification(
         subject_template: str,
         preheader_template: str,
         adapter_extra_parameters: dict | None = None,
+        attachments: list["NotificationAttachment"] | None = None,
     ) -> "Notification":
         raise NotImplementedError
 
@@ -87,6 +89,7 @@ def persist_one_off_notification(
         subject_template: str,
         preheader_template: str,
         adapter_extra_parameters: dict | None = None,
+        attachments: list["NotificationAttachment"] | None = None,
     ) -> "OneOffNotification":
         raise NotImplementedError