managers

`konigle.managers` ¶

Resource managers for the Konigle SDK.

This module exports all manager classes organized by resource category. Managers handle CRUD operations and business logic for API resources.

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

`AsyncConnectionManager` ¶

Bases: BaseConnectionManager, BaseAsyncManager

Async manager for connection resources.

`create(*args, **kwargs)` `async` ¶

Creation not supported via SDK.

`delete(*args, **kwargs)` `async` ¶

Deletion not supported via SDK.

`get(id_)` `async` ¶

Get a specific connection by ID.

`get_credentials(provider)` `async` ¶

Get connection credentials by ID.

Returns the raw credentials dictionary for the connection, which may include sensitive information like tokens.

PARAMETER	DESCRIPTION
`provider`	Connection provider code TYPE: `str`

Returns: Dictionary containing connection credentials

`update(*args, **kwargs)` `async` ¶

Update not supported via SDK.

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

`AsyncDocumentManager` ¶

Bases: BaseDocumentManager, BaseAsyncManager

Async manager for document assets.

`create(data)` `async` ¶

Create a new document asset.

`search(query, page=1, page_size=10)` `async` ¶

Search for documents by a text query.

PARAMETER	DESCRIPTION
`query`	The search query string. TYPE: `str`
`page`	The page number for pagination. TYPE: `int` DEFAULT: `1`
`page_size`	The number of items per page. TYPE: `int` DEFAULT: `10`

Returns: List of document assets matching the query.

`update(id_, data)` `async` ¶

Update an existing document asset.

`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.

`AsyncFormManager` ¶

Bases: BaseFormManager, BaseAsyncManager

Async manager for form resources.

`create(data)` `async` ¶

Create a new form.

PARAMETER	DESCRIPTION
`data`	Form creation data TYPE: `Union[FormCreate, Dict[str, str]]`

RETURNS	DESCRIPTION
`Form`	Created form instance

`delete(*args, **kwargs)` `async` ¶

Deletion not supported.

`get(slug)` `async` ¶

Get a specific form by slug

`list(page=1, page_size=20)` `async` ¶

List forms with pagination.

PARAMETER	DESCRIPTION
`page`	Page number (1-based) TYPE: `int` DEFAULT: `1`
`page_size`	Number of items per page TYPE: `int` DEFAULT: `20`
`**filter_kwargs`	Filter arguments as keyword arguments

RETURNS	DESCRIPTION
`PaginatedResponse[Form]`	Paginated response containing list of forms

`list_submissions(slug, page=1, page_size=20)` `async` ¶

List submissions for a specific form.

PARAMETER	DESCRIPTION
`slug`	Form slug TYPE: `str`
`page`	Page number (1-based) TYPE: `int` DEFAULT: `1`
`page_size`	Number of items per page TYPE: `int` DEFAULT: `20`

RETURNS	DESCRIPTION
`PaginatedResponse[FormSubmission]`	Paginated response containing list of form submissions

`update(*args, **kwargs)` `async` ¶

Update not supported.

`AsyncImageManager` ¶

Bases: BaseImageManager, BaseAsyncManager

Async manager for image assets.

`create(data)` `async` ¶

Create a new image asset.

`generate(data)` `async` ¶

Generate an image from a text prompt.

PARAMETER	DESCRIPTION
`data`	Image generation parameters including prompt and options TYPE: `ImageGenerate`

RETURNS	DESCRIPTION
`Image`	Generated image asset

`search(query, page=1, page_size=10)` `async` ¶

Search for images by a text query.

PARAMETER	DESCRIPTION
`query`	The search query string. TYPE: `str`
`page`	The page number for pagination. TYPE: `int` DEFAULT: `1`
`page_size`	The number of items per page. TYPE: `int` DEFAULT: `10`

Returns: List of image assets matching the query.

`update(id_, data)` `async` ¶

Update an existing image asset.

`AsyncProductImageManager` ¶

Bases: BaseProductImageManager, BaseAsyncManager

Asynchronous manager for product image resources.

`create(data)` `async` ¶

Create a new product image.

PARAMETER	DESCRIPTION
`data`	Product image creation data including file or URL TYPE: `ProductImageCreate`

RETURNS	DESCRIPTION
`ProductImage`	Created product image instance with Active Record capabilities

`get(id_)` `async` ¶

Get a specific product image by ID.

PARAMETER	DESCRIPTION
`id_`	Product image ID (UUID) TYPE: `str`

RETURNS	DESCRIPTION
`ProductImage`	Product image instance with full detail data

`update(id_, data)` `async` ¶

Update an existing product image.

PARAMETER	DESCRIPTION
`id_`	Product image ID (UUID) TYPE: `str`
`data`	Product image update data with partial fields TYPE: `ProductImageUpdate`

RETURNS	DESCRIPTION
`ProductImage`	Updated product image instance

`AsyncProductManager` ¶

Bases: BaseProductManager, BaseAsyncManager

Asynchronous manager for product resources.

`create(data)` `async` ¶

Create a new product.

PARAMETER	DESCRIPTION
`data`	Product creation data including all required fields TYPE: `ProductCreate`

RETURNS	DESCRIPTION
`Product`	Created product instance with Active Record capabilities

`get(id_)` `async` ¶

Get a specific product by ID.

PARAMETER	DESCRIPTION
`id_`	Product ID (UUID) TYPE: `str`

RETURNS	DESCRIPTION
`Product`	Product instance with full detail data and nested managers

`update(id_, data)` `async` ¶

Update an existing product.

PARAMETER	DESCRIPTION
`id_`	Product ID (UUID) TYPE: `str`
`data`	Product update data with partial fields TYPE: `ProductUpdate`

RETURNS	DESCRIPTION
`Product`	Updated product instance

`AsyncProductVariantManager` ¶

Bases: BaseProductVariantManager, BaseAsyncManager

Asynchronous manager for product variant resources.

`get(id_)` `async` ¶

Get a specific product variant by ID.

PARAMETER	DESCRIPTION
`id_`	Product variant ID (UUID) TYPE: `str`

RETURNS	DESCRIPTION
`ProductVariant`	Product variant instance with full detail data

`update(id_, data)` `async` ¶

Update an existing product variant.

PARAMETER	DESCRIPTION
`id_`	Product variant ID (UUID) TYPE: `str`
`data`	Product variant update data with partial fields TYPE: `ProductVariantUpdate`

RETURNS	DESCRIPTION
`ProductVariant`	Updated product variant instance

`AsyncUploadManager` ¶

Bases: BaseUploadManager, BaseAsyncManager

Async manager for upload resources.

`create(data)` `async` ¶

Create a new upload.

`create_cloud_upload(mime_type, file_size, filename, name=None)` `async` ¶

Create a presigned upload URL for direct to cloud upload.

PARAMETER	DESCRIPTION
`mime_type`	MIME type of the file to upload TYPE: `str`
`file_size`	Size of the file in bytes TYPE: `int`
`filename`	Name of the file TYPE: `str`
`name`	Display name for the upload (optional) TYPE: `Optional[str]` DEFAULT: `None`

RETURNS	DESCRIPTION
`Dict[str, Any]`	Upload info with presigned URL details

`mark_completed(id_)` `async` ¶

Mark the upload as completed.

This is called for direct to cloud uploads once the upload is completed.

PARAMETER	DESCRIPTION
`id_`	Upload ID TYPE: `str`

RETURNS	DESCRIPTION
`Upload`	Updated upload instance

`mark_failed(id_)` `async` ¶

Mark the upload as failed.

This is called for direct to cloud uploads if the upload fails.

PARAMETER	DESCRIPTION
`id_`	Upload ID TYPE: `str`

RETURNS	DESCRIPTION
`Upload`	Updated upload instance

`mark_started(id_)` `async` ¶

Mark the upload as started.

This is called for direct to cloud uploads once the upload is started.

PARAMETER	DESCRIPTION
`id_`	Upload ID TYPE: `str`

RETURNS	DESCRIPTION
`Upload`	Updated upload instance

`update(id_, data)` `async` ¶

Update operation not supported for uploads.

`AsyncVideoManager` ¶

Bases: BaseVideoManager, BaseAsyncManager

Async manager for video assets.

`create(data)` `async` ¶

Create a new video asset.

`search(query, page=1, page_size=10)` `async` ¶

Search for videos by a text query.

PARAMETER	DESCRIPTION
`query`	The search query string. TYPE: `str`
`page`	The page number for pagination. TYPE: `int` DEFAULT: `1`
`page_size`	The number of items per page. TYPE: `int` DEFAULT: `10`

Returns: List of video assets matching the query.

`update(id_, data)` `async` ¶

Update an existing video asset.

`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

`BaseConnectionManager` ¶

Base configuration for Connection resource managers.

`base_path = '/relay/api/v1/connections'` `class-attribute` `instance-attribute` ¶

The API base path for this resource type.

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

The resource model class this manager handles.

`BaseDocumentManager` ¶

`base_path = '/admin/api/storefront-assets'` `class-attribute` `instance-attribute` ¶

The API base path for this resource type.

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

The filter model class for this resource type.

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

The resource model class this manager handles.

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

The resource update model class this manager handles.

`BaseFilters` ¶

Bases: BaseModel

Base class for filter models.

Provides common filtering patterns and validation for all resource filter classes.

`BaseImageManager` ¶

`base_path = '/admin/api/storefront-assets'` `class-attribute` `instance-attribute` ¶

The API base path for this resource type.

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

The filter model class for this resource type.

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

The resource model class this manager handles.

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

The resource update model class this manager handles.

`BaseManager` ¶

Base manager class providing common functionality for both sync and async managers.

Contains shared methods for URL construction, resource creation, filtering, and file handling that are identical between sync and async implementations.

`base_path` `instance-attribute` ¶

The API base path for this resource type (e.g., '/media-assets').

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

The filter model class for this resource type (optional).

`resource_class` `instance-attribute` ¶

The resource model class this manager handles.

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

The update model class for this resource type.

`create_resource(data, is_partial=False)` ¶

Create resource instance with Active Record capabilities.

PARAMETER	DESCRIPTION
`data`	Resource data from API response TYPE: `Dict[str, Any]`
`is_partial`	Whether this resource was loaded from a list endpoint (may be missing detail-only fields) TYPE: `bool` DEFAULT: `False`

RETURNS	DESCRIPTION
`BaseResource`	Resource instance with manager attached

`BaseResource` ¶

Bases: BaseModel

Base class providing Active Record functionality to all resources.

This class enables resources to have methods like save(), delete(), and reload() by maintaining a reference to their manager and tracking field modifications.

`dirty_fields` `property` ¶

Get the names of fields that have been modified.

`is_dirty` `property` ¶

Check if the object has unsaved changes.

`setattr(name, value)` ¶

Track field modifications for dirty checking.

`adelete()` `async` ¶

Async version of delete() - Delete this resource from the API.

RETURNS	DESCRIPTION
`bool`	True if deletion was successful

RAISES	DESCRIPTION
`ValueError`	If resource is not associated with an async manager
`APIError`	If the API request fails

`aload_detail()` `async` ¶

Async version of load_detail() - Load full detail from API.

RETURNS	DESCRIPTION
`BaseResource`	Self with updated data from the detail API

RAISES	DESCRIPTION
`ValueError`	If resource is not associated with an async manager
`APIError`	If the API request fails

`areload()` `async` ¶

Async version of reload() - Reload fresh data from the API.

RETURNS	DESCRIPTION
`BaseResource`	Self with updated data from the server

RAISES	DESCRIPTION
`ValueError`	If resource is not associated with an async manager
`APIError`	If the API request fails

`asave()` `async` ¶

Async version of save() - Save changes back to the API.

RETURNS	DESCRIPTION
`BaseResource`	Updated resource instance with fresh data from the server

RAISES	DESCRIPTION
`ValueError`	If resource is not associated with an async manager
`APIError`	If the API request fails

`delete()` ¶

Delete this resource from the API.

RETURNS	DESCRIPTION
`bool`	True if deletion was successful

RAISES	DESCRIPTION
`ValueError`	If resource is not associated with a manager
`APIError`	If the API request fails

`is_detail_loaded()` ¶

Check if detail data is loaded.

RETURNS	DESCRIPTION
`bool`	True if this resource was loaded from a detail endpoint,
`bool`	or if there are no detail-only fields or foreign key fields.
`bool`	False if loaded from a list endpoint and has detail-only/foreign key fields.

`is_foreign_key_loaded(field_name)` ¶

Check if a foreign key field is fully loaded (not just IDs).

PARAMETER	DESCRIPTION
`field_name`	Name of the foreign key field to check TYPE: `str`

RETURNS	DESCRIPTION
`bool`	True if the field contains full objects, False if just IDs

RAISES	DESCRIPTION
`ValueError`	If field_name is not a defined foreign key field

`load_detail()` ¶

Load full detail from API and update this instance.

Fetches the complete resource from the detail endpoint and updates all fields, including detail-only fields and full foreign key objects.

RETURNS	DESCRIPTION
`BaseResource`	Self with updated data from the detail API

RAISES	DESCRIPTION
`ValueError`	If resource is not associated with a manager
`APIError`	If the API request fails

`reload()` ¶

Reload fresh data from the API.

RETURNS	DESCRIPTION
`BaseResource`	Self with updated data from the server

RAISES	DESCRIPTION
`ValueError`	If resource is not associated with a manager
`APIError`	If the API request fails

`reset_changes()` ¶

Reset all changes to original values.

`save()` ¶

Save changes back to the API.

RETURNS	DESCRIPTION
`BaseResource`	Updated resource instance with fresh data from the server

RAISES	DESCRIPTION
`ValueError`	If resource is not associated with a manager
`APIError`	If the API request fails

`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

`BaseUploadManager` ¶

`base_path = '/admin/api/uploads'` `class-attribute` `instance-attribute` ¶

The API base path for this resource type.

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

No filtering for uploads.

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

The resource model class this manager handles.

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

No update capability for uploads.

`BaseVideoManager` ¶

`base_path = '/admin/api/storefront-assets'` `class-attribute` `instance-attribute` ¶

The API base path for this resource type.

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

The filter model class for this resource type.

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

The resource model class this manager handles.

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

The resource update model class this manager handles.

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

`Connection` ¶

Bases: TimestampedResource

Connection resource model.

Represents a third-party API integration connection with OAuth or API key authentication. Connections are site-specific and store credentials, tokens, and metadata for external service integrations.

`provider = Field(..., title='Provider Code', description='Provider code identifying the third-party service.')` `class-attribute` `instance-attribute` ¶

Provider code identifying the third-party service.

`root_resource_id = Field(default=None, title='Root Resource ID', description='Authorized resource ID (e.g., account ID, email).')` `class-attribute` `instance-attribute` ¶

Authorized resource ID (e.g., account ID, email).

`root_resource_name = Field(default=None, title='Root Resource Name', description='Authorized resource name (e.g., account name).')` `class-attribute` `instance-attribute` ¶

Authorized resource name (e.g., account name).

`scopes = Field(default_factory=list, title='OAuth Scopes', description='List of OAuth scopes authorized for this connection.')` `class-attribute` `instance-attribute` ¶

List of OAuth scopes authorized for this connection.

`status = Field(default=(ConnectionStatus.ACTIVE), title='Connection Status', description='Current status of the connection.')` `class-attribute` `instance-attribute` ¶

Current status of the connection.

`token_expires_at = Field(default=None, title='Token Expiry', description='Expiration timestamp for the access token.')` `class-attribute` `instance-attribute` ¶

Expiration timestamp for the access token.

`ConnectionManager` ¶

Bases: BaseConnectionManager, BaseSyncManager

Manager for connection resources.

`create(*args, **kwargs)` ¶

Creation not supported via SDK.

`delete(*args, **kwargs)` ¶

Deletion not supported via SDK.

`get(id_)` ¶

Get a specific connection by ID.

`get_credentials(provider)` ¶

Get connection credentials by ID.

Returns the raw credentials dictionary for the connection, which may include sensitive information like tokens.

PARAMETER	DESCRIPTION
`provider`	Connection provider code TYPE: `str`

RETURNS	DESCRIPTION
`Dict[str, str]`	Dictionary containing connection credentials

`update(*args, **kwargs)` ¶

Update not supported via SDK.

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

`CreateModel` ¶

Bases: BaseModel

Base class for creation models.

Used for models that define fields required/allowed when creating new resources.

`Document` ¶

Bases: BaseAsset

Document asset model.

`DocumentCreate` ¶

Bases: BaseAssetCreate

Create model for document assets.

`file = Field(..., title='File', description='Document file to upload. Can be file path, bytes, BytesIO, BinaryIO, or tuple of (file_data, filename)')` `class-attribute` `instance-attribute` ¶

Document file to upload. Can be file path, bytes, BytesIO, BinaryIO, or tuple of (file_data, filename)

`normalize_file()` ¶

Convert file paths or bytes to FileInput.

`validate_document_file(v)` `classmethod` ¶

Ensure we're uploading a supported document file.

`DocumentFilters` ¶

Bases: MediaAssetFilters

Filters specific to document assets.

`mime_type = Field('document')` `class-attribute` `instance-attribute` ¶

Automatically set to 'document' for document filters

`DocumentManager` ¶

Bases: BaseDocumentManager, BaseSyncManager

Manager for document assets.

`create(data)` ¶

Create a new document asset.

`search(query, page=1, page_size=10)` ¶

Search for documents by a text query.

PARAMETER	DESCRIPTION
`query`	The search query string. TYPE: `str`
`page`	The page number for pagination. TYPE: `int` DEFAULT: `1`
`page_size`	The number of items per page. TYPE: `int` DEFAULT: `10`

Returns: List of document assets matching the query.

`update(id_, data)` ¶

Update an existing document asset.

`DocumentUpdate` ¶

Bases: BaseAssetUpdate

Update model for document assets - file cannot be updated.

`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.

`FormManager` ¶

Bases: BaseFormManager, BaseSyncManager

Manager for form resources.

`create(data)` ¶

Create a new form.

PARAMETER	DESCRIPTION
`data`	Form creation data TYPE: `Union[FormCreate, Dict[str, str]]`

RETURNS	DESCRIPTION
`Form`	Created form instance

`delete(*args, **kwargs)` ¶

Deletion not supported.

`get(slug)` ¶

Get a specific form by slug

`list(page=1, page_size=20)` ¶

List forms with pagination.

PARAMETER	DESCRIPTION
`page`	Page number (1-based) TYPE: `int` DEFAULT: `1`
`page_size`	Number of items per page TYPE: `int` DEFAULT: `20`
`**filter_kwargs`	Filter arguments as keyword arguments

RETURNS	DESCRIPTION
`PaginatedResponse[Form]`	Paginated response containing list of forms

`list_submissions(slug, page=1, page_size=20)` ¶

List submissions for a specific form.

PARAMETER	DESCRIPTION
`slug`	Form slug TYPE: `str`
`page`	Page number (1-based) TYPE: `int` DEFAULT: `1`
`page_size`	Number of items per page TYPE: `int` DEFAULT: `20`

RETURNS	DESCRIPTION
`PaginatedResponse[FormSubmission]`	Paginated response containing list of form submissions

`update(*args, **kwargs)` ¶

Update not supported.

`Image` ¶

Bases: BaseAsset

Image asset model.

`image_height = Field(None, ge=0, title='Image Height', description='Image height in pixels')` `class-attribute` `instance-attribute` ¶

Image height in pixels

`image_width = Field(None, ge=0, title='Image Width', description='Image width in pixels')` `class-attribute` `instance-attribute` ¶

Image width in pixels

`ImageCreate` ¶

Bases: BaseAssetCreate

Create model for image assets.

`image = Field(..., title='Image', description='Image file to upload. Can be file path, bytes, BytesIO, BinaryIO, or tuple of (file_data, filename)')` `class-attribute` `instance-attribute` ¶

Image file to upload. Can be file path, bytes, BytesIO, BinaryIO, or tuple of (file_data, filename)

`normalize_file()` ¶

Convert file paths or bytes to FileInput.

`validate_image_file(v)` `classmethod` ¶

Ensure we're uploading an image file.

`ImageFilters` ¶

Bases: MediaAssetFilters

Filters specific to image assets.

`mime_type = Field('image')` `class-attribute` `instance-attribute` ¶

Automatically set to 'image' for image filters

`ImageGenerate` ¶

Bases: CreateModel

Model for generating images from text prompts.

`aspect_ratio = Field(default='1:1', title='Aspect Ratio', description='Aspect ratio of the generated image')` `class-attribute` `instance-attribute` ¶

Aspect ratio of the generated image

`output_format = Field(default='webp', title='Output Format', description='Output format of the generated image')` `class-attribute` `instance-attribute` ¶

Output format of the generated image

`prompt = Field(..., title='Prompt', description='Text prompt to generate the image')` `class-attribute` `instance-attribute` ¶

Text prompt to generate the image

`ImageManager` ¶

Bases: BaseImageManager, BaseSyncManager

Manager for image assets.

`create(data)` ¶

Create a new image asset.

`generate(data)` ¶

Generate an image from a text prompt.

PARAMETER	DESCRIPTION
`data`	Image generation parameters including prompt and options TYPE: `ImageGenerate`

RETURNS	DESCRIPTION
`Image`	Generated image asset

`search(query, page=1, page_size=10)` ¶

Search for images by a text query.

PARAMETER	DESCRIPTION
`query`	The search query string. TYPE: `str`
`page`	The page number for pagination. TYPE: `int` DEFAULT: `1`
`page_size`	The number of items per page. TYPE: `int` DEFAULT: `10`

Returns: List of image assets matching the query.

`update(id_, data)` ¶

Update an existing image asset.

`ImageUpdate` ¶

Bases: BaseAssetUpdate

Update model for image assets - file cannot be updated.

`PaginatedResponse` ¶

Bases: BaseModel, Generic[T]

Standardized pagination response wrapper.

This generic class wraps paginated API responses with metadata about the pagination state and navigation.

`count = Field(..., ge=0, title='Count', description='Total number of resources.')` `class-attribute` `instance-attribute` ¶

Total number of resources.

`current_page = Field(..., ge=1, title='Current Page', description='Current page number.')` `class-attribute` `instance-attribute` ¶

Current page number.

`is_first_page` `property` ¶

Check if this is the first page.

`is_last_page` `property` ¶

Check if this is the last page.

`next = Field(None, title='Next', description='URL for next page.')` `class-attribute` `instance-attribute` ¶

URL for next page.

`num_pages = Field(..., ge=0, title='Number of Pages', description='Total number of pages.')` `class-attribute` `instance-attribute` ¶

Total number of pages.

`page_size = Field(..., gt=0, title='Page Size', description='Number of items per page.')` `class-attribute` `instance-attribute` ¶

Number of items per page.

`payload = Field(..., title='Payload', description='List of resources in this page.')` `class-attribute` `instance-attribute` ¶

List of resources in this page.

`previous = Field(None, title='Previous', description='URL for previous page.')` `class-attribute` `instance-attribute` ¶

URL for previous page.

`has_next()` ¶

Check if there is a next page.

`has_previous()` ¶

Check if there is a previous page.

`PaginationParams` ¶

Bases: BaseModel

Pagination parameter validation and defaults.

Used to validate and normalize pagination parameters across all list operations.

`page = Field(1, ge=1, title='Page', description='Page number.')` `class-attribute` `instance-attribute` ¶

Page number.

`page_size = Field(20, ge=1, le=200, title='Page Size', description='Items per page.')` `class-attribute` `instance-attribute` ¶

Items per page.

`ProductImageManager` ¶

Bases: BaseProductImageManager, BaseSyncManager

Synchronous manager for product image resources.

`create(data)` ¶

Create a new product image.

PARAMETER	DESCRIPTION
`data`	Product image creation data including file or URL TYPE: `ProductImageCreate`

RETURNS	DESCRIPTION
`ProductImage`	Created product image instance with Active Record capabilities

`get(id_)` ¶

Get a specific product image by ID.

PARAMETER	DESCRIPTION
`id_`	Product image ID (UUID) TYPE: `str`

RETURNS	DESCRIPTION
`ProductImage`	Product image instance with full detail data

`update(id_, data)` ¶

Update an existing product image.

PARAMETER	DESCRIPTION
`id_`	Product image ID (UUID) TYPE: `str`
`data`	Product image update data with partial fields TYPE: `ProductImageUpdate`

RETURNS	DESCRIPTION
`ProductImage`	Updated product image instance

`ProductManager` ¶

Bases: BaseProductManager, BaseSyncManager

Synchronous manager for product resources.

`create(data)` ¶

Create a new product.

PARAMETER	DESCRIPTION
`data`	Product creation data including all required fields TYPE: `ProductCreate`

RETURNS	DESCRIPTION
`Product`	Created product instance with Active Record capabilities

`get(id_)` ¶

Get a specific product by ID.

PARAMETER	DESCRIPTION
`id_`	Product ID (UUID) TYPE: `str`

RETURNS	DESCRIPTION
`Product`	Product instance with full detail data and nested managers

`update(id_, data)` ¶

Update an existing product.

PARAMETER	DESCRIPTION
`id_`	Product ID (UUID) TYPE: `str`
`data`	Product update data with partial fields TYPE: `ProductUpdate`

RETURNS	DESCRIPTION
`Product`	Updated product instance

`ProductVariantManager` ¶

Bases: BaseProductVariantManager, BaseSyncManager

Synchronous manager for product variant resources.

`get(id_)` ¶

Get a specific product variant by ID.

PARAMETER	DESCRIPTION
`id_`	Product variant ID (UUID) TYPE: `str`

RETURNS	DESCRIPTION
`ProductVariant`	Product variant instance with full detail data

`update(id_, data)` ¶

Update an existing product variant.

PARAMETER	DESCRIPTION
`id_`	Product variant ID (UUID) TYPE: `str`
`data`	Product variant update data with partial fields TYPE: `ProductVariantUpdate`

RETURNS	DESCRIPTION
`ProductVariant`	Updated product variant instance

`UpdateModel` ¶

Bases: BaseModel

Base class for update models.

Used for models that define fields allowed when updating existing resources. All fields are optional.

`Upload` ¶

Bases: Resource

Upload model for general purpose file uploads.

`created_at = Field(..., title='Created At', description='Creation timestamp.')` `class-attribute` `instance-attribute` ¶

Creation timestamp.

`description = Field(default='', max_length=1000, title='Description', description='Optional description of the upload')` `class-attribute` `instance-attribute` ¶

Optional description of the upload

`etag = Field(default='', max_length=64, title='ETag', description='ETag or checksum of the file content')` `class-attribute` `instance-attribute` ¶

ETag or checksum of the file content

`meta = Field(default_factory=dict, title='Metadata', description='Additional metadata such as width, height for images')` `class-attribute` `instance-attribute` ¶

Additional metadata such as width, height for images

`mime_type = Field(..., max_length=100, title='MIME Type', description='MIME type of the uploaded file')` `class-attribute` `instance-attribute` ¶

MIME type of the uploaded file

`name = Field(default='', max_length=255, title='Name', description='Name of the file for display and search purposes')` `class-attribute` `instance-attribute` ¶

Name of the file for display and search purposes

`size = Field(default=0, ge=0, title='Size', description='Size of the file in bytes')` `class-attribute` `instance-attribute` ¶

Size of the file in bytes

`status = Field(default=(UploadStatus.COMPLETED), title='Status', description='Status of the upload')` `class-attribute` `instance-attribute` ¶

Status of the upload

`storage_path = Field(default='', max_length=500, title='Storage Path', description='Object key or path in the storage backend')` `class-attribute` `instance-attribute` ¶

Object key or path in the storage backend

`tags = Field(default_factory=list, title='Tags', description='Tags for categorization')` `class-attribute` `instance-attribute` ¶

Tags for categorization

`upload_ended_at = Field(None, title='Upload Ended At', description='Timestamp when the upload ended')` `class-attribute` `instance-attribute` ¶

Timestamp when the upload ended

`upload_started_at = Field(None, title='Upload Started At', description='Timestamp when the upload started')` `class-attribute` `instance-attribute` ¶

Timestamp when the upload started

`url = Field(..., title='File URL', description='URL to access the uploaded file')` `class-attribute` `instance-attribute` ¶

URL to access the uploaded file

`UploadCreate` ¶

Bases: CreateModel

Create model for file uploads.

`description = Field(default=None, title='Description', description='Optional description of the upload')` `class-attribute` `instance-attribute` ¶

Optional description of the upload

`file = Field(..., title='File', description='File to upload. Can be file path, bytes, BytesIO, BinaryIO, or tuple of (file_data, filename)')` `class-attribute` `instance-attribute` ¶

File to upload. Can be file path, bytes, BytesIO, BinaryIO, or tuple of (file_data, filename)

`meta = Field(default=None, title='Meta', description='Optional metadata dictionary')` `class-attribute` `instance-attribute` ¶

Optional metadata dictionary

`name = Field(default='', max_length=255, title='Name', description='Name of the file for display and search purposes')` `class-attribute` `instance-attribute` ¶

Name of the file for display and search purposes

`tags = Field(default='', title='Tags', description='Tags for categorization. Comma-separated string.')` `class-attribute` `instance-attribute` ¶

Tags for categorization. Comma-separated string.

`normalize_file()` ¶

Convert file paths or bytes to FileInput.

`validate_file(v)` `classmethod` ¶

Ensure we're uploading a supported file type.

`UploadManager` ¶

Bases: BaseUploadManager, BaseSyncManager

Manager for upload resources.

`create(data)` ¶

Create a new upload.

`create_cloud_upload(mime_type, file_size, filename, name=None)` ¶

Create a presigned upload URL for direct to cloud upload.

PARAMETER	DESCRIPTION
`mime_type`	MIME type of the file to upload TYPE: `str`
`file_size`	Size of the file in bytes TYPE: `int`
`filename`	Name of the file TYPE: `str`
`name`	Display name for the upload (optional) TYPE: `Optional[str]` DEFAULT: `None`

RETURNS	DESCRIPTION
`Dict[str, Any]`	Upload info with presigned URL details

`mark_completed(id_)` ¶

Mark the upload as completed.

This is called for direct to cloud uploads once the upload is completed.

PARAMETER	DESCRIPTION
`id_`	Upload ID TYPE: `str`

RETURNS	DESCRIPTION
`Upload`	Updated upload instance

`mark_failed(id_)` ¶

Mark the upload as failed.

This is called for direct to cloud uploads if the upload fails.

PARAMETER	DESCRIPTION
`id_`	Upload ID TYPE: `str`

RETURNS	DESCRIPTION
`Upload`	Updated upload instance

`mark_started(id_)` ¶

Mark the upload as started.

This is called for direct to cloud uploads once the upload is started.

PARAMETER	DESCRIPTION
`id_`	Upload ID TYPE: `str`

RETURNS	DESCRIPTION
`Upload`	Updated upload instance

`update(id_, data)` ¶

Update operation not supported for uploads.

`Video` ¶

Bases: BaseAsset

Video asset model.

`VideoCreate` ¶

Bases: BaseAssetCreate

Create model for video assets.

`file = Field(..., title='File', description='Video file to upload. Can be file path, bytes, BytesIO, BinaryIO, or tuple of (file_data, filename)')` `class-attribute` `instance-attribute` ¶

Video file to upload. Can be file path, bytes, BytesIO, BinaryIO, or tuple of (file_data, filename)

`normalize_file()` ¶

Convert file paths or bytes to FileInput.

`validate_video_file(v)` `classmethod` ¶

Ensure we're uploading a video file.

`VideoFilters` ¶

Bases: MediaAssetFilters

Filters specific to video assets.

`mime_type = Field('video')` `class-attribute` `instance-attribute` ¶

Automatically set to 'video' for video filters

`VideoManager` ¶

Bases: BaseVideoManager, BaseSyncManager

Manager for video assets.

`create(data)` ¶

Create a new video asset.

`search(query, page=1, page_size=10)` ¶

Search for videos by a text query.

PARAMETER	DESCRIPTION
`query`	The search query string. TYPE: `str`
`page`	The page number for pagination. TYPE: `int` DEFAULT: `1`
`page_size`	The number of items per page. TYPE: `int` DEFAULT: `10`

Returns: List of video assets matching the query.

`update(id_, data)` ¶

Update an existing video asset.

`VideoUpdate` ¶

Bases: BaseAssetUpdate

Update model for video assets - file cannot be updated.

`get_logger()` ¶

Get the main Konigle logger.

RETURNS	DESCRIPTION
`Logger`	Main logger instance for the SDK

`model_to_dict(model)` ¶

Convert a Pydantic model to a dict, excluding None and unset fields and handling file fields appropriately for multipart requests.

managers

konigle.managers ¶

AsyncAudienceManager ¶

create(data) async ¶

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 ¶

AsyncConnectionManager ¶

create(*args, **kwargs) async ¶

delete(*args, **kwargs) async ¶

get(id_) async ¶

get_credentials(provider) async ¶

update(*args, **kwargs) async ¶

AsyncContactManager ¶

create(data) async ¶

update_ltv(data) async ¶

AsyncDocumentManager ¶

create(data) async ¶

search(query, page=1, page_size=10) async ¶

update(id_, data) async ¶

AsyncEmailAccountManager ¶

check_status() async ¶

create(data) async ¶

setup(data) async ¶

AsyncEmailChannelManager ¶

create(data) async ¶

set_engagement_tracking(channel_id, enable) async ¶

AsyncEmailIdentityManager ¶

check_verification_status(identity_id) async ¶

create(data) async ¶

setup_custom_mail_from(identity_id, mail_from_domain) async ¶

AsyncFormManager ¶

create(data) async ¶

delete(*args, **kwargs) async ¶

get(slug) async ¶

list(page=1, page_size=20) async ¶

list_submissions(slug, page=1, page_size=20) async ¶

update(*args, **kwargs) async ¶

AsyncImageManager ¶

create(data) async ¶

generate(data) async ¶

search(query, page=1, page_size=10) async ¶

update(id_, data) async ¶

AsyncProductImageManager ¶

create(data) async ¶

get(id_) async ¶

update(id_, data) async ¶

AsyncProductManager ¶

create(data) async ¶

get(id_) async ¶

update(id_, data) async ¶

AsyncProductVariantManager ¶

get(id_) async ¶

update(id_, data) async ¶

AsyncUploadManager ¶

create(data) async ¶

create_cloud_upload(mime_type, file_size, filename, name=None) async ¶

mark_completed(id_) async ¶

mark_failed(id_) async ¶

mark_started(id_) async ¶

update(id_, data) async ¶

AsyncVideoManager ¶

create(data) async ¶

search(query, page=1, page_size=10) async ¶

update(id_, data) async ¶

AudienceManager ¶

create(data) ¶

BaseAsyncManager ¶

__init_subclass__(**kwargs) ¶

create(data) async ¶

delete(id_) async ¶

get(id_) async ¶

iter_all(page_size=100, filters=None, **filter_kwargs) async ¶

`konigle.managers` ¶

`AsyncAudienceManager` ¶

`create(data)` `async` ¶

`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` ¶

`AsyncConnectionManager` ¶

`create(*args, **kwargs)` `async` ¶

`delete(*args, **kwargs)` `async` ¶

`get(id_)` `async` ¶

`get_credentials(provider)` `async` ¶

`update(*args, **kwargs)` `async` ¶

`AsyncContactManager` ¶

`create(data)` `async` ¶

`update_ltv(data)` `async` ¶

`AsyncDocumentManager` ¶

`create(data)` `async` ¶

`search(query, page=1, page_size=10)` `async` ¶

`update(id_, data)` `async` ¶

`AsyncEmailAccountManager` ¶

`check_status()` `async` ¶

`create(data)` `async` ¶

`setup(data)` `async` ¶

`AsyncEmailChannelManager` ¶

`create(data)` `async` ¶

`set_engagement_tracking(channel_id, enable)` `async` ¶

`AsyncEmailIdentityManager` ¶

`check_verification_status(identity_id)` `async` ¶

`create(data)` `async` ¶

`setup_custom_mail_from(identity_id, mail_from_domain)` `async` ¶

`AsyncFormManager` ¶

`create(data)` `async` ¶

`delete(*args, **kwargs)` `async` ¶

`get(slug)` `async` ¶

`list(page=1, page_size=20)` `async` ¶

`list_submissions(slug, page=1, page_size=20)` `async` ¶

`update(*args, **kwargs)` `async` ¶

`AsyncImageManager` ¶

`create(data)` `async` ¶

`generate(data)` `async` ¶

`search(query, page=1, page_size=10)` `async` ¶

`update(id_, data)` `async` ¶

`AsyncProductImageManager` ¶

`create(data)` `async` ¶

`get(id_)` `async` ¶

`update(id_, data)` `async` ¶

`AsyncProductManager` ¶

`create(data)` `async` ¶

`get(id_)` `async` ¶

`update(id_, data)` `async` ¶

`AsyncProductVariantManager` ¶

`get(id_)` `async` ¶

`update(id_, data)` `async` ¶

`AsyncUploadManager` ¶

`create(data)` `async` ¶

`create_cloud_upload(mime_type, file_size, filename, name=None)` `async` ¶

`mark_completed(id_)` `async` ¶

`mark_failed(id_)` `async` ¶

`mark_started(id_)` `async` ¶

`update(id_, data)` `async` ¶

`AsyncVideoManager` ¶

`create(data)` `async` ¶

`search(query, page=1, page_size=10)` `async` ¶

`update(id_, data)` `async` ¶

`AudienceManager` ¶

`create(data)` ¶

`BaseAsyncManager` ¶

`__init_subclass__(**kwargs)` ¶

`create(data)` `async` ¶

`delete(id_)` `async` ¶

`get(id_)` `async` ¶

`iter_all(page_size=100, filters=None, **filter_kwargs)` `async` ¶

`list(page=1, page_size=20, filters=None, **filter_kwargs)` `async` ¶