Postject Python SDK

Official Python SDK for the Postject email API.

Installation

pip install postject

Quick Start

from postject import Postject

client = Postject(api_key="pk_live_your_api_key_here")

# Send a simple email
result = client.send({
    "to": "customer@example.com",
    "subject": "Welcome to our platform!",
    "html": "<h1>Welcome!</h1><p>Thanks for signing up.</p>"
})

print(f"Message ID: {result.id}")
print(f"Status: {result.status}")

Features

✅ Full type hints (Pydantic models) ✅ Sync and async support ✅ Automatic retry on rate limits ✅ Comprehensive error handling ✅ All API endpoints covered

Configuration

client = Postject(
    api_key="pk_live_your_api_key_here",
    base_url="https://api.postject.com",  # optional
    timeout=30,  # optional, in seconds
    retries=3,   # optional, for rate limits
)

Sending Emails

Simple Email

result = client.send({
    "to": "user@example.com",
    "from": "noreply@yourapp.com",  # optional if identity configured
    "subject": "Hello World",
    "html": "<p>This is a test email</p>",
    "text": "This is a test email",  # optional plain text
})

Multiple Recipients

client.send({
    "to": ["user1@example.com", "user2@example.com", "user3@example.com"],
    "subject": "Bulk announcement",
    "html": "<p>Important update</p>",
})

Using Templates

result = client.send_with_template(
    template_id="template_id_here",
    to="user@example.com",
    variables={
        "name": "John Doe",
        "order_number": "12345",
        "order_total": "$99.00",
    },
    stream_id="transactional_stream_id",  # optional
)

Type-Safe Request

from postject import SendEmailRequest

request = SendEmailRequest(
    to="user@example.com",
    subject="Order Confirmation",
    html="<p>Your order has been confirmed</p>",
    tags=["order", "confirmation"],
    metadata={"order_id": "12345"},
    idempotency_key="order_12345",  # prevents duplicates
    analyze_content=True,  # check spam score
)

result = client.send(request)

Managing Resources

Servers

# List all servers
servers = client.get_servers()

# Create a server
server = client.create_server("Production Server")

# Delete a server
client.delete_server("server_id")

Message Streams

# List streams
streams = client.get_streams("server_id")

# Create a stream
stream = client.create_stream(
    server_id="server_id",
    name="Transactional Emails",
    slug="transactional",
    description="Order confirmations, password resets, etc.",
    stream_type="transactional",
)

# Delete a stream
client.delete_stream("server_id", "stream_id")

Templates

# List templates
templates = client.get_templates("server_id")

# Create a template
template = client.create_template(
    server_id="server_id",
    name="Welcome Email",
    subject="Welcome to {{company_name}}!",
    html="<h1>Welcome {{user_name}}!</h1><p>Thanks for signing up.</p>",
)

# Update a template
client.update_template(
    server_id="server_id",
    template_id="template_id",
    subject="Updated subject",
    html="<p>Updated content</p>",
)

# Delete a template
client.delete_template("server_id", "template_id")

Webhooks

# Create a webhook
webhook = client.create_webhook(
    stream_id="stream_id",
    url="https://yourapp.com/webhooks/postject",
    events=["delivered", "bounced", "complained"],  # or ["*"] for all
)

print(f"Webhook secret: {webhook.secret}")  # Use for signature verification

# List webhooks
webhooks = client.get_webhooks("stream_id")

# Get webhook logs
logs_data = client.get_webhook_logs("webhook_id", limit=50, offset=0)
print(f"Total logs: {logs_data['total']}")
for log in logs_data['logs']:
    print(f"{log.event_type}: {log.status_code}")

# Test a webhook
client.test_webhook("webhook_id")

# Delete a webhook
client.delete_webhook("stream_id", "webhook_id")

Message Tracking

# Get message details
message = client.get_message("message_id")
print(f"Status: {message.status}")
print(f"Route: {message.route_name}")
print(f"Latency: {message.latency_ms}ms")

# Get message events (full timeline)
events = client.get_message_events("message_id")
for event in events:
    print(f"{event.type} at {event.timestamp}")

API Key Management

# List API keys
keys = client.get_api_keys()

# Create a new key
new_key = client.create_api_key(
    name="Production Key",
    expires_in_days=365,
)
print(f"New API key: {new_key.key}")  # SAVE THIS!

# Revoke a key
client.revoke_api_key("key_id")

# Rotate a key
rotated_key = client.rotate_api_key("old_key_id", name="New Production Key")

Content Analysis

# Analyze content before sending
analysis = client.analyze_content(
    subject="Your subject line",
    html="<html><body>Email content</body></html>",
)

print(f"Overall score: {analysis.overall_score}/100")
print(f"Spam score: {analysis.spam_score}/100")

for issue in analysis.issues:
    print(f"[{issue.severity}] {issue.message}")
    print(f"  Suggestion: {issue.suggestion}")

for recommendation in analysis.recommendations:
    print(f"✓ {recommendation}")

Analytics

# Get server overview
analytics = client.get_server_analytics("server_id")

print(f"Total sent: {analytics.total_sent}")
print(f"Delivery rate: {analytics.delivery_rate * 100:.2f}%")
print(f"Bounce rate: {analytics.bounce_rate * 100:.2f}%")
print(f"Avg latency: {analytics.avg_latency}ms")

Error Handling

from postject import (
    PostjectError,
    AuthenticationError,
    RateLimitError,
    ValidationError,
    NotFoundError,
    ServerError,
)

try:
    client.send({"to": "user@example.com", "subject": "Test", "html": "..."})
except AuthenticationError:
    print("Invalid API key")
except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after}s")
except ValidationError as e:
    print(f"Validation error: {e.response}")
except NotFoundError:
    print("Resource not found")
except ServerError as e:
    print(f"Server error: {e.status_code}")
except PostjectError as e:
    print(f"API error: {e.message}")

Async Support

import asyncio
from postject import AsyncPostject

async def send_email():
    async with AsyncPostject(api_key="your_api_key") as client:
        result = await client.send({
            "to": "user@example.com",
            "subject": "Async Email",
            "html": "<p>Sent asynchronously</p>",
        })
        print(f"Sent: {result.id}")

asyncio.run(send_email())

Context Manager

# Automatic client cleanup
with Postject(api_key="your_api_key") as client:
    result = client.send({
        "to": "user@example.com",
        "subject": "Test",
        "html": "<p>Email body</p>",
    })
# Client automatically closed

Type Hints

Full type hints using Pydantic:

from postject.types import (
    SendEmailRequest,
    SendEmailResponse,
    Server,
    MessageStream,
    Template,
    Webhook,
    Message,
    Event,
)

def send_welcome_email(email: str) -> SendEmailResponse:
    client = Postject(api_key="your_key")
    request = SendEmailRequest(
        to=email,
        subject="Welcome!",
        html="<p>Welcome to our platform</p>",
    )
    return client.send(request)

Examples

# Send transactional email
client.send({
    "to": "customer@example.com",
    "subject": "Your order #12345 has shipped",
    "html": """
        <h2>Order Shipped!</h2>
        <p>Your order #12345 is on its way.</p>
        <p>Tracking: ABC123XYZ</p>
    """,
    "tags": ["shipping", "order"],
    "metadata": {"order_id": "12345"},
})

# Send password reset
client.send_with_template(
    template_id="password_reset_template",
    to="user@example.com",
    variables={
        "reset_link": "https://app.com/reset?token=abc123",
        "expires_in": "1 hour",
    },
)

# Batch send with template
users = [
    {"email": "user1@example.com", "name": "Alice"},
    {"email": "user2@example.com", "name": "Bob"},
]

for user in users:
    client.send_with_template(
        template_id="newsletter_template",
        to=user["email"],
        variables={"name": user["name"]},
        stream_id="newsletter_stream",
    )

Requirements

• Python 3.8+
• httpx>=0.24.0
• pydantic>=2.0.0

Support

• Documentation: https://postject.com/docs/sdk
• Email: support@postject.com

License

MIT