campaign

`konigle.managers.comm.campaign` ¶

Campaign managers for the Konigle SDK.

This module provides managers for campaign resources, enabling campaign management operations including CRUD and execution control.

`AsyncCampaignManager` ¶

Bases: BaseCampaignManager, BaseAsyncManager

Asynchronous manager for campaign resources.

`add_ltv(campaign_id, data)` `async` ¶

Add LTV (Lifetime Value) to a campaign from a purchase.

Records revenue generated from a contact's purchase and updates both the campaign and contact LTV values.

PARAMETER	DESCRIPTION
`campaign_id`	ID of the campaign to add LTV to TYPE: `str`
`data`	LTV data including contact email, value, and currency TYPE: `CampaignAddLTV`

RETURNS	DESCRIPTION
`Campaign`	Updated campaign with new LTV

RAISES	DESCRIPTION
`APIError`	If the API request fails

Example

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

ltv_data = CampaignAddLTV(
    contact_email="customer@example.com",
    value=Decimal("99.99"),
    currency="USD",
)
campaign = await client.campaigns.add_ltv(
    "campaign_id", ltv_data
)
print(f"Campaign LTV: {campaign.ltv} {campaign.ltv_currency}")

`cancel(campaign_id)` `async` ¶

Cancel a campaign. Works when status is running, draft, or scheduled.

PARAMETER	DESCRIPTION
`campaign_id`	ID of the campaign to cancel TYPE: `str`

RETURNS	DESCRIPTION
`Campaign`	Updated campaign with cancelled status

RAISES	DESCRIPTION
`ValueError`	If campaign cannot be cancelled
`APIError`	If the API request fails

Example

campaign = await client.campaigns.cancel("campaign_id")
print(f"Campaign cancelled: {campaign.status}")

`create(data)` `async` ¶

Create a new campaign.

PARAMETER	DESCRIPTION
`data`	Campaign creation data including all required fields TYPE: `CampaignCreate`

RETURNS	DESCRIPTION
`Campaign`	Created campaign instance with Active Record capabilities

Example

from konigle.models.comm import CampaignCreate

campaign_data = CampaignCreate(
    name="Summer Sale 2024",
    channel_type="email",
    email_channel="channel_id",
    email_template="template_id",
    audience="audience_id",
    description="Promotional campaign for summer sale",
    scheduled_at=None,  # Send immediately
    execution_duration_minutes=60,  # Spread over 1 hour
    utm_code="summer-sale-2024",
)
campaign = await client.campaigns.create(campaign_data)
print(f"Created campaign: {campaign.name}")

`new_email_campaign(data)` `async` ¶

Create a new email campaign with audience and template.

This hybrid API creates a campaign along with its associated audience and email template in a single operation.

PARAMETER	DESCRIPTION
`data`	Email campaign data with all required fields TYPE: `CampaignCreateEmail`

RETURNS	DESCRIPTION
`Campaign`	Created campaign instance

RAISES	DESCRIPTION
`APIError`	If the API request fails

Example

from konigle.models.comm import CampaignCreateEmail

campaign_data = CampaignCreateEmail(
    campaign_name="Summer Sale 2024",
    email_channel="marketing",
    contact_tags=["newsletter", "vip"],
    subject="Summer Sale - 50% Off!",
    body_html="<h1>Limited Time Offer!</h1>",
    body_text="Summer Sale - 50% Off!",
    utm_code="summer-sale-2024",
)
campaign = await client.campaigns.new_email_campaign(
    campaign_data
)
print(f"Created campaign: {campaign.name}")

`pause(campaign_id)` `async` ¶

Pause a running campaign. Only works when status is running.

PARAMETER	DESCRIPTION
`campaign_id`	ID of the campaign to pause TYPE: `str`

RETURNS	DESCRIPTION
`Campaign`	Updated campaign with paused status

RAISES	DESCRIPTION
`ValueError`	If campaign is not in running status
`APIError`	If the API request fails

Example

campaign = await client.campaigns.pause("campaign_id")
print(f"Campaign paused: {campaign.status}")

`resume(campaign_id)` `async` ¶

Resume a paused campaign. Only works when status is paused.

PARAMETER	DESCRIPTION
`campaign_id`	ID of the campaign to resume TYPE: `str`

RETURNS	DESCRIPTION
`Campaign`	Updated campaign with resumed status

RAISES	DESCRIPTION
`ValueError`	If campaign is not in paused status
`APIError`	If the API request fails

Example

campaign = await client.campaigns.resume("campaign_id")
print(f"Campaign resumed: {campaign.status}")

`schedule(campaign_id, scheduled_at)` `async` ¶

Schedule or reschedule a campaign. Args: campaign_id: ID of the campaign to schedule scheduled_at: ISO 8601 datetime string for when to send the campaign. None is allowd if the scheduled_at is already set. Returns: Updated campaign with new scheduled time Raises: ValueError: If campaign is not in draft status APIError: If the API request fails

`send_test_email(campaign_id, email_address)` `async` ¶

Send a test email for the specified email campaign.

PARAMETER	DESCRIPTION
`campaign_id`	ID of the email campaign to send a test for TYPE: `str`
`email_address`	Recipient email address for the test email TYPE: `str`

Returns: dict: API response indicating success or failure Raises: APIError: If the API request fails

`start(campaign_id)` `async` ¶

Start a campaign. Only works when status is draft or scheduled.

PARAMETER	DESCRIPTION
`campaign_id`	ID of the campaign to start TYPE: `str`

RETURNS	DESCRIPTION
`Campaign`	Updated campaign with new status

RAISES	DESCRIPTION
`ValueError`	If campaign is not in draft or scheduled status
`APIError`	If the API request fails

Example

campaign = await client.campaigns.start("campaign_id")
print(f"Campaign started: {campaign.status}")

`BaseCampaignManager` ¶

Base class for campaign managers with shared configuration.

`base_path = '/reachout/api/v1/campaigns'` `class-attribute` `instance-attribute` ¶

The API base path for this resource type.

`filter_class = CampaignFilters` `class-attribute` `instance-attribute` ¶

The filter model class for this resource type.

`resource_class = Campaign` `class-attribute` `instance-attribute` ¶

The resource model class this manager handles.

`resource_update_class = CampaignUpdate` `class-attribute` `instance-attribute` ¶

The model class used for updating resources.

`CampaignManager` ¶

Bases: BaseCampaignManager, BaseSyncManager

Synchronous manager for campaign resources.

`add_ltv(campaign_id, data)` ¶

Add LTV (Lifetime Value) to a campaign from a purchase.

Records revenue generated from a contact's purchase and updates both the campaign and contact LTV values.

PARAMETER	DESCRIPTION
`campaign_id`	ID of the campaign to add LTV to TYPE: `str`
`data`	LTV data including contact email, value, and currency TYPE: `CampaignAddLTV`

RETURNS	DESCRIPTION
`Campaign`	Updated campaign with new LTV

RAISES	DESCRIPTION
`APIError`	If the API request fails

Example

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

ltv_data = CampaignAddLTV(
    contact_email="customer@example.com",
    value=Decimal("99.99"),
    currency="USD",
)
campaign = client.campaigns.add_ltv("campaign_id", ltv_data)
print(f"Campaign LTV: {campaign.ltv} {campaign.ltv_currency}")

`cancel(campaign_id)` ¶

Cancel a campaign. Works when status is running, draft, scheduled or paused.

PARAMETER	DESCRIPTION
`campaign_id`	ID of the campaign to cancel TYPE: `str`

RETURNS	DESCRIPTION
`Campaign`	Updated campaign with cancelled status

RAISES	DESCRIPTION
`ValueError`	If campaign cannot be cancelled
`APIError`	If the API request fails

Example

campaign = client.campaigns.cancel("campaign_id")
print(f"Campaign cancelled: {campaign.status}")

`create(data)` ¶

Create a new campaign.

PARAMETER	DESCRIPTION
`data`	Campaign creation data including all required fields TYPE: `CampaignCreate`

RETURNS	DESCRIPTION
`Campaign`	Created campaign instance with Active Record capabilities

Example

from konigle.models.comm import CampaignCreate

campaign_data = CampaignCreate(
    name="Summer Sale 2024",
    channel_type="email",
    email_channel="channel_id",
    email_template="template_id",
    audience="audience_id",
    description="Promotional campaign for summer sale",
    scheduled_at=None,  # Send immediately
    execution_duration_minutes=60,  # Spread over 1 hour
    utm_code="summer-sale-2024",
)
campaign = client.campaigns.create(campaign_data)
print(f"Created campaign: {campaign.name}")

`new_email_campaign(data)` ¶

Create a new email campaign with audience and template.

This hybrid API creates a campaign along with its associated audience and email template in a single operation.

PARAMETER	DESCRIPTION
`data`	Email campaign data with all required fields TYPE: `CampaignCreateEmail`

RETURNS	DESCRIPTION
`Campaign`	Created campaign instance

RAISES	DESCRIPTION
`APIError`	If the API request fails

Example

from konigle.models.comm import CampaignCreateEmail

campaign_data = CampaignCreateEmail(
    campaign_name="Summer Sale 2024",
    email_channel="marketing",
    contact_tags=["newsletter", "vip"],
    subject="Summer Sale - 50% Off!",
    body_html="<h1>Limited Time Offer!</h1>",
    body_text="Summer Sale - 50% Off!",
    utm_code="summer-sale-2024",
)
campaign = client.campaigns.new_email_campaign(campaign_data)
print(f"Created campaign: {campaign.name}")

`pause(campaign_id)` ¶

Pause a running campaign. Only works when status is running.

PARAMETER	DESCRIPTION
`campaign_id`	ID of the campaign to pause TYPE: `str`

RETURNS	DESCRIPTION
`Campaign`	Updated campaign with paused status

RAISES	DESCRIPTION
`ValueError`	If campaign is not in running status
`APIError`	If the API request fails

Example

campaign = client.campaigns.pause("campaign_id")
print(f"Campaign paused: {campaign.status}")

`resume(campaign_id)` ¶

Resume a paused campaign. Only works when status is paused.

PARAMETER	DESCRIPTION
`campaign_id`	ID of the campaign to resume TYPE: `str`

RETURNS	DESCRIPTION
`Campaign`	Updated campaign with resumed status

RAISES	DESCRIPTION
`ValueError`	If campaign is not in paused status
`APIError`	If the API request fails

Example

campaign = client.campaigns.resume("campaign_id")
print(f"Campaign resumed: {campaign.status}")

`schedule(campaign_id, scheduled_at)` ¶

Schedule or reschedule a campaign.

PARAMETER	DESCRIPTION
`campaign_id`	ID of the campaign to schedule TYPE: `str`
`scheduled_at`	ISO 8601 datetime string for when to send the campaign. None is allowd if the scheduled_at is already set. TYPE: `Optional[str]`

Returns: Updated campaign with new scheduled time Raises: ValueError: If campaign is not in draft status APIError: If the API request fails

`send_test_email(campaign_id, email_address)` ¶

Send a test email for the specified email campaign.

PARAMETER	DESCRIPTION
`campaign_id`	ID of the email campaign to send a test for TYPE: `str`
`email_address`	Recipient email address for the test email TYPE: `str`

Returns: dict: API response indicating success or failure Raises: APIError: If the API request fails

`start(campaign_id)` ¶

Start a campaign. Only works when status is draft or scheduled.

PARAMETER	DESCRIPTION
`campaign_id`	ID of the campaign to start TYPE: `str`

RETURNS	DESCRIPTION
`Campaign`	Updated campaign with new status

RAISES	DESCRIPTION
`ValueError`	If campaign is not in draft or scheduled status
`APIError`	If the API request fails

Example

campaign = client.campaigns.start("campaign_id")
print(f"Campaign started: {campaign.status}")

campaign

konigle.managers.comm.campaign ¶

AsyncCampaignManager ¶

add_ltv(campaign_id, data) async ¶

cancel(campaign_id) async ¶

create(data) async ¶

new_email_campaign(data) async ¶

pause(campaign_id) async ¶

resume(campaign_id) async ¶

schedule(campaign_id, scheduled_at) async ¶

send_test_email(campaign_id, email_address) async ¶

start(campaign_id) async ¶

BaseCampaignManager ¶

base_path = '/reachout/api/v1/campaigns' class-attribute instance-attribute ¶

filter_class = CampaignFilters class-attribute instance-attribute ¶

resource_class = Campaign class-attribute instance-attribute ¶

resource_update_class = CampaignUpdate class-attribute instance-attribute ¶

CampaignManager ¶

add_ltv(campaign_id, data) ¶

cancel(campaign_id) ¶

create(data) ¶

new_email_campaign(data) ¶

pause(campaign_id) ¶

resume(campaign_id) ¶

schedule(campaign_id, scheduled_at) ¶

send_test_email(campaign_id, email_address) ¶

start(campaign_id) ¶

`konigle.managers.comm.campaign` ¶

`AsyncCampaignManager` ¶

`add_ltv(campaign_id, data)` `async` ¶

`cancel(campaign_id)` `async` ¶

`create(data)` `async` ¶

`new_email_campaign(data)` `async` ¶

`pause(campaign_id)` `async` ¶

`resume(campaign_id)` `async` ¶

`schedule(campaign_id, scheduled_at)` `async` ¶

`send_test_email(campaign_id, email_address)` `async` ¶

`start(campaign_id)` `async` ¶

`BaseCampaignManager` ¶

`base_path = '/reachout/api/v1/campaigns'` `class-attribute` `instance-attribute` ¶

`filter_class = CampaignFilters` `class-attribute` `instance-attribute` ¶

`resource_class = Campaign` `class-attribute` `instance-attribute` ¶

`resource_update_class = CampaignUpdate` `class-attribute` `instance-attribute` ¶

`CampaignManager` ¶

`add_ltv(campaign_id, data)` ¶

`cancel(campaign_id)` ¶

`create(data)` ¶

`new_email_campaign(data)` ¶

`pause(campaign_id)` ¶

`resume(campaign_id)` ¶

`schedule(campaign_id, scheduled_at)` ¶

`send_test_email(campaign_id, email_address)` ¶

`start(campaign_id)` ¶