Campaigns¶

The Konigle SDK provides campaign management for sending messages across multiple channels. Currently supports email campaigns with comprehensive execution tracking, scheduling, and revenue attribution.

The email campaign can be spread over a specified duration to manage sending rates and optimize deliverability.

Creating Campaigns¶

Basic Email Campaign¶

import konigle
from konigle.models.comm import CampaignCreate

client = konigle.Client(api_key="your-api-key")

# Create a basic email campaign
campaign_data = CampaignCreate(
    name="Weekly Newsletter",
    channel_type="email",
    email_channel="channel_123",
    email_template="template_456",
    audience="audience_789",
    description="Weekly newsletter for all subscribers",
    utm_code="weekly-newsletter-2024"
)

campaign = client.campaigns.create(campaign_data)
print(f"Created campaign: {campaign.id}")
print(f"Name: {campaign.name}")
print(f"Status: {campaign.status}")

Scheduled Campaign¶

from datetime import datetime, timedelta

# Schedule campaign for future delivery
scheduled_time = datetime.now() + timedelta(days=1)

campaign_data = CampaignCreate(
    name="Product Launch Announcement",
    channel_type="email",
    email_channel="channel_123",
    email_template="template_456",
    audience="audience_789",
    scheduled_at=scheduled_time,
    execution_duration_minutes=60,  # Spread sends over 60 minutes
    utm_code="product-launch-2024"
)

campaign = client.campaigns.create(campaign_data)
print(f"Campaign scheduled for: {campaign.scheduled_at}")

Note that this will not put the campaign in scheduled status immediately. You need to call the schedule method after creation to set it to scheduled status.

Campaign with Custom Sender Information¶

# Override default channel sender information
campaign_data = CampaignCreate(
    name="Special Promotion",
    channel_type="email",
    email_channel="channel_123",
    email_template="template_456",
    audience="audience_789",
    from_email="promo@company.com",
    from_name="Sales Team",
    reply_to_email="support@company.com",
    reply_to_name="Customer Support",
    utm_code="special-promo"
)

campaign = client.campaigns.create(campaign_data)

Note that the from_email must be a verified email identity or must belong to a verified domain identity in the account.

Quick Email Campaign Creation¶

Create a complete email campaign with audience and template in one step:

from konigle.models.comm import CampaignCreateEmail

# Create campaign, audience, and template all at once
campaign_data = CampaignCreateEmail(
    campaign_name="Flash Sale Announcement",
    email_channel="my-channel",
    contact_tags=["customer", "active"],
    subject="24-Hour Flash Sale - 50% Off!",
    body_html="<h1>Flash Sale!</h1><p>Get 50% off all products.</p>",
    body_text="Flash Sale! Get 50% off all products.",
    description="Flash sale campaign for active customers",
    utm_code="flash-sale-2024",
    audience_name="Active Customers",
    template_name="Flash Sale Template"
)

campaign = client.campaigns.new_email_campaign(campaign_data)
print(f"Created campaign: {campaign.name}")
print(f"Status: {campaign.status}")

Listing Campaigns¶

from konigle.filters.comm import CampaignFilters

# List all campaigns with pagination
campaigns = client.campaigns.list(page=1, page_size=20)

print(f"Total campaigns: {campaigns.count}")
for campaign in campaigns.payload:
    print(campaign.model_dump_json(indent=2))


# Filter by status
active_filters = CampaignFilters(
    status="running",
    ordering="-created_at"
)

active_campaigns = client.campaigns.list(filters=active_filters)
print(f"Active campaigns: {active_campaigns.count}")

# Search campaigns
search_filters = CampaignFilters(
    q="newsletter"
)

campaigns = client.campaigns.list(filters=search_filters)

# Filter by audience
audience_filters = CampaignFilters(
    audience="premium-customers"
)

premium_campaigns = client.campaigns.list(filters=audience_filters)

Getting Campaign Details¶

# Get campaign with execution details
campaign = client.campaigns.get("campaign_123456789")
print(f"Campaign Details:")
print(campaign.model_dump_json(indent=2))

# Check execution metrics
if campaign.execution:
    exec = campaign.execution
    print(f"\nExecution Status: {exec.status}")
    print(f"Total contacts: {exec.total_contacts}")
    print(f"Sent: {exec.total_sent}")
    print(f"Delivered: {exec.total_delivered}")
    print(f"Bounced: {exec.total_bounced}")
    if exec.total_opened:
        print(f"Opened: {exec.total_opened}")
    if exec.total_clicked:
        print(f"Clicked: {exec.total_clicked}")

Note that open and click tracking is not available as of now

Updating Campaigns¶

Campaigns can only be updated when in draft or scheduled status (except name and description which can be updated anytime):

from konigle.models.comm import CampaignUpdate

# Update campaign information
update_data = CampaignUpdate(
    name="Updated Newsletter Title",
    description="Updated description",
    utm_code="updated-newsletter-2024"
)

updated_campaign = client.campaigns.update(
    "campaign_123456789",
    update_data
)

# Update scheduling (only in draft/scheduled status)
schedule_update = CampaignUpdate(
    scheduled_at=datetime.now() + timedelta(hours=2),
    execution_duration_minutes=120
)

campaign = client.campaigns.update("campaign_123456789", schedule_update)

# Update sender information
sender_update = CampaignUpdate(
    from_email="new-sender@company.com",
    from_name="Marketing Team"
)

campaign = client.campaigns.update("campaign_123456789", sender_update)

Campaign Status Management¶

Starting a Campaign¶

# Start a campaign (works for draft or scheduled campaigns)
campaign = client.campaigns.start("campaign_123456789")
print(f"Campaign started: {campaign.status}")

Pausing a Campaign¶

# Pause a running campaign
campaign = client.campaigns.pause("campaign_123456789")
print(f"Campaign paused: {campaign.status}")

Resuming a Campaign¶

# Resume a paused campaign
campaign = client.campaigns.resume("campaign_123456789")
print(f"Campaign resumed: {campaign.status}")

Cancelling a Campaign¶

# Cancel a campaign (works for draft, scheduled, or running)
campaign = client.campaigns.cancel("campaign_123456789")
print(f"Campaign cancelled: {campaign.status}")

Scheduling a Campaign¶

from datetime import datetime, timedelta

# Schedule a draft campaign for future delivery
scheduled_time = datetime.now() + timedelta(days=2)
campaign = client.campaigns.schedule(
    "campaign_123456789",
    scheduled_time.isoformat()
)
print(f"Campaign scheduled for: {campaign.scheduled_at}")

# Reschedule an already scheduled campaign
new_time = datetime.now() + timedelta(hours=6)
campaign = client.campaigns.schedule(
    "campaign_123456789",
    new_time.isoformat()
)

# Schedule using the scheduled_at that was already set on the campaign
campaign = client.campaigns.schedule("campaign_123456789", None)

Sending Test Emails¶

# Send a test email to verify campaign content and formatting
result = client.campaigns.send_test_email(
    "campaign_123456789",
    "test@example.com"
)
print(f"Test email sent: {result}")

Tracking Campaign Revenue¶

from decimal import Decimal
from konigle.models.comm import CampaignAddLTV

# Record revenue from a campaign purchase
ltv_data = CampaignAddLTV(
    contact_email="customer@example.com",
    value=Decimal("149.99"),
    currency="USD"
)

campaign = client.campaigns.add_ltv("campaign_123456789", ltv_data)
print(f"Campaign LTV: {campaign.ltv} {campaign.ltv_currency}")

# Track aggregate campaign revenue (without specific contact)
aggregate_ltv = CampaignAddLTV(
    value=Decimal("299.99"),
    currency="USD"
)

campaign = client.campaigns.add_ltv("campaign_123456789", aggregate_ltv)

Deleting Campaigns¶

# Delete a campaign
success = client.campaigns.delete("campaign_123456789")
if success:
    print("Campaign deleted successfully")
else:
    print("Failed to delete campaign")

Async Operations¶

import asyncio
import konigle
from konigle.models.comm import CampaignCreate

async def manage_campaigns():
    async with konigle.AsyncClient(api_key="your-api-key") as client:
        # Create campaign
        campaign_data = CampaignCreate(
            name="Async Campaign",
            channel_type="email",
            email_channel="channel_123",
            email_template="template_456",
            audience="audience_789",
            utm_code="async-test"
        )
        campaign = await client.campaigns.create(campaign_data)

        # List campaigns
        from konigle.filters.comm import CampaignFilters
        filters = CampaignFilters(q="async")
        campaigns = await client.campaigns.list(filters=filters)

        # Start campaign
        started = await client.campaigns.start(campaign.id)

        # Get campaign details
        details = await client.campaigns.get(campaign.id)

        # Pause campaign
        paused = await client.campaigns.pause(campaign.id)

        # Schedule campaign
        from datetime import datetime, timedelta
        scheduled_time = datetime.now() + timedelta(hours=2)
        scheduled = await client.campaigns.schedule(
            campaign.id,
            scheduled_time.isoformat()
        )

        # Send test email
        test_result = await client.campaigns.send_test_email(
            campaign.id,
            "test@example.com"
        )

        # Delete campaign
        await client.campaigns.delete(campaign.id)

asyncio.run(manage_campaigns())

CLI Commands¶

The Konigle CLI provides comprehensive campaign management commands.

Campaign Creation¶

# Create a basic campaign
konigle comm campaigns create \
    --name "Weekly Newsletter" \
    --audience "audience_789" \
    --email-channel "channel_123" \
    --email-template "template_456" \
    --description "Weekly newsletter for subscribers" \
    --utm-code "weekly-newsletter"

# Create scheduled campaign
konigle comm campaigns create \
    --name "Product Launch" \
    --audience "audience_789" \
    --email-channel "channel_123" \
    --email-template "template_456" \
    --scheduled-at "2024-12-25T10:00:00" \
    --execution-duration 60 \
    --utm-code "product-launch"

# Create campaign with custom sender
konigle comm campaigns create \
    --name "Special Promotion" \
    --audience "audience_789" \
    --email-channel "channel_123" \
    --email-template "template_456" \
    --from-email "promo@company.com" \
    --from-name "Sales Team" \
    --reply-to-email "support@company.com" \
    --utm-code "special-promo"

Quick Email Campaign Creation¶

# Create complete email campaign in one step
konigle comm campaigns new-email-campaign \
    --name "Flash Sale" \
    --email-channel "my-channel" \
    --tag "customer" \
    --tag "active" \
    --subject "24-Hour Flash Sale!" \
    --body-html "<h1>Flash Sale!</h1><p>50% off all products.</p>" \
    --body-text "Flash Sale! 50% off all products." \
    --utm-code "flash-sale" \
    --audience-name "Active Customers" \
    --template-name "Flash Sale Template"

# Create scheduled email campaign
konigle comm campaigns new-email-campaign \
    --name "Weekly Digest" \
    --email-channel "my-channel" \
    --tag "newsletter" \
    --subject "Weekly Digest" \
    --body-html "<h1>This Week's Updates</h1>" \
    --scheduled-at "2024-12-15T09:00:00" \
    --execution-duration 30

Campaign Listing¶

# List all campaigns
konigle comm campaigns list

# List with pagination
konigle comm campaigns list --page 2 --page-size 5

# Search campaigns
konigle comm campaigns list --search "newsletter"

# Filter by status
konigle comm campaigns list --status "running"

# Filter by audience
konigle comm campaigns list --audience "premium-customers"

Campaign Details¶

# Get campaign details with execution metrics
konigle comm campaigns get campaign_123456789

Campaign Updates¶

# Update campaign name
konigle comm campaigns update campaign_123456789 \
    --name "Updated Campaign Name"

# Update scheduling
konigle comm campaigns update campaign_123456789 \
    --scheduled-at "2024-12-20T15:00:00" \
    --execution-duration 90

# Update sender information
konigle comm campaigns update campaign_123456789 \
    --from-email "new@company.com" \
    --from-name "New Team"

# Update UTM code
konigle comm campaigns update campaign_123456789 \
    --utm-code "updated-campaign-2024"

Campaign Status Management¶

# Start a campaign
konigle comm campaigns start campaign_123456789

# Pause a running campaign
konigle comm campaigns pause campaign_123456789

# Resume a paused campaign
konigle comm campaigns resume campaign_123456789

# Cancel a campaign with confirmation
konigle comm campaigns cancel campaign_123456789

# Cancel without confirmation
konigle comm campaigns cancel campaign_123456789 --yes

Campaign Scheduling¶

# Schedule a campaign for future delivery
konigle comm campaigns schedule campaign_123456789 \
    --scheduled-at "2024-12-20T10:00:00"

# Reschedule an already scheduled campaign
konigle comm campaigns schedule campaign_123456789 \
    --scheduled-at "2024-12-25T15:30:00"

# Schedule using the scheduled_at already set on the campaign
konigle comm campaigns schedule campaign_123456789

Sending Test Emails¶

# Send a test email to verify campaign content
konigle comm campaigns send-test-email campaign_123456789 \
    --email "test@example.com"

# Using short option
konigle comm campaigns send-test-email campaign_123456789 \
    -e "marketing@example.com"

Campaign Revenue Tracking¶

# Add revenue from a specific contact purchase
konigle comm campaigns add-ltv campaign_123456789 \
    --contact-email "customer@example.com" \
    --value 149.99 \
    --currency "USD"

# Add aggregate revenue
konigle comm campaigns add-ltv campaign_123456789 \
    --value 299.99 \
    --currency "USD"

Campaign Deletion¶

# Delete campaign with confirmation
konigle comm campaigns delete campaign_123456789

# Delete campaign without confirmation
konigle comm campaigns delete campaign_123456789 --yes