Skip to content

core

konigle.managers.core

Core resource managers for the Konigle SDK.

This module exports managers for core platform resources like media assets, uploads, connections, and forms.

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.

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.

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.

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.

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.

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.

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.

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.

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.

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.

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.

model_to_dict(model)

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