Skip to content

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: AudienceCreate

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: 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}")

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: ContactCreate

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: ContactUpdateLTV

RETURNS DESCRIPTION
Contact

Updated contact with new LTV

RAISES DESCRIPTION
APIError

If the API request fails

Example
from konigle.models.comm import ContactUpdateLTV
from decimal import Decimal

ltv_data = ContactUpdateLTV(
    email="customer@example.com",
    value=Decimal("149.99"),
    currency="USD",
)
contact = await client.contacts.update_ltv(ltv_data)
print(f"Contact LTV: {contact.ltv} {contact.ltv_currency}")

AsyncEmailAccountManager

Bases: BaseEmailAccountManager, BaseAsyncManager

Asynchronous manager for email account resources.

check_status() async

Check the status of the email account to ensure email sending is operational.

RETURNS DESCRIPTION
Dict[str, Any]

A dictionary containing the status information of the email account.

Example:

status = await client.email_accounts.check_status()
print(f"Email account status: {status}")

create(data) async

Create a new email account.

PARAMETER DESCRIPTION
data

Email account creation data including all required fields

TYPE: EmailAccountCreate

RETURNS DESCRIPTION
EmailAccount

Created email account instance with Active Record capabilities

Example
account_data = EmailAccountCreate(
    name="Marketing Account",
    default_from_email="noreply@example.com",
    default_from_name="Example Team",
)
account = await client.email_accounts.create(account_data)
print(f"Created account: {account.name}")

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: EmailChannelCreate

RETURNS DESCRIPTION
EmailChannel

Created email channel instance with Active Record capabilities

Example
channel_data = EmailChannelCreate(
    code="transactional",
    channel_type=EmailChannelType.TRANSACTIONAL,
)
channel = await client.email_channels.create(channel_data)
print(f"Created channel: {channel.code}")

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: str

enable

True to enable engagement tracking, False to disable

TYPE: bool

RETURNS DESCRIPTION
EmailChannel

Updated channel instance with engagement tracking setting

Example
channel = await client.email_channels.set_engagement_tracking(
    "ch_123", True
)
print(f"Engagement tracking: {channel.enable_engagement_tracking}")

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: str

Returns: Updated identity instance with verification status Example:

identity = await client.email_identities.check_verification_status("eid_123")
print(f"Verification status: {identity.verified}")

create(data) async

Create a new email identity.

PARAMETER DESCRIPTION
data

Email identity creation data including all required fields

TYPE: EmailIdentityCreate

RETURNS DESCRIPTION
EmailIdentity

Created email identity instance with Active Record capabilities

Example
identity_data = EmailIdentityCreate(
    identity_value="example.com",
)
identity = await client.email_identities.create(identity_data)
print(f"Created identity: {identity.identity_value}")

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: Union[Email, dict]

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: EmailTemplateCreate

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: AudienceCreate

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: Union[CreateModel, Dict[str, Any]]

RETURNS DESCRIPTION
BaseResource

Created resource instance

delete(id_) async

Delete a resource.

PARAMETER DESCRIPTION
id_

Unique identifier for the resource

TYPE: str

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: str

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: int DEFAULT: 100

filters

Filter object for type-safe filtering

TYPE: Optional[BaseFilters] DEFAULT: None

**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: int DEFAULT: 1

page_size

Number of items per page

TYPE: int DEFAULT: 20

filters

Filter object for type-safe filtering

TYPE: Optional[BaseFilters] DEFAULT: None

**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: str

data

Updated resource data model instance

TYPE: Union[UpdateModel, Dict[str, Any]]

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: Union[CreateModel, Dict[str, Any]]

RETURNS DESCRIPTION
BaseResource

Created resource instance

delete(id_)

Delete a resource.

PARAMETER DESCRIPTION
id_

Unique identifier for the resource

TYPE: str

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: str

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: int DEFAULT: 100

filters

Filter object for type-safe filtering

TYPE: Optional[BaseFilters] DEFAULT: None

**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: int DEFAULT: 1

page_size

Number of items per page

TYPE: int DEFAULT: 20

filters

Filter object for type-safe filtering

TYPE: Optional[BaseFilters] DEFAULT: None

**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: str

data

Updated resource data model instance or dict

TYPE: Union[UpdateModel, Dict[str, Any]]

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

Bases: BaseCampaign, CreateModel

Model for creating a new campaign.

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: 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}")

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: ContactCreate

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: ContactUpdateLTV

RETURNS DESCRIPTION
Contact

Updated contact with new LTV

RAISES DESCRIPTION
APIError

If the API request fails

Example
from konigle.models.comm import ContactUpdateLTV
from decimal import Decimal

ltv_data = ContactUpdateLTV(
    email="customer@example.com",
    value=Decimal("149.99"),
    currency="USD",
)
contact = client.contacts.update_ltv(ltv_data)
print(f"Contact LTV: {contact.ltv} {contact.ltv_currency}")

EmailAccountManager

Bases: BaseEmailAccountManager, BaseSyncManager

Synchronous manager for email account resources.

check_status()

Check the status of the email account to ensure email sending is operational.

RETURNS DESCRIPTION
Dict[str, Any]

A dictionary containing the status information of the email account.

Example
status = client.email_accounts.check_status()
print(f"Email account status: {status}")

create(data)

Create a new email account.

PARAMETER DESCRIPTION
data

Email account creation data including all required fields

TYPE: EmailAccountCreate

RETURNS DESCRIPTION
EmailAccount

Created email account instance with Active Record capabilities

Example
account_data = EmailAccountCreate(
    name="Marketing Account",
    default_from_email="noreply@example.com",
    default_from_name="Example Team",
)
account = client.email_accounts.create(account_data)
print(f"Created account: {account.name}")

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: EmailChannelCreate

RETURNS DESCRIPTION
EmailChannel

Created email channel instance with Active Record capabilities

Example
channel_data = EmailChannelCreate(
    code="transactional",
    channel_type=EmailChannelType.TRANSACTIONAL,
)
channel = client.email_channels.create(channel_data)
print(f"Created channel: {channel.code}")

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: str

enable

True to enable engagement tracking, False to disable

TYPE: bool

RETURNS DESCRIPTION
EmailChannel

Updated channel instance with engagement tracking setting

Example
channel = client.email_channels.set_engagement_tracking(
    "ch_123", True
)
print(f"Engagement tracking: {channel.enable_engagement_tracking}")

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: str

RETURNS DESCRIPTION
EmailIdentity

Updated identity instance with verification status

Example
identity = client.email_identities.check_verification_status("eid_123")
print(f"Verification status: {identity.verified}")

create(data)

Create a new email identity.

PARAMETER DESCRIPTION
data

Email identity creation data including all required fields

TYPE: EmailIdentityCreate

RETURNS DESCRIPTION
EmailIdentity

Created email identity instance with Active Record capabilities

Example
identity_data = EmailIdentityCreate(
    identity_value="example.com",
)
identity = client.email_identities.create(identity_data)
print(f"Created identity: {identity.identity_value}")

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: str

mail_from_domain

Custom MAIL FROM domain to be set

TYPE: str

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: Union[Email, dict]

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: EmailTemplateCreate

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.