Python：使い捨て受信箱APIでメールを受信、キー不要

{
  "address": "py-3f9a1c2e@grabmail.io",
  "alias": "k7m2p9x4q1wz@example.net",
  "count": 1,
  "next": null,
  "messages": [
    { "id": "01JR8W2K4Q", "from": "noreply@example.com", "subject": "Your verification code",
      "date": "2026-09-01T18:31:07Z", "seen": false, "attachments": 0, "expires_at": "2026-09-06T18:31:07Z" }
  ]
}

"""grabmail.py — a disposable inbox from Python. Three endpoints, no key, no account."""
from __future__ import annotations

import secrets
import time
from pathlib import Path

import requests

API = "https://grabmail.io/api/v1"
DOMAIN = "grabmail.io"


def fresh_address(prefix: str = "py") -> str:
    """A mailbox nothing else is using. Nothing has to be created first."""
    return f"{prefix}-{secrets.token_hex(4)}@{DOMAIN}"


class Inbox:
    def __init__(self, address: str | None = None, session: requests.Session | None = None):
        self.address = address or fresh_address()
        self.http = session or requests.Session()

    def _get(self, url: str, **params) -> requests.Response:
        """One GET, with the only retry that is ever right: waiting out a 429."""
        while True:
            r = self.http.get(url, params=params, timeout=15)
            if r.status_code == 429:
                time.sleep(float(r.headers.get("Retry-After", 1)))
                continue
            r.raise_for_status()
            return r

    def list(self, limit: int = 50, before: str | None = None) -> dict:
        params = {"address": self.address, "limit": limit}
        if before:
            params["before"] = before
        return self._get(f"{API}/mailbox", **params).json()

    def wait_for(self, timeout: float = 60, subject_contains: str | None = None,
                 from_contains: str | None = None) -> dict:
        """Block until a matching message arrives, then return it in full."""
        deadline = time.monotonic() + timeout
        while time.monotonic() < deadline:
            for m in self.list()["messages"]:
                if subject_contains and subject_contains.lower() not in m["subject"].lower():
                    continue
                if from_contains and from_contains.lower() not in m["from"].lower():
                    continue
                return self.read(m["id"])
            time.sleep(1)                       # one read a second, never throttled
        raise TimeoutError(f"no message for {self.address} within {timeout}s")

    def read(self, message_id: str) -> dict:
        return self._get(f"{API}/message/{message_id}", mailbox=self.address).json()

    def delete(self, message_id: str) -> None:
        """Optional and idempotent: everything expires on its own."""
        self.http.delete(f"{API}/message/{message_id}", params={"mailbox": self.address}, timeout=15)

    def download(self, attachment: dict, into: Path) -> Path:
        """Save one entry of a message's `attachments` list. The URL carries its own ?mailbox=."""
        into.mkdir(parents=True, exist_ok=True)
        target = into / attachment["filename"]
        with self.http.get("https://grabmail.io" + attachment["url"], stream=True, timeout=60) as r:
            r.raise_for_status()
            with target.open("wb") as f:
                for chunk in r.iter_content(65536):
                    f.write(chunk)
        return target

from grabmail import Inbox

inbox = Inbox()
print("sign up with:", inbox.address)

message = inbox.wait_for(subject_contains="code")
print(message["subject"])
print(message["text"])          # the plain-text part; message["html"] is the HTML part or None

"""grabmail_async.py — the same inbox for asyncio, on httpx."""
from __future__ import annotations

import asyncio
import time

import httpx

from grabmail import API, fresh_address


class AsyncInbox:
    def __init__(self, address: str | None = None, client: httpx.AsyncClient | None = None):
        self.address = address or fresh_address()
        self.http = client or httpx.AsyncClient(timeout=15)

    async def _get(self, url: str, **params) -> httpx.Response:
        while True:
            r = await self.http.get(url, params=params)
            if r.status_code == 429:
                await asyncio.sleep(float(r.headers.get("Retry-After", 1)))
                continue
            r.raise_for_status()
            return r

    async def list(self, limit: int = 50, before: str | None = None) -> dict:
        params = {"address": self.address, "limit": limit, **({"before": before} if before else {})}
        return (await self._get(f"{API}/mailbox", **params)).json()

    async def wait_for(self, timeout: float = 60, subject_contains: str | None = None) -> dict:
        deadline = time.monotonic() + timeout
        while time.monotonic() < deadline:
            for m in (await self.list())["messages"]:
                if not subject_contains or subject_contains.lower() in m["subject"].lower():
                    return await self.read(m["id"])
            await asyncio.sleep(1)
        raise TimeoutError(f"no message for {self.address} within {timeout}s")

    async def read(self, message_id: str) -> dict:
        return (await self._get(f"{API}/message/{message_id}", mailbox=self.address)).json()


async def main() -> None:
    inboxes = [AsyncInbox() for _ in range(5)]          # five sign-ups, waited on together
    for i in inboxes:
        print("sign up with:", i.address)
    messages = await asyncio.gather(*(i.wait_for(subject_contains="code") for i in inboxes))
    for m in messages:
        print(m["to"], "->", m["subject"])

asyncio.run(main())

from collections.abc import Iterator


def all_messages(inbox: Inbox) -> Iterator[dict]:
    """Every summary in the mailbox, newest first, however many pages it takes."""
    before = None
    while True:
        page = inbox.list(limit=200, before=before)
        yield from page["messages"]
        before = page["next"]
        if not before:
            return


for m in all_messages(inbox):
    print(m["date"], m["from"], m["subject"], "expires", m["expires_at"])

from pathlib import Path

message = inbox.wait_for(subject_contains="invoice")
for a in message["attachments"]:
    print(a["filename"], a["mime"], a["size"], "bytes")
    path = inbox.download(a, into=Path("downloads") / message["id"])
    print("saved to", path)

# conftest.py
import pytest

from grabmail import Inbox


@pytest.fixture
def inbox() -> Inbox:
    """A brand-new mailbox for this test, and this test only."""
    return Inbox()

# test_signup.py
import re

CODE = re.compile(r"code is\D{0,12}(\d{6})", re.I)      # anchored on YOUR template's wording


def test_signup_confirmation(client, inbox):
    client.post("/signup", data={"email": inbox.address, "password": "hunter2hunter2"})

    message = inbox.wait_for(subject_contains="confirm")
    code = CODE.search(f"{message.get('text') or ''} {message.get('html') or ''}").group(1)

    response = client.post("/confirm", data={"email": inbox.address, "code": code})
    assert response.status_code == 200