API ve otomasyon

Python: tek kullanımlık kutu API'siyle e-posta almak

Altmış satır Python, bir bağımlılık ve API anahtarı yok: uydurduğunuz bir adres, son tarihli bir bekleme, bir dict olarak mesaj. İşte requests üzerine kurulu modül, asyncio için httpx üzerindeki aynısı, yoğun bir posta kutusunda sayfalama, bir eki kaydetme, bir pytest fixture'ı ve gözetimsiz ilk çalıştığında ortaya çıkan altı hata.

  • Orta düzey
  • 23 dk okuma
Gevşekçe kıvrılmış gri bir borunun açık ucundan küçük gri bir tepsiye çıkan mavi bir zarf

API, Python'ın gördüğü şekliyle

Sunucu tarafında kurulacak hiçbir şey ve karşı kimlik doğrulaması yapılacak hiçbir şey yoktur: genel bir alan adındaki bir posta kutusu, adresini bilen herkes tarafından, düz HTTPS üzerinden, JSON olarak okunabilir. Yüzeyin tamamı üç çağrıdır:

GET /api/v1/mailbox?address=…
Bir adreste bekleyen her şey, en yeniden başlayarak, bir özet listesi olarak. Boş bir posta kutusu, count: 0 ile 200'dür — asla 404 değil. limit, tek bir yanıtı sınırlar (1-200, varsayılan 50) ve before bunun ötesinde sayfalar.
GET /api/v1/message/{id}?mailbox=…
Bir mesajın tamamı: gönderen, alıcı, konu, tarih, düz metin parçası, HTML parçası (ya da null) ve her birinin hazır bir URL'si olan bir ek listesi.
DELETE /api/v1/message/{id}?mailbox=…
Onu 5 gün içinde değil, şimdi kaldırır. İdempotenttir: iki kez silmek yine de 200 yanıtı verir.

Bir listeleme şöyle görünür. alias, aynı posta kutusuna teslimat yapan ama onu okumak için kullanılamayan, ayrı bir alan adındaki ikinci bir adrestir — gelen kutusunu açamamasını tercih ettiğinizde bir siteye vereceğiniz adres.

GET /api/v1/mailbox — yanıt
{
  "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" }
  ]
}

Modül

Tek bir dosya, tek bir sınıf ve tek bağımlılık olarak requests. Bilinçli olarak sıradandır: bir session, bir son tarih döngüsü ve şimdiye kadar doğru olan tek yeniden deneme — bir 429'u uyuyarak geçirmek.

grabmail.py
"""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

Kullanımı dört satırdır. Adresi yazdırın, bir adres istenen her yerde kullanın ve bekleyin:

ilk çalıştırma
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

asyncio için httpx üzerinde aynısı

Bir betiğin aynı anda birkaç gelen kutusunu izlemesi gerektiğinde — toplu bir job'da beş kayıt, hesapları jonglör gibi yöneten bir ajan — senkron döngü beklemeleri sıralı hale getirir. httpx.AsyncClient, aynı sınıfı awaitable yapar ve asyncio.gather hepsini birlikte bekler:

grabmail_async.py
"""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())

Her biri saniyede bir kez yoklanan beş gelen kutusu, saniyede beş istektir; bu, dakikada 1200'lük istemci başına üst sınırın oldukça altındadır. Aynı anda yirmi gelen kutusunu aştığınızda üst sınıra ulaşırsınız ve 429 dalı uyumaya başlar — bu bir başarısızlık değil, doğru davranıştır.

Yoğun bir posta kutusu: before ile sayfalama

Bir listeleme en fazla 200 özet döndürür. Bundan fazlasını alan bir posta kutusu — örneğin kendi alan adınızdaki, bir günlük geri dönen postaları toplayan bir catch-all adresi — sayfa sayfa okunur: bir yanıtın next değerini bir sonraki isteğin before parametresi olarak geçirin ve next, null olduğunda durun.

kaç sayfa olursa olsun, her mesaj
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"])

İmleç, zaten sahip olduğunuz en eski mesajın kimliğidir, bu yüzden üstte yeni posta gelirken bile bir sayfa kararlı kalır. Bir gelen kutusunu bir betikten otomatikleştirme, imleci zamanlama ve saklama süresiyle birlikte daha ayrıntılı ele alır.

Diske ekler

Her mesaj, eklerini bir dosya adı, beyan edilmiş bir tür, bayt cinsinden bir boyut ve bir URL ile listeler. URL zaten ?mailbox= parametresini taşır, bu yüzden olduğu gibi alınır. Yanıt, göndereninin dosyayı ne olarak etiketlediğinden bağımsız olarak her zaman bir Content-Disposition: attachment başlığıyla application/octet-stream'dır — gerçek tür, JSON'daki, bir talimat değil veri olan mime alanıdır.

bir mesajın her ekini kaydetme
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)

Bir pytest fixture'ı olarak

Taze bir Inbox döndüren bir fixture, her teste kendi posta kutusunu verir; bu, bir posta testinin tek en önemli özelliğidir: hiçbir çalıştırma önceki bir çalıştırmanın mesajını asla okuyamaz. Ayıklama deseni, otomatik testlerde OTP kodları rehberinin ayrıntılı olarak açıkladığı nedenlerle, "altı rakam" yerine şablonun ifadesine sabitlenmiştir.

conftest.py
# 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
# 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

client fixture'ı, framework'ünüzün sağladığı her şeydir — Flask'ın test istemcisi, Django'nunki, çalışan bir sunucuya yönlendirilmiş bir httpx.Client. Posta kutusu tarafı umursamaz. Paketi bir CI runner'ında çalıştırmak bir giden trafik kuralı ve bir zaman aşımı ekler; ikisi de GitHub Actions rehberinde ele alınır.

Gözetimsiz ilk çalıştığında ortaya çıkan hatalar

Bunların hiçbiri bir dizüstü bilgisayarda bozulmaz. Hepsi bir salı gecesi bir cron job'da bozulur.

BelirtiNedenÇözüm
Gönderen bozuk olsa bile her seferinde geçiyorHer çalıştırmada aynı adres; ilk yoklama önceki çalıştırmanın mesajını buluyor.Çalıştırma başına fresh_address(). Önemli olan budur.
Günlükte 429, ardından bir çökmeUyumayan bir döngü ya da tek bir adresi yoklayan iki betik.Adres başına saniyede bir okuma; Retry-After'ı uyuyarak geçirin; betik başına bir adres.
Yavaş bir günde zaman aşımına uğruyor, yeniden denemede geçiyorBir son tarih yerine bir deneme sayısı, ya da göndereninin kuyruğundan daha kısa bir son tarih.time.monotonic() son tarihi, işlemsel bir posta için altmış saniye.
/mailbox'tan 404Alan adı burada barındırılmıyor — bir yazım hatası ya da MX'i eksik kendi alan adınız.Adresi kontrol edin; kendi alan adınız için, MX'in smtp.grabmail.io'i gösterdiğini kontrol edin.
Yanlış mesajı okuyorAkış iki mesaj gönderdiğinde en yenisini aldı.subject_contains ya da from_contains ile filtreleyin.
Bir hafta çalışıyor, sonra bir mesajda 4045 günden eski, saklanmış bir mesaj kimliği.Hiçbir şey 5 günden fazla hayatta kalmaz. Önbelleğe almak yerine yeniden alın.

Bitti demeden önce

  • Çalıştırma başına, test başına ya da ajan başına taze bir adres — asla bir sabit değil.
  • Monotonik bir son tarih; saniyede bir okuma; uyuyarak geçirilen, asla fırlatılmayan 429.
  • Bir akış birden fazla mesaj gönderdiğinde konu ya da gönderen üzerinde bir filtre.
  • Kendi ifadenize sabitlenmiş bir desenle önce ayrıştırılan metin parçası.
  • Güvenilmeyen dosyalar olarak ele alınan, mesaj kimliği altında kaydedilen ekler.
  • Günler boyunca önbelleğe alınan bir mesaj kimliği yok; burada hiçbir şey 5 günden uzun yaşamaz.

İstemcinin tamamı bu kadar. Node, Deno ve Bun için TypeScript'teki aynı modül, Node.js rehberinde; her durum koduyla birlikte istek ve yanıt biçimleri API referansında, ve istemciyi yazmak yerine üretmeyi tercih edecekler için bir OpenAPI 3.1 belgesi var.

Sorular

Bir API anahtarına ihtiyacım var mı?

Hayır. Genel alan adları anahtar, hesap ya da başlık gerektirmez. Yalnızca tek kullanımlık posta kara listelerinin dışında tutulan ücretli alan adı havuzu bir Authorization: Bearer başlığı kullanır ve bunun dışında yukarıdaki kod onun için de aynıdır.

Bunun yerine istemciyi OpenAPI belgesinden üretebilir miyim?

Evet — /openapi.json, OpenAPI 3.1'dir ve herhangi bir üretici üç çağrıyı da üretecektir. Yine de listeleme çağrısının etrafında bu rehberdeki son tarih döngüsünü isteyeceksiniz, çünkü hiçbir üretici sizin için bunu yazmaz.

Bir betik aynı anda kaç posta kutusunu yoklayabilir?

Rahatlıkla yirmi: adres başına sınır saniyede bir okumadır ve istemci başına üst sınır dakikada 1200 istektir, bu da saniyede bir kez yoklanan yirmi adres demektir. Yukarıdaki async sürüm ikisine de uyar ve üst sınırın ötesinde başarısız olmak yerine 429'u uyuyarak geçirir.

Python'dan kendi alan adımı kullanabilir miyim?

Evet, DOMAIN sabiti dışında kodda hiçbir değişiklik yapmadan. smtp.grabmail.io'i gösteren tek bir MX kaydı ve alan adı üzerindeki her adres, aynı modülün okuduğu bir posta kutusu haline gelir — kurulum burada. Uygulamanız genel tek kullanımlık alan adlarını reddettiğinde bu doğru yanıttır.

Betiğim onu kullanırken posta kutusu özel midir?

Hayır. Genel bir alan adında da kendi alan adınızda da, adresi bilen herkes onu okuyabilir. Birkaç saniye boyunca tek bir onay kodu tutan rastgele bir adres sorun değildir; gerçek müşteri postasını birine yönlendiren bir betik ise sorundur.

Bir betik yerine bir yapay zekâ ajanı için bir şey var mı?

Aynı origin'de, anahtarsız bir MCP sunucusu vardır; wait_for_message aracı, posta düşene kadar çağrıyı açık tutar — her yoklamanın ona token'a mal olduğu bir ajanın ihtiyaç duyduğu şekil budur. Bir yapay zekâ ajanının okuyabileceği bir gelen kutusu bunu ele alır.

Henüz tazeyken deneyin

Bir adres tek tıkla alınır, hesap ve kart gerekmez. Bu rehberdeki her şey onunla hemen çalışır.

Tekrar hoş geldiniz

Kutularınız ve alan adlarınız tek bir yerde.