Add binary blob upload, download client methods

jmapc/__init__.py

@@ -6,6 +6,7 @@
 from .models import (
     AddedItem,
     Address,
+    Blob,
     Comparator,
     Delivered,
     DeliveryStatus,
@@ -44,6 +45,7 @@
 __all__ = [
     "AddedItem",
     "Address",
+    "Blob",
     "Client",
     "Comparator",
     "Delivered",

jmapc/client.py

@@ -2,7 +2,9 @@
 
 import functools
 import json
+import mimetypes
 from dataclasses import asdict, dataclass
+from pathlib import Path
 from typing import (
     Any,
     Dict,
@@ -35,7 +37,7 @@
     Response,
     ResponseOrError,
 )
-from .models import Event
+from .models import Blob, EmailBodyPart, Event
 from .session import Session
 
 RequestsAuth = Union[requests.auth.AuthBase, Tuple[str, str]]
@@ -128,6 +130,40 @@ def account_id(self) -> str:
             raise Exception("No primary account ID found")
         return primary_account_id
 
+    def upload_blob(self, file_name: Union[str, Path]) -> Blob:
+        mime_type, mime_encoding = mimetypes.guess_type(file_name)
+        upload_url = self.jmap_session.upload_url.format(
+            accountId=self.account_id
+        )
+        with open(file_name, "rb") as f:
+            r = self.requests_session.post(
+                upload_url,
+                stream=True,
+                data=f,
+                headers={"Content-Type": mime_type},
+            )
+        r.raise_for_status()
+        return Blob.from_dict(r.json())
+
+    def download_attachment(
+        self,
+        attachment: EmailBodyPart,
+        file_name: Union[str, Path],
+    ) -> None:
+        if not file_name:
+            raise Exception("Destination file name is required")
+        file_name = Path(file_name)
+        blob_url = self.jmap_session.download_url.format(
+            accountId=self.account_id,
+            blobId=attachment.blob_id,
+            name=attachment.name,
+            type=attachment.type,
+        )
+        r = self.requests_session.get(blob_url, stream=True)
+        r.raise_for_status()
+        with open(file_name, "wb") as f:
+            f.write(r.raw.data)
+
     @overload
     def request(
         self,

jmapc/models/__init__.py

@@ -29,6 +29,7 @@
 )
 from .models import (
     AddedItem,
+    Blob,
     Comparator,
     EmailAddress,
     ListOrRef,
@@ -43,6 +44,7 @@
 __all__ = [
     "AddedItem",
     "Address",
+    "Blob",
     "Comparator",
     "Delivered",
     "DeliveryStatus",

jmapc/models/models.py

@@ -15,6 +15,13 @@
 TypeOrRef = Union[T, ResultReference, Ref]
 
 
+@dataclass
+class Blob(Model):
+    id: str = field(metadata=config(field_name="blobId"))
+    type: str
+    size: int
+
+
 @dataclass
 class AddedItem(Model):
     id: str = field(metadata=config(field_name="id"))

jmapc/session.py

@@ -12,6 +12,8 @@
 class Session(Model):
     username: str
     api_url: str
+    download_url: str
+    upload_url: str
     event_source_url: str
     primary_accounts: "SessionPrimaryAccount"
     capabilities: "SessionCapabilities"

tests/conftest.py

@@ -1,6 +1,8 @@
 import json
 import logging
+import tempfile
 import time
+from pathlib import Path
 from typing import Iterable
 
 import pytest
@@ -52,3 +54,9 @@ def http_responses(
         body=json.dumps(make_session_response()),
     )
     yield http_responses_base
+
+
+@pytest.fixture
+def tempdir() -> Iterable[Path]:
+    with tempfile.TemporaryDirectory(suffix=".unit_test") as td:
+        yield Path(td)

tests/data.py

@@ -4,8 +4,13 @@
 def make_session_response() -> Dict[str, Any]:
     return {
         "apiUrl": "https://jmap-api.localhost/api",
+        "downloadUrl": (
+            "https://jmap-api.localhost/jmap/download"
+            "/{accountId}/{blobId}/{name}?type={type}"
+        ),
+        "uploadUrl": "https://jmap-api.localhost/jmap/upload/{accountId}/",
         "eventSourceUrl": (
-            "https://jmap-api.localhost/events/" "{types}/{closeafter}/{ping}"
+            "https://jmap-api.localhost/events/{types}/{closeafter}/{ping}"
         ),
         "username": "ness@onett.example.net",
         "capabilities": {

tests/test_client.py

@@ -1,11 +1,12 @@
 import json
+from pathlib import Path
 from typing import List, Set
 
 import pytest
 import requests
 import responses
 
-from jmapc import Client, constants
+from jmapc import Blob, Client, EmailBodyPart, constants
 from jmapc.auth import BearerAuth
 from jmapc.methods import (
     CoreEcho,
@@ -57,6 +58,11 @@ def test_jmap_session(
     assert test_client.jmap_session == Session(
         username="ness@onett.example.net",
         api_url="https://jmap-api.localhost/api",
+        download_url=(
+            "https://jmap-api.localhost/jmap/download"
+            "/{accountId}/{blobId}/{name}?type={type}"
+        ),
+        upload_url="https://jmap-api.localhost/jmap/upload/{accountId}/",
         event_source_url=(
             "https://jmap-api.localhost/events/{types}/{closeafter}/{ping}"
         ),
@@ -354,3 +360,55 @@ def test_error_unauthorized(
     with pytest.raises(requests.exceptions.HTTPError) as e:
         client.request(CoreEcho(data=echo_test_data))
     assert e.value.response.status_code == 401
+
+
+def test_upload_blob(
+    client: Client, http_responses: responses.RequestsMock, tempdir: Path
+) -> None:
+    blob_content = "test upload blob content"
+    source_file = tempdir / "upload.txt"
+    source_file.write_text(blob_content)
+    upload_response = {
+        "accountId": "u1138",
+        "blobId": "C2187",
+        "type": "text/plain",
+        "size": len(blob_content),
+    }
+    http_responses.add(
+        method=responses.POST,
+        url="https://jmap-api.localhost/jmap/upload/u1138/",
+        body=json.dumps(upload_response),
+    )
+    response = client.upload_blob(source_file)
+    assert response == Blob(
+        id="C2187", type="text/plain", size=len(blob_content)
+    )
+
+
+def test_download_attachment(
+    client: Client, http_responses: responses.RequestsMock, tempdir: Path
+) -> None:
+    blob_content = "test download blob content"
+    http_responses.add(
+        method=responses.GET,
+        url=(
+            "https://jmap-api.localhost/jmap/download"
+            "/u1138/C2187/download.txt?type=text/plain"
+        ),
+        body=blob_content,
+    )
+    dest_file = tempdir / "download.txt"
+    with pytest.raises(Exception) as e:
+        client.download_attachment(
+            EmailBodyPart(
+                name="download.txt", blob_id="C2187", type="text/plain"
+            ),
+            "",
+        )
+    assert str(e.value) == "Destination file name is required"
+    assert not dest_file.exists()
+    client.download_attachment(
+        EmailBodyPart(name="download.txt", blob_id="C2187", type="text/plain"),
+        dest_file,
+    )
+    assert dest_file.read_text() == blob_content