comm
konigle.managers.comm
¶
Communication managers for the Konigle SDK.
This module provides manager classes for email, SMS, WhatsApp, and other communication channels supported by the Konigle platform.
AsyncAudienceManager
¶
Bases: BaseAudienceManager, BaseAsyncManager
Asynchronous manager for audience resources.
create(data)
async
¶
Create a new audience.
| PARAMETER | DESCRIPTION |
|---|---|
data
|
Audience creation data including all required fields
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Audience
|
Created audience instance with Active Record capabilities |
Example
from konigle.models.comm import AudienceCreate
audience_data = AudienceCreate(
name="Newsletter Subscribers",
code="newsletter-subscribers",
description="All contacts who subscribed to newsletter",
tags=["newsletter", "engaged"],
)
audience = await client.audiences.create(audience_data)
print(f"Created audience: {audience.name}")
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:
|
data
|
LTV data including contact email, value, and currency
TYPE:
|
| 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:
|
| RETURNS | DESCRIPTION |
|---|---|
Campaign
|
Updated campaign with cancelled status |
| RAISES | DESCRIPTION |
|---|---|
ValueError
|
If campaign cannot be cancelled |
APIError
|
If the API request fails |
create(data)
async
¶
Create a new campaign.
| PARAMETER | DESCRIPTION |
|---|---|
data
|
Campaign creation data including all required fields
TYPE:
|
| 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:
|
| 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
¶
resume(campaign_id)
async
¶
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:
|
email_address
|
Recipient email address for the test email
TYPE:
|
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:
|
| 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 |
AsyncContactManager
¶
Bases: BaseContactManager, BaseAsyncManager
Asynchronous manager for contact resources.
create(data)
async
¶
Create a new contact.
| PARAMETER | DESCRIPTION |
|---|---|
data
|
Contact creation data including all required fields
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Contact
|
Created contact instance with Active Record capabilities |
Example
from konigle.models.comm import (
ContactCreate,
MarketingConsent,
)
contact_data = ContactCreate(
email="john@example.com",
first_name="John",
last_name="Doe",
phone="+1234567890",
source="lead",
tags=["newsletter", "webinar"],
marketing_consent=MarketingConsent(
email=True,
sms=False,
whatsapp=True,
),
)
contact = await client.contacts.create(contact_data)
print(f"Created contact: {contact.email}")
update_ltv(data)
async
¶
Update contact LTV (Lifetime Value) from a purchase.
Adds revenue to a contact's lifetime value based on email address. This is a non-detail interface that doesn't require a contact ID.
| PARAMETER | DESCRIPTION |
|---|---|
data
|
LTV update data including email, value, and currency
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Contact
|
Updated contact with new LTV |
| RAISES | DESCRIPTION |
|---|---|
APIError
|
If the API request fails |
AsyncEmailAccountManager
¶
Bases: BaseEmailAccountManager, BaseAsyncManager
Asynchronous manager for email account resources.
check_status()
async
¶
create(data)
async
¶
Create a new email account.
| PARAMETER | DESCRIPTION |
|---|---|
data
|
Email account creation data including all required fields
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
EmailAccount
|
Created email account instance with Active Record capabilities |
setup(data)
async
¶
Set up a new email account along with its default channel and identity. Args: data: Email account setup data including all required fields Returns: A dictionary containing the created email account, its default channel, and identity. Example:
setup_data = EmailAccountSetup(
name="Marketing Account",
default_from_email=""
default_reply_to_email=""
identity_value="example.com",
)
setup_result = await client.email_accounts.setup(setup_data)
account = setup_result["account"]
channel = setup_result["channels"][0]
identity = setup_result["identity"]
AsyncEmailChannelManager
¶
Bases: BaseEmailChannelManager, BaseAsyncManager
Asynchronous manager for email channel resources.
create(data)
async
¶
Create a new email channel.
| PARAMETER | DESCRIPTION |
|---|---|
data
|
Email channel creation data including all required fields
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
EmailChannel
|
Created email channel instance with Active Record capabilities |
set_engagement_tracking(channel_id, enable)
async
¶
Enable or disable engagement tracking for the channel.
| PARAMETER | DESCRIPTION |
|---|---|
channel_id
|
ID of the channel to update
TYPE:
|
enable
|
True to enable engagement tracking, False to disable
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
EmailChannel
|
Updated channel instance with engagement tracking setting |
AsyncEmailIdentityManager
¶
Bases: BaseEmailIdentityManager, BaseAsyncManager
Asynchronous manager for email identity resources.
check_verification_status(identity_id)
async
¶
Check verification status for an email identity.
| PARAMETER | DESCRIPTION |
|---|---|
identity_id
|
ID of the identity to verify
TYPE:
|
Returns: Updated identity instance with verification status Example:
create(data)
async
¶
Create a new email identity.
| PARAMETER | DESCRIPTION |
|---|---|
data
|
Email identity creation data including all required fields
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
EmailIdentity
|
Created email identity instance with Active Record capabilities |
setup_custom_mail_from(identity_id, mail_from_domain)
async
¶
Setup custom MAIL FROM domain for an email identity. Args: identity_id: ID of the identity to setup custom MAIL FROM mail_from_domain: Custom MAIL FROM domain to be set Returns: Updated identity instance with custom MAIL FROM domain that is pending verification. The returned instance includes the dns records that need to be added for verification.
AsyncEmailManager
¶
Bases: BaseEmailManager
Asynchronous manager for email operations.
Provides async functionality for sending emails and other email-related actions that don't involve resource management.
send(data)
async
¶
Send an email through the Konigle email service.
| PARAMETER | DESCRIPTION |
|---|---|
data
|
Email data model instance or dict containing email details
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
EmailResponse
|
SendEmailResponse with success message and optional IDs |
Example
from konigle.models.comm import Email
email_data = Email(
to_email=["recipient@example.com"],
subject="Welcome to Konigle",
body_html="<h1>Welcome!</h1><p>Thanks for signing up.</p>",
body_text="Welcome! Thanks for signing up.",
channel="welcome-emails",
save_as_template=False
)
response = await client.emails.send(email_data)
print(f"Email sent: {response.message}")
AsyncEmailTemplateManager
¶
Bases: BaseEmailTemplateManager, BaseAsyncManager
Asynchronous manager for email template resources.
create(data)
async
¶
Create a new email template.
| PARAMETER | DESCRIPTION |
|---|---|
data
|
Email template creation data including all required fields
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
EmailTemplate
|
Created email template instance with Active Record capabilities |
Example
template_data = EmailTemplateCreate(
name="Welcome Email",
code="welcome-email",
subject="Welcome to {{company_name}}!",
body_html="<h1>Welcome!</h1><p>Thanks for joining.</p>",
body_text="Welcome! Thanks for joining.",
tags=["welcome", "onboarding"]
)
template = await client.email_templates.create(template_data)
print(f"Created template: {template.name}")
AudienceManager
¶
Bases: BaseAudienceManager, BaseSyncManager
Synchronous manager for audience resources.
create(data)
¶
Create a new audience.
| PARAMETER | DESCRIPTION |
|---|---|
data
|
Audience creation data including all required fields
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Audience
|
Created audience instance with Active Record capabilities |
Example
from konigle.models.comm import AudienceCreate
audience_data = AudienceCreate(
name="Newsletter Subscribers",
code="newsletter-subscribers",
description="All contacts who subscribed to newsletter",
tags=["newsletter", "engaged"],
)
audience = client.audiences.create(audience_data)
print(f"Created audience: {audience.name}")
BaseAsyncManager
¶
Bases: BaseManager
Asynchronous manager providing CRUD operations.
All async resource managers inherit from this class to get standard functionality for listing, creating, updating, and deleting resources.
__init_subclass__(**kwargs)
¶
Ensure subclasses define required class attributes.
create(data)
async
¶
Create a new resource.
| PARAMETER | DESCRIPTION |
|---|---|
data
|
Resource data model instance or dict
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
BaseResource
|
Created resource instance |
delete(id_)
async
¶
Delete a resource.
| PARAMETER | DESCRIPTION |
|---|---|
id_
|
Unique identifier for the resource
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
bool
|
True if deletion was successful |
get(id_)
async
¶
Get a specific resource by ID.
| PARAMETER | DESCRIPTION |
|---|---|
id_
|
Unique identifier for the resource
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
BaseResource
|
Resource instance with full detail data |
iter_all(page_size=100, filters=None, **filter_kwargs)
async
¶
Memory-efficient iteration over all matching resources.
| PARAMETER | DESCRIPTION |
|---|---|
page_size
|
Number of items to fetch per page
TYPE:
|
filters
|
Filter object for type-safe filtering
TYPE:
|
**filter_kwargs
|
Filter arguments as keyword arguments
DEFAULT:
|
| YIELDS | DESCRIPTION |
|---|---|
|
Resource instances one by one |
list(page=1, page_size=20, filters=None, **filter_kwargs)
async
¶
List resources with pagination and filtering.
| PARAMETER | DESCRIPTION |
|---|---|
page
|
Page number (1-based)
TYPE:
|
page_size
|
Number of items per page
TYPE:
|
filters
|
Filter object for type-safe filtering
TYPE:
|
**filter_kwargs
|
Filter arguments as keyword arguments
DEFAULT:
|
| RETURNS | DESCRIPTION |
|---|---|
PaginatedResponse[BaseResource]
|
Paginated response containing list of resources |
update(id_, data)
async
¶
Update an existing resource.
| PARAMETER | DESCRIPTION |
|---|---|
id_
|
Unique identifier for the resource
TYPE:
|
data
|
Updated resource data model instance
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
BaseResource
|
Updated resource instance |
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.
BaseSyncManager
¶
Bases: BaseManager
Synchronous manager providing CRUD operations.
All sync resource managers inherit from this class to get standard functionality for listing, creating, updating, and deleting resources.
__init_subclass__(**kwargs)
¶
Ensure subclasses define required class attributes.
create(data)
¶
Create a new resource.
| PARAMETER | DESCRIPTION |
|---|---|
data
|
Resource data model instance or dict
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
BaseResource
|
Created resource instance |
delete(id_)
¶
Delete a resource.
| PARAMETER | DESCRIPTION |
|---|---|
id_
|
Unique identifier for the resource
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
bool
|
True if deletion was successful |
get(id_)
¶
Get a specific resource by ID.
| PARAMETER | DESCRIPTION |
|---|---|
id_
|
Unique identifier for the resource
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
BaseResource
|
Resource instance with full detail data |
iter_all(page_size=100, filters=None, **filter_kwargs)
¶
Memory-efficient iteration over all matching resources.
| PARAMETER | DESCRIPTION |
|---|---|
page_size
|
Number of items to fetch per page
TYPE:
|
filters
|
Filter object for type-safe filtering
TYPE:
|
**filter_kwargs
|
Filter arguments as keyword arguments
DEFAULT:
|
| YIELDS | DESCRIPTION |
|---|---|
|
Resource instances one by one |
list(page=1, page_size=20, filters=None, **filter_kwargs)
¶
List resources with pagination and filtering.
| PARAMETER | DESCRIPTION |
|---|---|
page
|
Page number (1-based)
TYPE:
|
page_size
|
Number of items per page
TYPE:
|
filters
|
Filter object for type-safe filtering
TYPE:
|
**filter_kwargs
|
Filter arguments as keyword arguments
DEFAULT:
|
| RETURNS | DESCRIPTION |
|---|---|
PaginatedResponse[BaseResource]
|
Paginated response containing list of resources |
update(id_, data)
¶
Update an existing resource.
| PARAMETER | DESCRIPTION |
|---|---|
id_
|
Unique identifier for the resource
TYPE:
|
data
|
Updated resource data model instance or dict
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
BaseResource
|
Updated resource instance |
Campaign
¶
Bases: BaseCampaign, TimestampedResource
Campaign configuration for sending messages across multiple channels.
Currently supports email campaigns. Designed to support additional channels (WhatsApp, SMS, etc.) in the future.
execution = Field(default=None, title='Execution Details', description='Execution tracking details (available in detail view).')
class-attribute
instance-attribute
¶
Execution tracking details (available in detail view).
ltv = Field(default=None, title='Lifetime Value', description='Lifetime value (revenue) generated by this campaign.', max_digits=12, decimal_places=2)
class-attribute
instance-attribute
¶
Lifetime value (revenue) generated by this campaign.
ltv_currency = Field(default='USD', title='LTV Currency', description='Currency code for lifetime value (ISO 4217).', max_length=3)
class-attribute
instance-attribute
¶
Currency code for lifetime value (ISO 4217).
status = Field(default='draft', title='Status', description='Current status of the campaign.')
class-attribute
instance-attribute
¶
Current status of the campaign.
CampaignAddLTV
¶
Bases: BaseModel
Model for adding LTV to a campaign.
Records revenue generated from a specific contact purchase and updates both campaign and contact LTV.
contact_email = Field(default=None, title='Contact Email', description='Email address of the contact who made the purchase. Optional if tracking aggregate campaign revenue.')
class-attribute
instance-attribute
¶
Email address of the contact who made the purchase.
currency = Field(title='Currency', description='Currency code (ISO 4217) for the LTV value.', max_length=3)
class-attribute
instance-attribute
¶
Currency code (ISO 4217) for the LTV value.
value = Field(title='LTV Value', description='Value to add to campaign and contact LTV (positive).', gt=0, max_digits=19, decimal_places=2)
class-attribute
instance-attribute
¶
Value to add to campaign and contact LTV (must be positive).
CampaignCreate
¶
CampaignCreateEmail
¶
Bases: BaseModel
Model for creating an email campaign with audience and template.
This is a hybrid API that creates a campaign along with its associated audience and email template in a single operation.
audience_name = Field(default='', title='Audience Name', description='Name for audience (defaults to campaign_name).', max_length=255)
class-attribute
instance-attribute
¶
Name for audience (defaults to campaign_name).
body_html = Field(title='HTML Body', description='HTML body content.')
class-attribute
instance-attribute
¶
HTML body content.
body_text = Field(default='', title='Text Body', description='Plain text body content.')
class-attribute
instance-attribute
¶
Plain text body content.
campaign_name = Field(title='Campaign Name', description='Name of the campaign.', max_length=255)
class-attribute
instance-attribute
¶
Name of the campaign.
contact_tags = Field(title='Contact Tags', description='List of tags to filter contacts for audience.')
class-attribute
instance-attribute
¶
List of tags to filter contacts for audience.
description = Field(default='', title='Description', description='Campaign description.')
class-attribute
instance-attribute
¶
Campaign description.
email_channel = Field(title='Email Channel Code', description='Code of email channel to use.', max_length=50)
class-attribute
instance-attribute
¶
Code of email channel to use.
execution_duration_minutes = Field(default=None, title='Execution Duration (Minutes)', description='Duration in minutes to spread campaign sends.', ge=1, le=1440)
class-attribute
instance-attribute
¶
Duration in minutes to spread campaign sends.
from_email = Field(default='', title='From Email', description='Override from email.')
class-attribute
instance-attribute
¶
Override from email.
from_name = Field(default='', title='From Name', description='Override from name. Required if account does not have default.', max_length=255)
class-attribute
instance-attribute
¶
Override from name. Required if account does not have default.
reply_to_email = Field(default='', title='Reply-To Email', description='Override reply-to email. Required if account does not have default.')
class-attribute
instance-attribute
¶
Override reply-to email. Required if account does not have default.
reply_to_name = Field(default='', title='Reply-To Name', description='Override reply-to name. Required if account does not have default.', max_length=255)
class-attribute
instance-attribute
¶
Override reply-to name. Required if account does not have default.
scheduled_at = Field(default=None, title='Scheduled At', description='When to send the campaign.')
class-attribute
instance-attribute
¶
When to send the campaign.
subject = Field(title='Email Subject', description='Email subject line.', max_length=255)
class-attribute
instance-attribute
¶
Email subject line.
template_name = Field(default='', title='Template Name', description='Name for email template (defaults to campaign_name).', max_length=255)
class-attribute
instance-attribute
¶
Name for email template (defaults to campaign_name).
utm_code = Field(default='', title='UTM Code', description='UTM campaign code for tracking.', max_length=100)
class-attribute
instance-attribute
¶
UTM campaign code for tracking.
CampaignFilters
¶
Bases: BaseFilters
Type-safe filters for campaign queries.
audience = Field(default=None, title='Audience code', description='Filter by audience code.')
class-attribute
instance-attribute
¶
Filter by audience code.
channel_type = Field(default=None, title='Channel Type', description='Filter by channel type.')
class-attribute
instance-attribute
¶
Filter by channel type.
q = Field(default=None, title='Search Query', description='Search in campaign name and description.')
class-attribute
instance-attribute
¶
Search in campaign name and description.
status = Field(default=None, title='Status', description='Filter by campaign status.')
class-attribute
instance-attribute
¶
Filter by campaign status.
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:
|
data
|
LTV data including contact email, value, and currency
TYPE:
|
| 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:
|
| RETURNS | DESCRIPTION |
|---|---|
Campaign
|
Updated campaign with cancelled status |
| RAISES | DESCRIPTION |
|---|---|
ValueError
|
If campaign cannot be cancelled |
APIError
|
If the API request fails |
create(data)
¶
Create a new campaign.
| PARAMETER | DESCRIPTION |
|---|---|
data
|
Campaign creation data including all required fields
TYPE:
|
| 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:
|
| 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)
¶
resume(campaign_id)
¶
schedule(campaign_id, scheduled_at)
¶
Schedule or reschedule a campaign.
| PARAMETER | DESCRIPTION |
|---|---|
campaign_id
|
ID of the campaign to schedule
TYPE:
|
scheduled_at
|
ISO 8601 datetime string for when to send the campaign. None is allowd if the scheduled_at is already set.
TYPE:
|
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:
|
email_address
|
Recipient email address for the test email
TYPE:
|
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:
|
| 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 |
CampaignUpdate
¶
Bases: UpdateModel
Model for updating an existing campaign.
Note: ltv and ltv_currency are managed automatically by the system based on campaign performance and cannot be updated directly. Status transitions are handled by separate action endpoints.
Campaign can only be updated when in draft and scheduled states. This does not hold for name and description which can be updated at any time.
audience = Field(default=None, title='Audience ID', description='Target audience segment for this campaign.')
class-attribute
instance-attribute
¶
Target audience segment for this campaign.
description = Field(default=None, title='Description', description='Description of the campaign purpose.')
class-attribute
instance-attribute
¶
Description of the campaign purpose.
email_channel = Field(default=None, title='Email Channel ID', description='Channel to use for sending emails.')
class-attribute
instance-attribute
¶
Channel to use for sending emails.
email_template = Field(default=None, title='Email Template ID', description='Email template to use.')
class-attribute
instance-attribute
¶
Email template to use.
execution_duration_minutes = Field(default=None, title='Execution Duration (Minutes)', description='Duration in minutes over which to spread campaign sends (max 1440).', ge=1, le=1440)
class-attribute
instance-attribute
¶
Duration in minutes over which to spread campaign sends.
from_email = Field(default=None, title='From Email', description='Override from email.')
class-attribute
instance-attribute
¶
Override from email.
from_name = Field(default=None, title='From Name', description='Override from name.', max_length=255)
class-attribute
instance-attribute
¶
Override from name.
name = Field(default=None, title='Campaign Name', description='Name of the campaign.', max_length=255)
class-attribute
instance-attribute
¶
Name of the campaign.
reply_to_email = Field(default=None, title='Reply-To Email', description='Override reply-to email.')
class-attribute
instance-attribute
¶
Override reply-to email.
reply_to_name = Field(default=None, title='Reply-To Name', description='Override reply-to name.', max_length=255)
class-attribute
instance-attribute
¶
Override reply-to name.
scheduled_at = Field(default=None, title='Scheduled At', description='When to send the campaign.')
class-attribute
instance-attribute
¶
When to send the campaign.
utm_code = Field(default=None, title='UTM Code', description='UTM campaign code for tracking.', max_length=100)
class-attribute
instance-attribute
¶
UTM campaign code for tracking.
ContactManager
¶
Bases: BaseContactManager, BaseSyncManager
Synchronous manager for contact resources.
create(data)
¶
Create a new contact.
| PARAMETER | DESCRIPTION |
|---|---|
data
|
Contact creation data including all required fields
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Contact
|
Created contact instance with Active Record capabilities |
Example
from konigle.models.comm import (
ContactCreate,
MarketingConsent,
)
contact_data = ContactCreate(
email="john@example.com",
first_name="John",
last_name="Doe",
phone="+1234567890",
source="lead",
tags=["newsletter", "webinar"],
marketing_consent=MarketingConsent(
email=True,
sms=False,
whatsapp=True,
),
)
contact = client.contacts.create(contact_data)
print(f"Created contact: {contact.email}")
update_ltv(data)
¶
Update contact LTV (Lifetime Value) from a purchase.
Adds revenue to a contact's lifetime value based on email address. This is a non-detail interface that doesn't require a contact ID.
| PARAMETER | DESCRIPTION |
|---|---|
data
|
LTV update data including email, value, and currency
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Contact
|
Updated contact with new LTV |
| RAISES | DESCRIPTION |
|---|---|
APIError
|
If the API request fails |
EmailAccountManager
¶
Bases: BaseEmailAccountManager, BaseSyncManager
Synchronous manager for email account resources.
check_status()
¶
create(data)
¶
Create a new email account.
| PARAMETER | DESCRIPTION |
|---|---|
data
|
Email account creation data including all required fields
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
EmailAccount
|
Created email account instance with Active Record capabilities |
setup(data)
¶
Set up a new email account along with its default channel and identity. Args: data: Email account setup data including all required fields Returns: A dictionary containing the created email account, its default channel, and identity. Example:
setup_data = EmailAccountSetup(
name="Marketing Account",
default_from_email=""
default_reply_to_email=""
identity_value="example.com",
)
setup_result = client.email_accounts.setup(setup_data)
account = setup_result["account"]
channel = setup_result["channels"][0]
identity = setup_result["identity"]
EmailChannelManager
¶
Bases: BaseEmailChannelManager, BaseSyncManager
Synchronous manager for email channel resources.
create(data)
¶
Create a new email channel.
| PARAMETER | DESCRIPTION |
|---|---|
data
|
Email channel creation data including all required fields
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
EmailChannel
|
Created email channel instance with Active Record capabilities |
set_engagement_tracking(channel_id, enable)
¶
Enable or disable engagement tracking for the channel.
| PARAMETER | DESCRIPTION |
|---|---|
channel_id
|
ID of the channel to update
TYPE:
|
enable
|
True to enable engagement tracking, False to disable
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
EmailChannel
|
Updated channel instance with engagement tracking setting |
EmailIdentityManager
¶
Bases: BaseEmailIdentityManager, BaseSyncManager
Synchronous manager for email identity resources.
check_verification_status(identity_id)
¶
Check verification status for an email identity.
| PARAMETER | DESCRIPTION |
|---|---|
identity_id
|
ID of the identity to verify
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
EmailIdentity
|
Updated identity instance with verification status |
create(data)
¶
Create a new email identity.
| PARAMETER | DESCRIPTION |
|---|---|
data
|
Email identity creation data including all required fields
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
EmailIdentity
|
Created email identity instance with Active Record capabilities |
setup_custom_mail_from(identity_id, mail_from_domain)
¶
Setup custom MAIL FROM domain for an email identity.
| PARAMETER | DESCRIPTION |
|---|---|
identity_id
|
ID of the identity to setup custom MAIL FROM
TYPE:
|
mail_from_domain
|
Custom MAIL FROM domain to be set
TYPE:
|
Returns: Updated identity instance with custom MAIL FROM domain that is pending verification. The returned instance includes the dns records that need to be added for verification.
EmailManager
¶
Bases: BaseEmailManager
Synchronous manager for email operations.
Provides functionality for sending emails and other email-related actions that don't involve resource management.
send(data)
¶
Send an email through the Konigle email service.
| PARAMETER | DESCRIPTION |
|---|---|
data
|
Email data model instance or dict containing email details
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
EmailResponse
|
SendEmailResponse with success message and optional IDs |
Example
from konigle.models.comm import Email
email_data = Email(
to_email=["recipient@example.com"],
subject="Welcome to Konigle",
body_html="<h1>Welcome!</h1><p>Thanks for signing up.</p>",
body_text="Welcome! Thanks for signing up.",
channel="welcome-emails",
save_as_template=False
)
response = client.emails.send(email_data)
print(f"Email sent: {response.message}")
EmailTemplateManager
¶
Bases: BaseEmailTemplateManager, BaseSyncManager
Synchronous manager for email template resources.
create(data)
¶
Create a new email template.
| PARAMETER | DESCRIPTION |
|---|---|
data
|
Email template creation data including all required fields
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
EmailTemplate
|
Created email template instance with Active Record capabilities |
Example
template_data = EmailTemplateCreate(
name="Welcome Email",
code="welcome-email",
subject="Welcome to {{company_name}}!",
body_html="<h1>Welcome!</h1><p>Thanks for joining.</p>",
body_text="Welcome! Thanks for joining.",
tags=["welcome", "onboarding"]
)
template = client.email_templates.create(template_data)
print(f"Created template: {template.name}")
model_to_dict(model)
¶
Convert a Pydantic model to a dict, excluding None and unset fields and handling file fields appropriately for multipart requests.