pip install northrelay

import asyncio
import os

from northrelay import NorthRelay, SendEmailRequest


async def main() -> None:
    async with NorthRelay(api_key=os.environ["NORTHRELAY_API_KEY"]) as client:
        response = await client.emails.send(
            SendEmailRequest(
                from_={"email": "you@yourdomain.com", "name": "Your App"},
                to=[{"email": "recipient@example.com"}],
                content={
                    "subject": "Hello from NorthRelay",
                    "html": "<h1>Welcome!</h1><p>Your first email via NorthRelay.</p>",
                },
            )
        )
        print("Queued:", response.message_id)


asyncio.run(main())

client = NorthRelay(
    api_key="nr_live_xxx",                  # required: nr_live_, nr_test_ or nr_app_
    base_url="https://app.northrelay.ca",   # optional, defaults shown
    timeout=30.0,                           # seconds
    max_retries=3,
    retry_delay=1.0,                        # seconds, exponential backoff
    max_retry_delay=10.0,
)

from northrelay import NorthRelayError

# Subscribe with double opt-in and consent evidence (email is positional)
try:
    result = await client.subscriptions.subscribe(
        "jane@example.com",
        list_ids=["list_id"],
        topic_ids=["topic_id"],
        double_opt_in=True,
        redirect_url="https://yourapp.com/newsletter/confirmed",
        consent={"ip": "203.0.113.24", "user_agent": "Mozilla/5.0 ...", "text": "Yes, send me the monthly newsletter."},
    )
    print(result.status, result.requires_confirmation, result.confirmation_sent)
except NorthRelayError as error:
    if error.details.get("code") == "RECIPIENT_UNSUBSCRIBED":
        ...  # retry with resubscribe=True only after fresh consent
    else:
        raise

# Preference page: read, then save changes
status = await client.subscriptions.get("jane@example.com")
for topic in status.topics:
    print(topic.id, topic.label or topic.name, topic.subscribed)

status = await client.subscriptions.update_preferences(
    "jane@example.com",
    topics={"topic_id": False},
    lists={"list_id": True},
)

# Opt out from your server: scope "all" (default), "list" or "topic"
result = await client.subscriptions.unsubscribe("jane@example.com", scope="topic", topic_id="topic_id")
print(result.changed)

# Re-send the confirmation email for a PENDING contact
sent = await client.subscriptions.resend_confirmation("jane@example.com")
print(sent.expires_at)

contact = await client.contacts.create(
    {"email": "user@example.com", "first_name": "Jane", "last_name": "Doe", "tags": ["trial"]}
)
page = await client.contacts.list(status="ACTIVE", tag="trial", limit=50)
contact = await client.contacts.update(contact.id, first_name="Janet")
await client.contacts.add_tags(contact.id, ["vip"])
await client.contacts.remove_tag(contact.id, "trial")
await client.contacts.bulk_upsert([{"email": "a@example.com"}, {"email": "b@example.com"}], skip_duplicates=True)
await client.contacts.bulk_delete(["contact_id_1", "contact_id_2"])
await client.contacts.delete(contact.id)

lst = await client.contacts.create_list("Product updates", topic_id="topic_id")  # tag with a topic

added = await client.contacts.add_to_list(lst.id, emails=["jane@example.com", "new@example.com"], create_missing=True)
print(added.added, added.already_members, added.not_found, added.blocked)

members = await client.contacts.list_members(lst.id, filter="active", limit=100)
print(members.counts, members.total, members.has_more)

await client.contacts.remove_from_list(lst.id, emails=["jane@example.com"])

result = await client.contacts.import_csv(
    "subscribers.csv",                     # path, or the CSV content as bytes
    mappings=[{"csvColumn": "Email Address", "field": "email"}, {"csvColumn": "First", "field": "firstName"}],
    list_id="list_id",
    tags=["imported", "spring-2026"],
)
print(result.imported, result.skipped, result.added_to_list, result.errors)

topic = await client.topics.create("product-updates", "Release notes")
topic_id = topic["data"]["id"]

await client.topics.update(
    topic_id,
    public_label="Product updates",
    public_description="New features, about twice a month.",
    is_published_on_unsub=True,
)

published = await client.topics.list(published=True)
opt_outs = await client.topics.list_members(topic_id, page=1, limit=25)
await client.topics.add_member(topic_id, "jane@example.com")     # opt out
await client.topics.remove_member(topic_id, "jane@example.com")  # opt back in

created = await client.forms.create(list_id="list_id", name="Newsletter", slug="newsletter")
form_id = created["form"]["id"]

await client.forms.update(form_id, status="PUBLISHED", require_confirmation=True, success_redirect_url="https://yourapp.com/thanks")
await client.forms.set_fields(form_id, [
    {"fieldKey": "email", "label": "Email", "type": "EMAIL", "required": True, "bindsToContact": "EMAIL"},
    {"fieldKey": "first_name", "label": "First name", "type": "TEXT", "bindsToContact": "FIRST_NAME"},
])
forms = await client.forms.list(limit=20)
await client.forms.delete(form_id)  # archives the form

from northrelay import CreateWebhookRequest, UpdateWebhookRequest, WebhookEventType

webhook = await client.webhooks.create(
    CreateWebhookRequest(
        url="https://yourapp.com/webhooks/northrelay",
        events=[WebhookEventType.CONTACT_UNSUBSCRIBED, WebhookEventType.LIST_MEMBER_REMOVED, "email.bounced"],
        description="Newsletter sync",
    )
)
print(webhook.secret)  # shown once: store it

await client.webhooks.update(webhook.id, UpdateWebhookRequest(active=False))
new_secret = await client.webhooks.rotate_secret(webhook.id)
await client.webhooks.delete(webhook.id)

await client.suppressions.add("optout@example.com")              # reason defaults to "Manual"
await client.suppressions.add("bounced@example.com", reason="Bounce")
status = await client.suppressions.check("optout@example.com")   # 404 NotFoundError when not suppressed
await client.suppressions.remove("optout@example.com")
page = await client.suppressions.list()

page = (await client.designs.list_page(application_key="account", search="welcome"))["data"]
design = (await client.designs.get(page["items"][0]["id"]))["data"]
await client.designs.preview(design["id"], sample=True, brand_id=other_brand_id)
await client.designs.publish(design["id"], design["revision"])
await client.designs.send(design["id"], to=[{"email": "ana@example.com"}],
                          variables={"first_name": "Ana"}, idempotency_key="welcome-ana-1")
usage = (await client.designs.usage(design["id"]))["data"]

asset = (await client.designs.upload_asset(image_base64))["data"]   # PNG, JPEG or WebP
brand = (await client.designs.create_brand(theme={"name": "Acme", "logoUrl": asset["url"]}))["data"]
await client.designs.set_default_brand(brand["id"])
affected = (await client.designs.brand_usage(brand["id"]))["data"]

from northrelay import (
    AuthenticationError,
    NorthRelayError,
    NotFoundError,
    QuotaExceededError,
    RateLimitError,
    ScopeError,
    ServerError,
    ValidationError,
)

try:
    await client.subscriptions.subscribe("jane@example.com", list_ids=["list_id"])
except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after} seconds.")
except ValidationError as e:
    print("Validation failed:", e.message, e.errors)
except NotFoundError as e:
    print(e.details.get("code"), e.details.get("fix_action"))  # e.g. LIST_NOT_FOUND
except NorthRelayError as e:
    print(e.status_code, e.details.get("code"), e.details.get("fix_action"))

import hashlib
import hmac
import os

from fastapi import FastAPI, HTTPException, Request
from northrelay import WebhookPayload

app = FastAPI()
SECRET = os.environ["NORTHRELAY_WEBHOOK_SECRET"]


@app.post("/webhooks/northrelay")
async def northrelay_webhook(request: Request):
    raw = await request.body()
    expected = hmac.new(SECRET.encode(), raw, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(expected, request.headers.get("X-NorthRelay-Signature", "")):
        raise HTTPException(status_code=401, detail="Invalid signature")

    event = WebhookPayload.model_validate_json(raw)
    if event.event_type == "contact.unsubscribed":
        scope = event.details.get("scope")          # "all" | "list" | "topic"
        email = event.details.get("contactEmail")
        ...
    return {"received": True}