postject 1.0.1

Official Python SDK for Postject email API

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": "[e-mail]",
    "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": "[e-mail]",
    "from": "[e-mail]",  # 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": ["[e-mail]", "[e-mail]", "[e-mail]"],
    "subject": "Bulk announcement",
    "html": "<p>Important update</p>",
})

Using Templates

result = client.send_with_template(
    template_id="template_id_here",
    to="[e-mail]",
    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="[e-mail]",
    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": "[e-mail]", "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": "[e-mail]",
            "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": "[e-mail]",
        "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": "[e-mail]",
    "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="[e-mail]",
    variables={
        "reset_link": "https://app.com/reset?token=abc123",
        "expires_in": "1 hour",
    },
)

# Batch send with template
users = [
    {"email": "[e-mail]", "name": "Alice"},
    {"email": "[e-mail]", "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: [e-mail]

License

MIT