Python SDK - OneShotMail Docs

Installation

pip install oneshot-mail

Requires Python 3.9+. The SDK depends on httpx for HTTP and pydantic for response models.

Configuration

From environment variable (recommended)

import oneshot

# Uses ONESHOT_API_KEY from environment automatically
address = oneshot.create()

Explicit client

from oneshot import Client

client = Client(
    api_key="osm_live_your_key",
    base_url="https://api.oneshotemail.com/v1",  # default
    timeout=30.0,  # default HTTP timeout in seconds
)

Module-level configuration

import oneshot

oneshot.configure(
    api_key="osm_live_your_key",
    base_url="https://api.oneshotemail.com/v1",
    timeout=30.0,
)

Context manager

The client implements the context manager protocol for clean resource management:

from oneshot import Client

with Client(api_key="osm_live_your_key") as client:
    addr = client.create(ttl_seconds=300)
    email = client.wait_for_email(addr.id)
# Connection pool is automatically released

API Reference

`create(ttl_seconds, label, mode)`

Create a new one-shot email address.

Parameter	Type	Default	Description
`ttl_seconds`	int	`3600`	Time-to-live before auto-expiry.
`label`	str	`None`	Optional label for filtering.
`mode`	str	`"receive"`	`"receive"` or `"send"`.

Returns: Address object.

addr = client.create(ttl_seconds=300, label="signup-test")
print(addr.id)        # "abc123xyz789def456"
print(addr.address)   # "abc123xyz789def456@in.oneshotemail.com"
print(addr.status)    # "waiting"
print(addr.mode)      # "receive"
print(addr.expires_at)  # datetime
print(addr.label)     # "signup-test"

`get(address_id)`

Retrieve an address and its current status.

Parameter	Type	Description
`address_id`	str	The address ID.

Returns: Address object (with email summary if received).

Raises: NotFoundError, ExpiredError.

addr = client.get("abc123xyz789def456")
if addr.status == "received":
    print(f"Email from: {addr.email.from_address}")
    print(f"Subject: {addr.email.subject}")

`get_email(address_id)`

Retrieve the full parsed email content.

Parameter	Type	Description
`address_id`	str	The address ID.

Returns: Email object with from_address, to, subject, text_body, html_body, headers, received_at, size_bytes, and attachments.

Raises: NotFoundError (no email yet), ExpiredError.

email = client.get_email("abc123xyz789def456")
print(email.from_address)   # "noreply@example.com"
print(email.subject)        # "Verify your account"
print(email.text_body)      # "Click here to verify..."
print(email.html_body)      # "<html>...</html>"
print(email.headers)        # {"Message-ID": "...", "DKIM-Signature": "..."}
print(email.size_bytes)     # 15234
print(email.attachments)    # [Attachment(...), ...]

`get_email_raw(address_id)`

Retrieve the raw RFC 822 email source as a string.

raw = client.get_email_raw("abc123xyz789def456")
# Full email source with all headers
print(raw)

`download_attachment(address_id, index)`

Download a single attachment by zero-based index.

Returns: bytes — the raw attachment content.

pdf_bytes = client.download_attachment("abc123xyz789def456", 0)
with open("invoice.pdf", "wb") as f:
    f.write(pdf_bytes)

`wait_for_email(address_id, timeout, poll_interval)`

Poll until an email arrives, using exponential backoff. This is the primary method for test suites.

Parameter	Type	Default	Description
`address_id`	str		The address ID to watch.
`timeout`	float	`60`	Max seconds to wait.
`poll_interval`	float	`2.0`	Initial polling interval.

Returns: Email object.

Raises: WaitTimeoutError if no email arrives within timeout. ExpiredError if the address expires while waiting.

Backoff behavior: starts at poll_interval, multiplied by 1.5 after each poll, capped at 10 seconds.

# Basic usage
email = client.wait_for_email(addr.id, timeout=30)

# Aggressive polling for fast email delivery
email = client.wait_for_email(addr.id, timeout=10, poll_interval=0.5)

# Patient waiting for slow external services
email = client.wait_for_email(addr.id, timeout=120, poll_interval=5.0)

`send(to, subject, text_body, html_body, attachments, ttl_seconds, label)`

Create a one-shot address and send a single email from it.

Parameter	Type	Default	Description
`to`	str		Destination address.
`subject`	str		Email subject.
`text_body`	str	`None`	Plain text body.
`html_body`	str	`None`	HTML body.
`attachments`	list of tuples	`None`	Attachments (see below).
`ttl_seconds`	int	`300`	TTL for the address record.
`label`	str	`None`	Optional label.

Attachment format: Each attachment is a tuple of (filename, content_bytes) or (filename, content_type, content_bytes). File-like objects are also accepted instead of bytes.

Returns: Address object with mode="send" and status="sent".

# Simple text email
result = client.send(
    to="intake@myapp.com",
    subject="Test invoice",
    text_body="Please process this invoice.",
)

# HTML email with attachments
with open("invoice.pdf", "rb") as f:
    pdf_bytes = f.read()

result = client.send(
    to="intake@myapp.com",
    subject="Invoice #1234",
    text_body="See attached invoice.",
    html_body="<p>See attached invoice.</p>",
    attachments=[
        ("invoice.pdf", "application/pdf", pdf_bytes),
        ("receipt.txt", b"Total: $42.00"),
    ],
    label="invoice-test",
)

`list(status, label, mode, limit, cursor)`

List addresses for the authenticated user.

Parameter	Type	Default	Description
`status`	str	`None`	Filter by status.
`label`	str	`None`	Filter by label.
`mode`	str	`None`	Filter by mode.
`limit`	int	`20`	Max results.
`cursor`	str	`None`	Pagination cursor.

Returns: AddressListResponse with items (list of Address) and next_cursor.

# List all waiting addresses
result = client.list(status="waiting")
for addr in result.items:
    print(f"{addr.id}: {addr.status}")

# Paginate
result = client.list(limit=10)
while result.next_cursor:
    result = client.list(limit=10, cursor=result.next_cursor)

`delete(address_id)`

Delete an address and its email data immediately.

client.delete("abc123xyz789def456")

`delete_by_label(label)`

Bulk-delete all addresses matching a label.

client.delete_by_label("ci-run-abc123")

`account()`

Get account details and usage.

Returns: Account object.

acct = client.account()
print(f"Plan: {acct.plan}")
print(f"Receive usage: {acct.usage.receive.used}/{acct.usage.receive.limit}")
print(f"Credits: {acct.credits_remaining}")

`buy_credits(amount)`

Initiate a credit purchase.

Returns: CheckoutURL with checkout_url string.

checkout = client.buy_credits(500)
print(f"Complete purchase at: {checkout.checkout_url}")

`health()`

Check API health (does not require authentication).

Returns: HealthStatus with status, region, version.

h = client.health()
print(f"API status: {h.status}, region: {h.region}, version: {h.version}")

`save_attachments(address_id, directory)`

Convenience method: download all attachments for an address and save them to a directory.

Returns: list of Path objects pointing to the saved files.

paths = client.save_attachments("abc123xyz789def456", "./downloads")
for path in paths:
    print(f"Saved: {path}")

Error handling

All SDK exceptions inherit from OneShotError:

Exception	HTTP Status	When
`UnauthorizedError`	401	Invalid or missing API key.
`QuotaExceededError`	402	Quota and credits exhausted.
`NotFoundError`	404	Address not found or no email yet.
`ExpiredError`	410	Address expired and deleted.
`ValidationError`	400/422	Invalid request parameters.
`RateLimitedError`	429	Rate limit exceeded.
`WaitTimeoutError`	N/A	`wait_for_email()` timed out.
`OneShotError`	Any	Base class for all other errors.

Every exception has these attributes:

message — Human-readable description.
code — Machine-readable code (e.g., "QUOTA_EXCEEDED").
status_code — HTTP status code (if from an API response).
response_body — Raw response dict (if available).

QuotaExceededError additionally has upgrade_url. RateLimitedError additionally has retry_after (seconds).

Behave (BDD) integration

Behave is the Python BDD framework. Here is a complete worked example.

Feature file: `features/signup.feature`

Feature: User signup
  As a new user
  I want to sign up for an account
  So that I can access the application

  Scenario: Successful signup sends verification email
    Given I have a temporary email address labelled "signup"
    When I sign up with that email address
    Then I should receive a verification email within 30 seconds
    And the email subject should contain "Verify your account"
    And the email body should contain a verification link

Step definitions: `features/steps/signup_steps.py`

import re
from behave import given, when, then
import oneshot


@given('I have a temporary email address labelled "{label}"')
def step_create_address(context, label):
    run_id = context.config.userdata.get("run_id", "local")
    context.address = oneshot.create(
        ttl_seconds=300,
        label=f"{label}-{run_id}",
    )
    context.email_address = context.address.address


@when("I sign up with that email address")
def step_signup(context):
    # Replace with your app's signup API
    context.app_client.post("/signup", json={
        "email": context.email_address,
        "password": "SecureP@ss1",
    })


@then("I should receive a verification email within {timeout:d} seconds")
def step_wait_for_email(context, timeout):
    context.email = oneshot.wait_for_email(context.address.id, timeout=timeout)


@then('the email subject should contain "{text}"')
def step_check_subject(context, text):
    assert text in context.email.subject, (
        f"Expected '{text}' in subject, got: {context.email.subject}"
    )


@then("the email body should contain a verification link")
def step_check_verification_link(context):
    body = context.email.text_body or context.email.html_body
    match = re.search(r"https?://\S+/verify\S*", body)
    assert match, f"No verification link found in email body"
    context.verification_link = match.group(0)

Environment setup: `features/environment.py`

import os
import uuid
import oneshot


def before_all(context):
    oneshot.configure(api_key=os.environ["ONESHOT_API_KEY"])
    context.run_id = str(uuid.uuid4())[:8]
    context.config.userdata["run_id"] = context.run_id


def after_all(context):
    # Clean up all addresses from this test run
    oneshot.delete_by_label(f"signup-{context.run_id}")


def after_scenario(context, scenario):
    # Optional: clean up per-scenario if you prefer
    if hasattr(context, "address"):
        try:
            oneshot.delete(context.address.id)
        except oneshot.NotFoundError:
            pass  # Already cleaned up or expired

Running

pip install behave oneshot-mail
export ONESHOT_API_KEY="osm_live_your_key"
behave

pytest integration

Fixture: `conftest.py`

import os
import uuid
import pytest
import oneshot


@pytest.fixture(scope="session")
def oneshot_client():
    """Shared OneShotMail client for the test session."""
    client = oneshot.Client(api_key=os.environ["ONESHOT_API_KEY"])
    yield client
    client.close()


@pytest.fixture(scope="session")
def run_label():
    """Unique label for this test run, used for bulk cleanup."""
    return f"pytest-{uuid.uuid4().hex[:8]}"


@pytest.fixture
def email_address(oneshot_client, run_label):
    """Create a temporary email address for a single test."""
    addr = oneshot_client.create(ttl_seconds=300, label=run_label)
    yield addr
    try:
        oneshot_client.delete(addr.id)
    except oneshot.NotFoundError:
        pass


@pytest.fixture(scope="session", autouse=True)
def cleanup_all(oneshot_client, run_label):
    """After the entire test session, bulk-delete all addresses by label."""
    yield
    oneshot_client.delete_by_label(run_label)

Test file: `tests/test_signup.py`

import oneshot


def test_signup_sends_verification_email(oneshot_client, email_address, app):
    # Trigger signup
    app.post("/signup", json={
        "email": email_address.address,
        "name": "Test User",
    })

    # Wait for the verification email
    email = oneshot_client.wait_for_email(email_address.id, timeout=30)

    assert "Verify your account" in email.subject
    assert "Test User" in email.text_body


def test_password_reset_sends_email(oneshot_client, email_address, app):
    # Create the user first, then trigger reset
    app.post("/signup", json={"email": email_address.address, "name": "Test"})
    app.post("/password-reset", json={"email": email_address.address})

    email = oneshot_client.wait_for_email(email_address.id, timeout=30)
    assert "Reset your password" in email.subject

Parallel execution with pytest-xdist

The label-based cleanup pattern works perfectly with pytest-xdist:

pip install pytest-xdist
pytest -n 4 --dist loadfile

Each worker gets its own email_address fixture, so there is no cross-contamination between parallel tests.

unittest integration

import os
import unittest
import oneshot


class EmailTestCase(unittest.TestCase):
    """Base test case with OneShotMail support."""

    @classmethod
    def setUpClass(cls):
        cls.client = oneshot.Client(api_key=os.environ["ONESHOT_API_KEY"])
        cls.addresses = []

    @classmethod
    def tearDownClass(cls):
        for addr_id in cls.addresses:
            try:
                cls.client.delete(addr_id)
            except oneshot.NotFoundError:
                pass
        cls.client.close()

    def create_address(self, label=None, ttl=300):
        addr = self.client.create(ttl_seconds=ttl, label=label)
        self.addresses.append(addr.id)
        return addr


class TestSignup(EmailTestCase):
    def test_verification_email(self):
        addr = self.create_address(label="test-signup")

        # Trigger your app's signup
        self.app.signup(email=addr.address)

        email = self.client.wait_for_email(addr.id, timeout=30)
        self.assertIn("Verify your account", email.subject)

Installation

Configuration

From environment variable (recommended)

Explicit client

Module-level configuration

Context manager

API Reference

create(ttl_seconds, label, mode)

get(address_id)

get_email(address_id)

get_email_raw(address_id)

download_attachment(address_id, index)

wait_for_email(address_id, timeout, poll_interval)

send(to, subject, text_body, html_body, attachments, ttl_seconds, label)

list(status, label, mode, limit, cursor)

delete(address_id)

delete_by_label(label)

account()

buy_credits(amount)

health()

save_attachments(address_id, directory)