Delivery Strategies

VaultSandbox Python Client supports two email delivery strategies: Server-Sent Events (SSE) for real-time updates and Polling for compatibility. SSE is the default strategy, providing instant email notifications.

Overview

When you wait for emails or subscribe to new email notifications, the SDK needs to know when emails arrive. It does this using one of two strategies:

1. SSE (Server-Sent Events): Real-time push notifications from the server
2. Polling: Periodic checking for new emails

Strategy Comparison

Feature	SSE	Polling
Latency	Near-instant (~100ms)	Poll interval (default: 2s)
Server Load	Lower (persistent connection)	Higher (repeated requests)
Network Traffic	Lower (only when emails arrive)	Higher (constant polling)
Compatibility	Requires persistent connections	Works everywhere
Firewall/Proxy	May be blocked	Always works
Resource Impact	Lower (push-based)	Higher (constant requests)

SSE Strategy (Default)

Server-Sent Events provide real-time push notifications when emails arrive.

Advantages

• Near-instant delivery: Emails appear within milliseconds
• Lower server load: Single persistent connection
• Efficient: Only transmits when emails arrive
• Resource-friendly: No constant polling

Configuration

SSE is the default strategy, so you don’t need to specify it explicitly:

from vaultsandbox import VaultSandboxClient

# SSE is used by default
async with VaultSandboxClient(
    api_key=os.environ["VAULTSANDBOX_API_KEY"],
    base_url="https://smtp.vaultsandbox.com",
    sse_reconnect_interval=5000,      # Wait 5s before reconnecting
    sse_max_reconnect_attempts=10,    # Try up to 10 reconnections
) as client:
    inbox = await client.create_inbox()
    # ...

SSE Configuration Options

Option	Type	Default	Description
sse_reconnect_interval	int	5000	Initial delay before reconnection (ms)
sse_max_reconnect_attempts	int	10	Maximum reconnection attempts

Reconnection Behavior

SSE uses exponential backoff for reconnections:

1st attempt: sse_reconnect_interval (5s)
2nd attempt: sse_reconnect_interval * 2 (10s)
3rd attempt: sse_reconnect_interval * 4 (20s)
...up to sse_max_reconnect_attempts

Example Usage

import asyncio
import os
import re
from vaultsandbox import VaultSandboxClient

async def main():
    # SSE is used by default
    async with VaultSandboxClient(
        api_key=os.environ["VAULTSANDBOX_API_KEY"],
        base_url="https://smtp.vaultsandbox.com",
    ) as client:
        inbox = await client.create_inbox()

        # Real-time subscription (uses SSE)
        async def handle_email(email):
            print(f"Instant notification: {email.subject}")

        unsubscribe = await inbox.on_new_email(handle_email)

        # Waiting also uses SSE (faster than polling)
        email = await inbox.wait_for_email(
            timeout=10000,
            subject=re.compile(r"Welcome"),
        )

        await unsubscribe()

asyncio.run(main())

When to Use SSE

• Real-time monitoring: When you need instant email notifications
• Long-running tests: Reduces overall test time
• High email volume: More efficient than polling
• Development/local: Fast feedback during development

Limitations

• Requires persistent HTTP connection support
• May not work behind some corporate proxies
• Some cloud environments may close long-lived connections
• Requires server-side SSE support

Polling Strategy

Polling periodically checks for new emails at a configured interval.

Advantages

• Universal compatibility: Works in all environments
• Firewall-friendly: Standard HTTP requests
• Predictable: Easy to reason about behavior
• Resilient: Automatically recovers from transient failures

Configuration

from vaultsandbox import VaultSandboxClient, DeliveryStrategyType

async with VaultSandboxClient(
    api_key=os.environ["VAULTSANDBOX_API_KEY"],
    base_url="https://smtp.vaultsandbox.com",
    strategy=DeliveryStrategyType.POLLING,
    polling_interval=2000,     # Check every 2 seconds
    polling_max_backoff=30000, # Max backoff delay
) as client:
    inbox = await client.create_inbox()
    # ...

Polling Configuration Options

Option	Type	Default	Description
polling_interval	int	2000	How often to poll for emails (ms)
polling_max_backoff	int	30000	Maximum backoff delay (ms)

Backoff Behavior

Polling uses exponential backoff with jitter when no new emails arrive:

• Backoff multiplier: 1.5x
• Jitter factor: 30% random variation
• Maximum backoff: Configurable via polling_max_backoff

When new emails arrive, the backoff resets to polling_interval.

Example Usage

import asyncio
import re
from vaultsandbox import VaultSandboxClient, DeliveryStrategyType

async def main():
    async with VaultSandboxClient(
        api_key=os.environ["VAULTSANDBOX_API_KEY"],
        base_url="https://smtp.vaultsandbox.com",
        strategy=DeliveryStrategyType.POLLING,
        polling_interval=1000,  # Poll every 1 second
    ) as client:
        inbox = await client.create_inbox()

        # Polling-based subscription
        async def handle_email(email):
            print(f"Polled notification: {email.subject}")

        unsubscribe = await inbox.on_new_email(handle_email)

        # Waiting uses polling (checks every polling_interval)
        email = await inbox.wait_for_email(
            timeout=10000,
            subject=re.compile(r"Welcome"),
        )

        await unsubscribe()

asyncio.run(main())

Choosing Poll Interval

Different intervals suit different scenarios:

from vaultsandbox import VaultSandboxClient, DeliveryStrategyType

# Fast polling (500ms) - Development/local testing
async with VaultSandboxClient(
    api_key="dev-key",
    base_url="http://localhost:3000",
    strategy=DeliveryStrategyType.POLLING,
    polling_interval=500,
) as fast_client:
    pass

# Standard polling (2000ms) - Default, good balance
async with VaultSandboxClient(
    api_key=os.environ["VAULTSANDBOX_API_KEY"],
    base_url="https://smtp.vaultsandbox.com",
    strategy=DeliveryStrategyType.POLLING,
    polling_interval=2000,
) as standard_client:
    pass

# Slow polling (5000ms) - CI/CD or rate-limited environments
async with VaultSandboxClient(
    api_key=os.environ["VAULTSANDBOX_API_KEY"],
    base_url="https://smtp.vaultsandbox.com",
    strategy=DeliveryStrategyType.POLLING,
    polling_interval=5000,
) as slow_client:
    pass

When to Use Polling

• Corporate networks: Restrictive firewall/proxy environments
• CI/CD pipelines: Guaranteed compatibility
• Rate-limited APIs: Control request frequency
• Debugging: Predictable request timing
• Low email volume: Polling overhead is minimal

Performance Optimization

For wait_for_email(), you can override the polling interval:

from vaultsandbox import WaitForEmailOptions

# Default client polling: 2s
async with VaultSandboxClient(
    api_key=os.environ["VAULTSANDBOX_API_KEY"],
    strategy=DeliveryStrategyType.POLLING,
    polling_interval=2000,
) as client:
    inbox = await client.create_inbox()

    # Override for specific operation (faster)
    email = await inbox.wait_for_email(
        timeout=30000,
        poll_interval=1000,  # Check every 1s for this operation
    )

Choosing the Right Strategy

Use SSE (Default)

SSE is the default strategy and is best for most use cases:

from vaultsandbox import VaultSandboxClient

async with VaultSandboxClient(
    api_key=os.environ["VAULTSANDBOX_API_KEY"],
    base_url=os.environ["VAULTSANDBOX_URL"],
    # SSE is used by default
) as client:
    pass

Best for:

• General testing
• Real-time monitoring dashboards
• High-volume email testing
• Latency-sensitive tests
• Local development
• When you want instant email notifications

Caveat: Requires persistent HTTP connection support. Will raise SSEError if SSE is unavailable.

Use Polling

When compatibility is more important than speed:

import os
from vaultsandbox import VaultSandboxClient, DeliveryStrategyType

polling_interval = 3000 if os.environ.get("CI") else 1000

async with VaultSandboxClient(
    api_key=os.environ["VAULTSANDBOX_API_KEY"],
    base_url=os.environ["VAULTSANDBOX_URL"],
    strategy=DeliveryStrategyType.POLLING,
    polling_interval=polling_interval,
) as client:
    pass

Best for:

• CI/CD environments (guaranteed to work)
• Corporate networks with restrictive proxies
• When SSE is known to be problematic
• Rate-limited scenarios

Environment-Specific Configuration

Development

Fast feedback with SSE (default):

import os
from vaultsandbox import VaultSandboxClient

# SSE is the default, providing instant email notifications
async with VaultSandboxClient(
    api_key=os.environ["VAULTSANDBOX_API_KEY"],
    base_url=os.environ.get("VAULTSANDBOX_URL", "http://localhost:3000"),
) as client:
    pass

CI/CD

Reliable polling:

.env.ci

# VAULTSANDBOX_URL=https://smtp.vaultsandbox.com
# VAULTSANDBOX_STRATEGY=polling
# VAULTSANDBOX_POLLING_INTERVAL=3000

import os
from vaultsandbox import VaultSandboxClient, DeliveryStrategyType

async with VaultSandboxClient(
    api_key=os.environ["VAULTSANDBOX_API_KEY"],
    base_url=os.environ["VAULTSANDBOX_URL"],
    strategy=DeliveryStrategyType.POLLING,
    polling_interval=int(os.environ.get("VAULTSANDBOX_POLLING_INTERVAL", "2000")),
) as client:
    pass

Production Testing

SSE with tuned reconnection:

import os
from vaultsandbox import VaultSandboxClient

async with VaultSandboxClient(
    api_key=os.environ["VAULTSANDBOX_API_KEY"],
    base_url=os.environ["VAULTSANDBOX_URL"],
    # SSE configuration
    sse_reconnect_interval=10000,
    sse_max_reconnect_attempts=5,
) as client:
    pass

Monitoring Strategy Performance

Measure Email Delivery Latency

import asyncio
import time
from vaultsandbox import VaultSandboxClient

async def measure_delivery_latency(send_test_email):
    async with VaultSandboxClient(
        api_key=os.environ["VAULTSANDBOX_API_KEY"],
        base_url=os.environ["VAULTSANDBOX_URL"],
    ) as client:
        inbox = await client.create_inbox()

        start_time = time.time()

        # Send email
        await send_test_email(inbox.email_address)

        # Wait for email
        email = await inbox.wait_for_email(timeout=30000)

        latency = (time.time() - start_time) * 1000
        print(f"Email delivery latency: {latency:.0f}ms")

        await inbox.delete()

asyncio.run(measure_delivery_latency(send_test_email))

Compare Strategies

import asyncio
import os
import time
from vaultsandbox import VaultSandboxClient, DeliveryStrategyType

async def compare_strategies(send_test_email):
    # Test SSE (default)
    async with VaultSandboxClient(
        api_key=os.environ["VAULTSANDBOX_API_KEY"],
        base_url=os.environ["VAULTSANDBOX_URL"],
    ) as sse_client:
        sse_inbox = await sse_client.create_inbox()
        sse_start = time.time()
        await send_test_email(sse_inbox.email_address)
        await sse_inbox.wait_for_email(timeout=10000)
        sse_latency = (time.time() - sse_start) * 1000
        await sse_inbox.delete()

    # Test Polling
    async with VaultSandboxClient(
        api_key=os.environ["VAULTSANDBOX_API_KEY"],
        base_url=os.environ["VAULTSANDBOX_URL"],
        strategy=DeliveryStrategyType.POLLING,
        polling_interval=2000,
    ) as poll_client:
        poll_inbox = await poll_client.create_inbox()
        poll_start = time.time()
        await send_test_email(poll_inbox.email_address)
        await poll_inbox.wait_for_email(timeout=10000)
        poll_latency = (time.time() - poll_start) * 1000
        await poll_inbox.delete()

    print(f"SSE latency: {sse_latency:.0f}ms")
    print(f"Polling latency: {poll_latency:.0f}ms")
    print(f"Difference: {poll_latency - sse_latency:.0f}ms")

asyncio.run(compare_strategies(send_test_email))

Troubleshooting

SSE Connection Failures

from vaultsandbox import VaultSandboxClient, DeliveryStrategyType, SSEError

try:
    # SSE is the default strategy
    async with VaultSandboxClient(
        api_key=os.environ["VAULTSANDBOX_API_KEY"],
        base_url=os.environ["VAULTSANDBOX_URL"],
    ) as client:
        inbox = await client.create_inbox()
        # Use inbox...
except SSEError as e:
    print(f"SSE failed: {e}")
    print("Falling back to polling...")

    # Recreate with polling
    async with VaultSandboxClient(
        api_key=os.environ["VAULTSANDBOX_API_KEY"],
        base_url=os.environ["VAULTSANDBOX_URL"],
        strategy=DeliveryStrategyType.POLLING,
    ) as client:
        inbox = await client.create_inbox()
        # Continue with polling...

Polling Too Slow

If emails arrive slowly with polling:

from vaultsandbox import VaultSandboxClient, DeliveryStrategyType

# Problem: Default 2s polling is too slow

# Solution 1: Faster polling
async with VaultSandboxClient(
    api_key=os.environ["VAULTSANDBOX_API_KEY"],
    base_url=os.environ["VAULTSANDBOX_URL"],
    strategy=DeliveryStrategyType.POLLING,
    polling_interval=500,  # Check every 500ms
) as faster_client:
    pass

# Solution 2: Use SSE (default) for real-time delivery
async with VaultSandboxClient(
    api_key=os.environ["VAULTSANDBOX_API_KEY"],
    base_url=os.environ["VAULTSANDBOX_URL"],
) as sse_client:
    pass

# Solution 3: Override for specific wait
async with VaultSandboxClient(
    api_key=os.environ["VAULTSANDBOX_API_KEY"],
    base_url=os.environ["VAULTSANDBOX_URL"],
    strategy=DeliveryStrategyType.POLLING,
    polling_interval=2000,
) as client:
    inbox = await client.create_inbox()
    email = await inbox.wait_for_email(
        timeout=10000,
        poll_interval=500,  # Fast polling for this operation
    )

Best Practices

1. Use SSE Strategy by Default

SSE is the default and recommended strategy for most use cases:

from vaultsandbox import VaultSandboxClient, DeliveryStrategyType

# Good: Use SSE (default)
async with VaultSandboxClient(
    api_key=os.environ["VAULTSANDBOX_API_KEY"],
    base_url=os.environ["VAULTSANDBOX_URL"],
) as client:
    pass

# Only specify polling when needed
async with VaultSandboxClient(
    api_key=os.environ["VAULTSANDBOX_API_KEY"],
    base_url=os.environ["VAULTSANDBOX_URL"],
    strategy=DeliveryStrategyType.POLLING,  # CI needs guaranteed compatibility
) as ci_client:
    pass

2. Tune for Environment

Configure differently for each environment:

import os
from vaultsandbox import VaultSandboxClient, DeliveryStrategyType

def create_client():
    api_key = os.environ["VAULTSANDBOX_API_KEY"]
    base_url = os.environ["VAULTSANDBOX_URL"]

    if os.environ.get("CI"):
        # CI: Reliable polling
        return VaultSandboxClient(
            api_key=api_key,
            base_url=base_url,
            strategy=DeliveryStrategyType.POLLING,
            polling_interval=3000,
        )
    else:
        # Default: SSE with tuning
        return VaultSandboxClient(
            api_key=api_key,
            base_url=base_url,
            sse_reconnect_interval=5000,
            sse_max_reconnect_attempts=10,
        )

3. Handle SSE Errors Gracefully

If SSE fails, you can fall back to polling:

from vaultsandbox import VaultSandboxClient, DeliveryStrategyType, SSEError

async def create_client_with_fallback():
    try:
        # SSE is default
        client = VaultSandboxClient(
            api_key=os.environ["VAULTSANDBOX_API_KEY"],
            base_url=os.environ["VAULTSANDBOX_URL"],
        )
        # Test connection
        await client.check_key()
        return client
    except SSEError:
        print("SSE unavailable, using polling")
        return VaultSandboxClient(
            api_key=os.environ["VAULTSANDBOX_API_KEY"],
            base_url=os.environ["VAULTSANDBOX_URL"],
            strategy=DeliveryStrategyType.POLLING,
        )

4. Don’t Poll Too Aggressively

Avoid very short polling intervals in production:

from vaultsandbox import VaultSandboxClient, DeliveryStrategyType

# Avoid: Too aggressive
async with VaultSandboxClient(
    api_key=os.environ["VAULTSANDBOX_API_KEY"],
    base_url=os.environ["VAULTSANDBOX_URL"],
    strategy=DeliveryStrategyType.POLLING,
    polling_interval=100,  # 100ms - too frequent!
) as aggressive_client:
    pass

# Good: Reasonable interval
async with VaultSandboxClient(
    api_key=os.environ["VAULTSANDBOX_API_KEY"],
    base_url=os.environ["VAULTSANDBOX_URL"],
    strategy=DeliveryStrategyType.POLLING,
    polling_interval=2000,  # 2s - balanced
) as balanced_client:
    pass

Next Steps

• Real-time Monitoring Guide - Using subscriptions
• Configuration Reference - All config options
• Error Handling - Handle SSE errors
• CI/CD Integration - Strategy for CI environments