plexus.dashboard.api.models package

Plexus Dashboard API Models

class plexus.dashboard.api.models.Account(id: str, name: str, key: str, description: str | None = None, settings: Dict | None = None, client: _BaseAPIClient | None = None)

Bases: BaseModel

Account model for interacting with the API.

__init__(id: str, name: str, key: str, description: str | None = None, settings: Dict | None = None, client: _BaseAPIClient | None = None)

description: str | None = None

classmethod fields() → str: Return the GraphQL fields to query for this model

classmethod from_dict(data: Dict[str, Any], client: _BaseAPIClient) → Account: Create an Account instance from a dictionary.

classmethod get_by_id(id: str, client: _BaseAPIClient) → Account: Get an account by its ID.

classmethod get_by_key(key: str, client: _BaseAPIClient) → Account | None

Get an account by its key.

Args:: key: The account key to look up client: The API client instance
Returns:: Account: The found account or None

key: str

classmethod list_by_key(client: _BaseAPIClient, key: str) → Account | None

Get an account by its key.

Args:: client: The API client instance key: The account key to look up
Returns:: Account: The found account or None

name: str

settings: Dict | None = None

update(**kwargs) → Account: Update the account.

class plexus.dashboard.api.models.BatchJob(id: str, accountId: str, status: str, type: str, batchId: str, modelProvider: str, modelName: str, scoringJobCountCache: int | None = None, startedAt: datetime | None = None, estimatedEndAt: datetime | None = None, completedAt: datetime | None = None, totalRequests: int | None = None, completedRequests: int | None = None, failedRequests: int | None = None, errorMessage: str | None = None, errorDetails: Dict | None = None, scorecardId: str | None = None, scoreId: str | None = None, client: _BaseAPIClient | None = None)

Bases: BaseModel

Represents a batch processing job for scoring multiple items.

A BatchJob manages the execution of scoring operations across multiple items, tracking progress, handling errors, and maintaining state throughout the batch processing lifecycle.

Parameters:

accountId – Identifier for the account owning this batch job
status – Current status of the batch job (PENDING, RUNNING, etc.)
type – Type of batch operation being performed
batchId – Unique identifier for this batch of operations
modelProvider – Provider of the model being used
modelName – Name of the model being used
scoringJobCountCache – Number of scoring jobs in this batch
startedAt – Timestamp when the batch job started
estimatedEndAt – Estimated completion timestamp
completedAt – Timestamp when the batch job completed
totalRequests – Total number of requests in this batch
completedRequests – Number of completed requests
failedRequests – Number of failed requests
errorMessage – Error message if the batch job failed
errorDetails – Detailed error information
scorecardId – Associated scorecard identifier
scoreId – Associated score identifier

__init__(id: str, accountId: str, status: str, type: str, batchId: str, modelProvider: str, modelName: str, scoringJobCountCache: int | None = None, startedAt: datetime | None = None, estimatedEndAt: datetime | None = None, completedAt: datetime | None = None, totalRequests: int | None = None, completedRequests: int | None = None, failedRequests: int | None = None, errorMessage: str | None = None, errorDetails: Dict | None = None, scorecardId: str | None = None, scoreId: str | None = None, client: _BaseAPIClient | None = None)

accountId: str

batchId: str

completedAt: datetime | None = None

completedRequests: int | None = None

classmethod create(client: _BaseAPIClient, accountId: str, type: str, modelProvider: str, modelName: str, parameters: Dict, **kwargs) → BatchJob

Creates a new batch job.

Parameters:

client – API client instance for making requests
accountId – Account identifier for the batch job
type – Type of batch operation
modelProvider – Provider of the model
modelName – Name of the model
parameters – Additional parameters for job creation
kwargs – Optional parameters (status, scorecardId, etc.)

Returns:

New BatchJob instance

Raises:

Exception if creation fails

errorDetails: Dict | None = None

errorMessage: str | None = None

estimatedEndAt: datetime | None = None

failedRequests: int | None = None

classmethod fields() → str: Return the GraphQL fields to query for this model

classmethod from_dict(data: Dict[str, Any], client: _BaseAPIClient) → BatchJob

Creates a BatchJob instance from dictionary data.

Parameters:

data – Dictionary containing batch job data
client – API client instance

Returns:

BatchJob instance

classmethod get_by_id(id: str, client: _BaseAPIClient) → BatchJob

Retrieves a batch job by its identifier.

Parameters:

id – Batch job identifier
client – API client instance

Returns:

BatchJob instance

Raises:

Exception if batch job not found

modelName: str

modelProvider: str

scoreId: str | None = None

scorecardId: str | None = None

scoringJobCountCache: int | None = None

startedAt: datetime | None = None

status: str

totalRequests: int | None = None

type: str

update(**kwargs) → BatchJob

Updates batch job fields.

Parameters:: kwargs – Fields to update and their new values
Returns:: Updated BatchJob instance
Raises:: ValueError if attempting to modify createdAt

class plexus.dashboard.api.models.DataSet(id: str, client: PlexusDashboardClient | None = None)

Bases: BaseModel

createdAt: str

dataSource: DataSource | None

dataSourceId: str

errorMessage: str | None

filePath: str | None

id: str

name: str

source: str | None

status: str | None

updatedAt: str

Bases: BaseModel

__init__(id: str, name: str, key: str | None = None, description: str | None = None, yamlConfiguration: str | None = None, attachedFiles: List[str] | None = None, createdAt: str | None = None, updatedAt: str | None = None, owner: str | None = None, accountId: str | None = None, currentVersionId: str | None = None, scoreId: str | None = None, scorecardId: str | None = None, dataSets: List[DataSet] | None = None, client: PlexusDashboardClient | None = None)

classmethod fields() → str: Return the GraphQL fields to query for this model

classmethod from_dict(data: Dict[str, Any], client: PlexusDashboardClient) → DataSource: Create a DataSource instance from a dictionary.

async classmethod get(client: PlexusDashboardClient, id: str) → DataSource | None: Fetch a DataSource by its ID.

async classmethod get_by_name(name: str, client: PlexusDashboardClient) → DataSource | None: Fetch a DataSource by its name.

async classmethod list(client: PlexusDashboardClient, filter: Dict | None = None) → List[DataSource]: List DataSources with optional filtering.

async classmethod list_by_key(client: PlexusDashboardClient, key: str) → List[DataSource]: Fetch DataSources by key using the GSI.

async classmethod list_by_name(client: PlexusDashboardClient, name: str) → List[DataSource]: Fetch DataSources by name using the GSI.

class plexus.dashboard.api.models.Evaluation(id: str, type: str, accountId: str, status: str, createdAt: datetime.datetime, updatedAt: datetime.datetime, client: ForwardRef('_BaseAPIClient') | None = None, parameters: Dict | None = None, metrics: Dict | None = None, inferences: int | None = None, accuracy: float | None = None, cost: float | None = None, startedAt: datetime.datetime | None = None, elapsedSeconds: int | None = None, estimatedRemainingSeconds: int | None = None, totalItems: int | None = None, processedItems: int | None = None, errorMessage: str | None = None, errorDetails: Dict | None = None, scorecardId: str | None = None, scoreId: str | None = None, scoreVersionId: str | None = None, confusionMatrix: Dict | None = None, scoreGoal: str | None = None, datasetClassDistribution: Dict | None = None, isDatasetClassDistributionBalanced: bool | None = None, predictedClassDistribution: Dict | None = None, isPredictedClassDistributionBalanced: bool | None = None, taskId: str | None = None)

Bases: BaseModel

__init__(id: str, type: str, accountId: str, status: str, createdAt: datetime, updatedAt: datetime, client: _BaseAPIClient | None = None, parameters: Dict | None = None, metrics: Dict | None = None, inferences: int | None = None, accuracy: float | None = None, cost: float | None = None, startedAt: datetime | None = None, elapsedSeconds: int | None = None, estimatedRemainingSeconds: int | None = None, totalItems: int | None = None, processedItems: int | None = None, errorMessage: str | None = None, errorDetails: Dict | None = None, scorecardId: str | None = None, scoreId: str | None = None, scoreVersionId: str | None = None, confusionMatrix: Dict | None = None, scoreGoal: str | None = None, datasetClassDistribution: Dict | None = None, isDatasetClassDistributionBalanced: bool | None = None, predictedClassDistribution: Dict | None = None, isPredictedClassDistributionBalanced: bool | None = None, taskId: str | None = None)

accountId: str

accuracy: float | None = None

confusionMatrix: Dict | None = None

cost: float | None = None

classmethod create(client: _BaseAPIClient, type: str, accountId: str, *, status: str = 'PENDING', scorecardId: str | None = None, scoreId: str | None = None, taskId: str | None = None, **kwargs) → Evaluation: Create a new Evaluation.

createdAt: datetime

datasetClassDistribution: Dict | None = None

elapsedSeconds: int | None = None

errorDetails: Dict | None = None

errorMessage: str | None = None

estimatedRemainingSeconds: int | None = None

classmethod fields() → str: Fields to request in queries and mutations

classmethod from_dict(data: Dict[str, Any], client: _BaseAPIClient) → Evaluation: Create an instance from a dictionary of data

classmethod get_by_id(id: str, client: _BaseAPIClient, include_score_results: bool = False) → Evaluation

inferences: int | None = None

isDatasetClassDistributionBalanced: bool | None = None

isPredictedClassDistributionBalanced: bool | None = None

metrics: Dict | None = None

parameters: Dict | None = None

predictedClassDistribution: Dict | None = None

processedItems: int | None = None

scoreGoal: str | None = None

scoreId: str | None = None

scoreVersionId: str | None = None

scorecardId: str | None = None

startedAt: datetime | None = None

status: str

taskId: str | None = None

totalItems: int | None = None

type: str

update(**kwargs) → None

Update Evaluation fields in a background thread.

This is a non-blocking operation - the mutation is performed in a background thread.

Args:: **kwargs: Fields to update

updatedAt: datetime

Bases: BaseModel

Represents an individual change record contributing to a FeedbackItem. Corresponds to the FeedbackChangeDetail model in the GraphQL schema.

GRAPHQL_BASE_FIELDS = ['id', 'feedbackItemId', 'changeType', 'externalId', 'changedAt', 'changedBy', 'initialAnswerValue', 'finalAnswerValue', 'initialCommentValue', 'finalCommentValue', 'editCommentValue', 'createdAt', 'updatedAt']

__init__(id: str | None = None, feedbackItemId: str | None = None, changeType: Literal['Response', 'Score', 'Calibration'] | None = None, externalId: str | None = None, changedAt: datetime | None = None, changedBy: str | None = None, initialAnswerValue: str | None = None, finalAnswerValue: str | None = None, initialCommentValue: str | None = None, finalCommentValue: str | None = None, editCommentValue: str | None = None, createdAt: datetime | None = None, updatedAt: datetime | None = None, feedbackItem: FeedbackItem | None = None, _client: PlexusDashboardClient | None = None, _raw_data: Dict[str, Any] | None = None) → None

changeType: Literal['Response', 'Score', 'Calibration'] | None = None

changedAt: datetime | None = None

changedBy: str | None = None

classmethod count_by_account_id(account_id: str, client: PlexusDashboardClient) → int

Count the number of FeedbackChangeDetail records associated with a specific account.

Since FeedbackChangeDetail doesn’t have a direct accountId field, we’ll need to: 1. Get all FeedbackItems for the account 2. Count all associated FeedbackChangeDetails

Args:: account_id: The ID of the account to count records for. client: The API client to use.
Returns:: int: The number of FeedbackChangeDetail records for the account.

classmethod create(client: PlexusDashboardClient, data: Dict[str, Any]) → FeedbackChangeDetail | None: Create a new FeedbackChangeDetail.

createdAt: datetime | None = None

classmethod delete_all_by_account_id(account_id: str, client: PlexusDashboardClient, progress=None, task_id=None) → int

Delete all FeedbackChangeDetail records associated with a specific account.

Since FeedbackChangeDetail doesn’t have a direct accountId field, we’ll need to: 1. Get all FeedbackItems for the account 2. For each FeedbackItem, delete all its FeedbackChangeDetails

Args:: account_id: The ID of the account to delete records for. client: The API client to use. progress: Optional progress bar instance to use for tracking progress. task_id: Optional task ID within the progress bar to update.
Returns:: int: The number of records deleted.

editCommentValue: str | None = None

externalId: str | None = None

feedbackItem: FeedbackItem | None = None

feedbackItemId: str | None = None

finalAnswerValue: str | None = None

finalCommentValue: str | None = None

classmethod from_dict(data: Dict[str, Any], client: PlexusDashboardClient | None = None) → FeedbackChangeDetail

Create a FeedbackChangeDetail instance from a dictionary.

Args:: data: Dictionary containing the FeedbackChangeDetail data. client: Optional API client to associate with the instance.
Returns:: A new FeedbackChangeDetail instance.

classmethod get(id: str, client: PlexusDashboardClient, fields: List[str] | None = None) → FeedbackChangeDetail | None: Retrieve a specific FeedbackChangeDetail by its ID.

id: str | None = None

initialAnswerValue: str | None = None

initialCommentValue: str | None = None

classmethod list(client: PlexusDashboardClient, feedback_item_id: str | None = None, limit: int = 100, next_token: str | None = None, fields: List[str] | None = None): List FeedbackChangeDetails, typically filtered by feedbackItemId.

updatedAt: datetime | None = None

Bases: BaseModel

Represents aggregated feedback results for a specific form/question combination. Corresponds to the FeedbackItem model in the GraphQL schema.

GRAPHQL_BASE_FIELDS = ['id', 'accountId', 'scorecardId', 'cacheKey', 'scoreId', 'itemId', 'initialAnswerValue', 'finalAnswerValue', 'initialCommentValue', 'finalCommentValue', 'editCommentValue', 'editedAt', 'editorName', 'isAgreement', 'createdAt', 'updatedAt']

GRAPHQL_RELATIONSHIP_FIELDS = {}

__init__(id: str | None = None, accountId: str | None = None, scorecardId: str | None = None, cacheKey: str | None = None, scoreId: str | None = None, itemId: str | None = None, initialAnswerValue: str | None = None, finalAnswerValue: str | None = None, initialCommentValue: str | None = None, finalCommentValue: str | None = None, editCommentValue: str | None = None, editedAt: datetime | None = None, editorName: str | None = None, isAgreement: bool | None = None, createdAt: datetime | None = None, updatedAt: datetime | None = None, account: Account | None = None, scorecard: Scorecard | None = None, score: Score | None = None, item: Item | None = None, scoreResults: Any | None = None, _client: PlexusDashboardClient | None = None, _raw_data: Dict[str, Any] | None = None) → None

account: Account | None = None

accountId: str | None = None

cacheKey: str | None = None

classmethod count_by_account_id(account_id: str, client: PlexusDashboardClient) → int

Count the number of FeedbackItem records for a specific account.

Args:: account_id: The ID of the account to count records for. client: The API client to use.
Returns:: int: The number of FeedbackItem records for the account.

classmethod create(client: PlexusDashboardClient, data: Dict[str, Any]) → FeedbackItem | None: Create a new FeedbackItem.

createdAt: datetime | None = None

classmethod delete_all_by_account_id(account_id: str, client: PlexusDashboardClient, progress=None, task_id=None) → int

Delete all FeedbackItem records for a specific account.

Args:: account_id: The ID of the account to delete records for. client: The API client to use. progress: Optional progress bar instance to use for tracking progress. task_id: Optional task ID within the progress bar to update.
Returns:: int: The number of records deleted.

editCommentValue: str | None = None

editedAt: datetime | None = None

editorName: str | None = None

finalAnswerValue: str | None = None

finalCommentValue: str | None = None

classmethod from_dict(data: Dict[str, Any], client: PlexusDashboardClient | None = None) → FeedbackItem

Create a FeedbackItem instance from a dictionary.

Args:: data: Dictionary containing the FeedbackItem data. client: Optional API client to associate with the instance.
Returns:: A new FeedbackItem instance.

classmethod generate_cache_key(score_id: str, form_id: str) → str

Generate a cache key for a feedback item based on score ID and form ID.

Args:: score_id: The score ID form_id: The form ID
Returns:: A string that can be used as a cache key

classmethod get(id: str, client: PlexusDashboardClient, fields: List[str] | None = None, relationship_fields: Dict[str, List[str]] | None = None) → FeedbackItem | None: Retrieve a specific FeedbackItem by its ID.

classmethod get_by_composite_key(client: PlexusDashboardClient, account_id: str, scorecard_id: str, score_id: str, cache_key: str) → FeedbackItem | None

Efficiently retrieve a FeedbackItem using the combination of account, scorecard, score, and cache key.

Args:: client: The API client account_id: Account ID scorecard_id: Scorecard ID score_id: Score ID cache_key: Cache key for this feedback item
Returns:: FeedbackItem or None if not found

id: str | None = None

initialAnswerValue: str | None = None

initialCommentValue: str | None = None

isAgreement: bool | None = None

item: Item | None = None

itemId: str | None = None

List FeedbackItems with optional filtering.

Args:: client: The API client to use account_id: Optional account ID to filter by scorecard_id: Optional scorecard ID to filter by score_id: Optional score ID to filter by cache_key: Optional cache key to filter by limit: Maximum number of items to return next_token: Pagination token from a previous request fields: Optional list of fields to include in the response relationship_fields: Optional relationship fields to include filter: Optional additional filter parameters index_name: Optional GSI name to use (e.g., “byAccountScorecardScoreUpdatedAt”) sort_condition: Optional sort condition for the GSI query, particularly for date range filtering
Returns:: Tuple of (list of FeedbackItem objects, next pagination token)

load_change_details(force_reload: bool = False): Loads the related FeedbackChangeDetail records.

score: Score | None = None

scoreId: str | None = None

scoreResults: Any | None = None

scorecard: Scorecard | None = None

scorecardId: str | None = None

updatedAt: datetime | None = None

classmethod upsert_by_cache_key(client: PlexusDashboardClient, account_id: str, scorecard_id: str, score_id: str, cache_key: str, initial_answer_value: str | None = None, final_answer_value: str | None = None, initial_comment_value: str | None = None, final_comment_value: str | None = None, edit_comment_value: str | None = None, is_agreement: bool | None = None, edited_at: str | None = None, editor_name: str | None = None, item_id: str | None = None, debug: bool = False) → Tuple[str, bool, str | None]

Create or update a FeedbackItem using cache key lookup for deduplication.

Uses the byCacheKey GSI for efficient lookup to prevent duplicates.

Args:: client: The PlexusDashboardClient instance account_id: Account ID scorecard_id: Scorecard ID score_id: Score ID cache_key: Cache key for lookup (typically “{score_id}:{form_id}”) initial_answer_value: Initial answer value final_answer_value: Final answer value initial_comment_value: Initial comment value final_comment_value: Final comment value edit_comment_value: Edit comment value is_agreement: Whether initial and final answers agree edited_at: ISO timestamp when edited editor_name: Name of editor item_id: Associated Item ID debug: Enable debug logging
Returns:: Tuple of (feedback_item_id, was_created, error_message) - feedback_item_id: ID of the FeedbackItem (None if error) - was_created: True if created, False if updated - error_message: Error description (None if successful)

class plexus.dashboard.api.models.Identifier(itemId: str, name: str, value: str, accountId: str, createdAt: datetime.datetime, updatedAt: datetime.datetime, url: str | None = None, position: int | None = None, client: ForwardRef('PlexusDashboardClient') | None = None)

Bases: BaseModel

__init__(itemId: str, name: str, value: str, accountId: str, createdAt: datetime, updatedAt: datetime, url: str | None = None, position: int | None = None, client: PlexusDashboardClient | None = None)

accountId: str

classmethod batch_create_for_item(client: PlexusDashboardClient, item_id: str, account_id: str, identifiers_data: List[Dict[str, Any]]) → List[Identifier]

Create multiple identifiers for an item in a single operation.

Args:: client: API client item_id: ID of the item these identifiers belong to account_id: Account ID for data isolation identifiers_data: List of identifier data dicts with ‘name’, ‘value’, and optional ‘url’
Returns:: List of created Identifier objects

classmethod batch_find_by_values(values: List[str], account_id: str, client: PlexusDashboardClient) → Dict[str, Identifier]

Batch exact-match lookup for multiple identifier values.

Returns a dictionary mapping value -> Identifier for found identifiers. Missing values will not appear in the result dictionary.

Args:: values: List of identifier values to search for account_id: Account to search within client: API client
Returns:: Dict mapping found values to their Identifier objects

classmethod create(client: PlexusDashboardClient, itemId: str, name: str, value: str, accountId: str, url: str | None = None, position: int | None = None) → Identifier: Create a new identifier record.

createdAt: datetime

delete() → bool: Delete this identifier.

classmethod fields() → str: Return the GraphQL fields to query for this model

classmethod find_by_name_and_value(name: str, value: str, account_id: str, client: PlexusDashboardClient) → Identifier | None

Find an identifier by name and exact value within an account.

Uses regular listIdentifiers with compound filter for compatibility.

classmethod find_by_value(value: str, account_id: str, client: PlexusDashboardClient) → Identifier | None

Find an identifier by exact value within an account.

Uses the byAccountAndValue GSI for O(1) performance. Returns the first matching identifier or None if not found.

classmethod from_dict(data: Dict[str, Any], client: PlexusDashboardClient) → Identifier: Create an Identifier instance from API response data.

itemId: str

classmethod list_by_item_id(item_id: str, client: PlexusDashboardClient) → List[Identifier]

Get all identifiers for a specific item.

Uses the byItemId GSI for efficient lookup.

name: str

position: int | None = None

update(**kwargs) → Identifier: Update this identifier with new values.

updatedAt: datetime

url: str | None = None

value: str

class plexus.dashboard.api.models.Item(id: str, evaluationId: str, createdAt: datetime.datetime, updatedAt: datetime.datetime, accountId: str, isEvaluation: bool, createdByType: str | None = None, text: str | None = None, metadata: Dict | None = None, identifiers: Dict | None = None, externalId: str | None = None, description: str | None = None, scoreId: str | None = None, attachedFiles: list | None = None, client: ForwardRef('_BaseAPIClient') | None = None)

Bases: BaseModel

__init__(id: str, evaluationId: str, createdAt: datetime, updatedAt: datetime, accountId: str, isEvaluation: bool, createdByType: str | None = None, text: str | None = None, metadata: Dict | None = None, identifiers: Dict | None = None, externalId: str | None = None, description: str | None = None, scoreId: str | None = None, attachedFiles: list | None = None, client: _BaseAPIClient | None = None)

accountId: str

attachedFiles: list | None = None

classmethod create(client: _BaseAPIClient, evaluationId: str, text: str | None = None, metadata: Dict | None = None, createdByType: str | None = None, deterministic_id: str | None = None, **kwargs) → Item

createdAt: datetime

createdByType: str | None = None

description: str | None = None

evaluationId: str

externalId: str | None = None

classmethod fields() → str: Return the GraphQL fields to query for this model

classmethod find_by_identifier(client: _BaseAPIClient, account_id: str, identifier_key: str, identifier_value: str, debug: bool = False) → Item | None

Find an Item using a specific identifier (reportId, formId, sessionId, etc.).

This method provides a general-purpose way to find Items by any identifier type, not limited to cache-specific use cases.

Args:

client: API client for database operations account_id: Account ID to scope the search identifier_key: The type of identifier (e.g., ‘reportId’, ‘formId’, ‘sessionId’) identifier_value: The value to search for debug: Enable debug logging

Returns:

Item object if found, None otherwise

Example:

# Find item by report ID item = Item.find_by_identifier(client, account_id, ‘reportId’, ‘277307013’)

# Find item by form ID item = Item.find_by_identifier(client, account_id, ‘formId’, ‘12345’)

classmethod from_dict(data: Dict[str, Any], client: _BaseAPIClient) → Item: Create an instance from a dictionary of data

classmethod get_by_id(item_id: str, client: _BaseAPIClient) → Item | None

Get a single item by its ID.

Args:: item_id: The ID of the item to retrieve client: The API client to use
Returns:: Item object if found, None otherwise

identifiers: Dict | None = None

isEvaluation: bool

classmethod list(client: _BaseAPIClient, filter: Dict[str, Any] | None = None, sort: Dict[str, str] | None = None, limit: int | None = None, next_token: str | None = None) → List[Item]

List items with optional filtering and sorting.

Args:: client: The API client to use filter: Filter conditions (e.g., {‘accountId’: {‘eq’: ‘account123’}}) sort: Sort configuration (e.g., {‘createdAt’: ‘DESC’}) limit: Maximum number of items to return next_token: Pagination token
Returns:: List of Item objects

metadata: Dict | None = None

scoreId: str | None = None

text: str | None = None

update(**kwargs) → Item

updatedAt: datetime

classmethod upsert_by_identifiers(client: _BaseAPIClient, account_id: str, identifiers: Dict[str, Any], external_id: str | None = None, description: str | None = None, text: str | None = None, metadata: Dict | None = None, evaluation_id: str | None = None, is_evaluation: bool = False, score_id: str | None = None, debug: bool = False) → Tuple[str, bool, str | None]

Upsert an Item using identifier-based lookup to prevent duplicates.

This method provides the core functionality for preventing duplicate Items by using the Identifier table’s GSI for efficient lookups.

Args:

client: PlexusDashboardClient instance account_id: The Plexus account ID identifiers: Dict containing identifier values like {‘formId’: ‘12345’, ‘reportId’: ‘67890’} external_id: Optional external ID for the item description: Optional item description text: Optional text content metadata: Optional metadata dict evaluation_id: Optional evaluation ID (defaults to ‘prediction-default’ for non-evaluation items) is_evaluation: Whether this is an evaluation item debug: Enable debug logging

Returns:

Tuple of (item_id, was_created, error_msg) - item_id: The ID of the created/updated Item, or None if failed - was_created: True if item was created, False if updated - error_msg: Error message if operation failed, None if successful

Example:

identifiers = {‘formId’: ‘12345’, ‘reportId’: ‘67890’} item_id, was_created, error = Item.upsert_by_identifiers(

client=client, account_id=’account-123’, identifiers=identifiers, external_id=’report-67890’, description=’Call transcript’, text=’Full transcript text…’

)

class plexus.dashboard.api.models.Report(id: str, reportConfigurationId: str, accountId: str, name: str, taskId: str, createdAt: datetime.datetime, updatedAt: datetime.datetime, parameters: Dict[str, Any] | None = None, output: str | None = None, client: ForwardRef('_BaseAPIClient') | None = None)

Bases: BaseModel

__init__(id: str, reportConfigurationId: str, accountId: str, name: str, taskId: str, createdAt: datetime, updatedAt: datetime, parameters: Dict[str, Any] | None = None, output: str | None = None, client: _BaseAPIClient | None = None)

accountId: str

classmethod create(client: _BaseAPIClient, reportConfigurationId: str, accountId: str, taskId: str, name: str, parameters: Dict[str, Any] | None = None, output: str | None = None) → Report: Create a new Report record via GraphQL mutation, linked to a Task.

createdAt: datetime

delete() → bool

Delete this Report record and its associated ReportBlock records.

Performs a GraphQL mutation to delete the report. The API is expected to cascade delete all associated ReportBlock records.

Returns:: bool: True if deletion was successful, False otherwise.
Raises:: ValueError: If no client is associated with this instance. Exception: If the GraphQL mutation returns an error.

classmethod delete_by_id(report_id: str, client: _BaseAPIClient) → bool

Delete a Report by ID without first retrieving it.

Args:: report_id: The ID of the report to delete. client: The API client to use for the deletion.
Returns:: bool: True if deletion was successful, False otherwise.
Raises:: Exception: If the GraphQL mutation returns an error.

classmethod delete_multiple(report_ids: List[str], client: _BaseAPIClient) → Dict[str, bool]

Delete multiple Reports by IDs.

Args:: report_ids: A list of report IDs to delete. client: The API client to use for the deletion.
Returns:: Dict[str, bool]: A dictionary mapping report IDs to deletion success status.

classmethod fields() → str: Return the GraphQL fields to query for this model.

classmethod from_dict(data: Dict[str, Any], client: _BaseAPIClient) → Report: Create an instance from a dictionary of data.

classmethod get_by_name(name: str, account_id: str, client: _BaseAPIClient) → Report | None

Get a Report by its name within a specific account.

Uses the ‘listReportByAccountIdAndUpdatedAt’ index and filters results client-side to find the matching name. Returns the first match found. NOTE: This can be inefficient if there are many reports for the account.

classmethod list_by_account_id(account_id: str, client: _BaseAPIClient, limit: int = 100, max_items: int | None = None) → list[Report]

List Reports for a specific account using pagination.

Uses the ‘listReportByAccountIdAndUpdatedAt’ index.

Args:: account_id: The ID of the account to list reports for. client: The API client instance. limit: The number of items to fetch per page. max_items: Optional limit on the total number of items to return.
Returns:: A list of Report instances.

name: str

output: str | None = None

parameters: Dict[str, Any] | None

reportConfigurationId: str

taskId: str

update(**kwargs) → Report: Update an existing Report record via GraphQL mutation.

updatedAt: datetime

class plexus.dashboard.api.models.ReportBlock(reportId: str, position: int, type: str, output: Optional[Dict[str, Any]] = <factory>, name: Optional[str] = None, log: Optional[str] = None, attachedFiles: Optional[str] = None, dataSetId: Optional[str] = None, createdAt: Optional[datetime.datetime] = None, updatedAt: Optional[datetime.datetime] = None)

Bases: BaseModel

__init__(reportId: str, position: int, type: str, output: ~typing.Dict[str, ~typing.Any] | None = <factory>, name: str | None = None, log: str | None = None, attachedFiles: str | None = None, dataSetId: str | None = None, createdAt: ~datetime.datetime | None = None, updatedAt: ~datetime.datetime | None = None) → None

attachedFiles: str | None = None

classmethod create(client: _BaseAPIClient, reportId: str, position: int, type: str, output: str | None = None, name: str | None = None, log: str | None = None, attachedFiles: str | None = None, dataSetId: str | None = None) → ReportBlock: Create a new ReportBlock record via GraphQL mutation.

createdAt: datetime | None = None

dataSetId: str | None = None

classmethod fields() → str: Return the GraphQL fields to query for this model.

classmethod from_dict(data: Dict[str, Any], client: _BaseAPIClient) → ReportBlock: Create an instance from a dictionary of data (typically from API response).

classmethod list_by_report_id(report_id: str, client: _BaseAPIClient, limit: int = 100, max_items: int | None = None) → list[ReportBlock]

List ReportBlocks for a specific report using pagination.

Assumes a GraphQL query ‘listReportBlocksByReportIdAndPosition’ exists, using an index like ‘byReportIdAndPosition’.

Args:: report_id: The ID of the Report to list blocks for. client: The API client instance. limit: The number of items to fetch per page. max_items: Optional limit on the total number of items to return.
Returns:: A list of ReportBlock instances, sorted by position ascending.

log: str | None = None

name: str | None = None

output: Dict[str, Any] | None

position: int

reportId: str

type: str

update(client: _BaseAPIClient | None = None, **kwargs) → ReportBlock: Update an existing ReportBlock record via GraphQL mutation.

updatedAt: datetime | None = None

class plexus.dashboard.api.models.ReportConfiguration(id: str, name: str, accountId: str, configuration: str, description: str | None = None, createdAt: datetime.datetime | None = None, updatedAt: datetime.datetime | None = None, client: ForwardRef('_BaseAPIClient') | None = None)

Bases: BaseModel

__init__(id: str, name: str, accountId: str, configuration: str, description: str | None = None, createdAt: datetime | None = None, updatedAt: datetime | None = None, client: _BaseAPIClient | None = None)

accountId: str

configuration: str

classmethod create(client: _BaseAPIClient, name: str, accountId: str, configuration: str, description: str | None = None) → ReportConfiguration: Create a new ReportConfiguration record via GraphQL mutation.

createdAt: datetime | None = None

description: str | None = None

classmethod fields() → str: Return the GraphQL fields to query for this model.

classmethod from_dict(data: Dict[str, Any], client: _BaseAPIClient) → ReportConfiguration: Create an instance from a dictionary of data.

classmethod get_by_name(name: str, account_id: str, client: _BaseAPIClient) → ReportConfiguration | None

Get a ReportConfiguration by its name within a specific account.

Uses the GSI ‘byAccountIdAndName’. Returns the first match found.

classmethod list_by_account_and_name(account_id: str, name: str, client: _BaseAPIClient, limit: int = 50, next_token: str | None = None) → Dict[str, Any]: List Report Configurations for a specific account matching a name using the GSI.

classmethod list_by_account_id(account_id: str, client: _BaseAPIClient, limit: int = 50, next_token: str | None = None) → Dict[str, Any]: List Report Configurations for a specific account using the GSI.

name: str

update(client: _BaseAPIClient, name: str, accountId: str, configuration: str, description: str | None = None) → ReportConfiguration: Update the ReportConfiguration record.

updatedAt: datetime | None = None

class plexus.dashboard.api.models.Score(id: str, name: str, key: str, externalId: str, type: str, order: int, sectionId: str, accuracy: float | None = None, version: str | None = None, aiProvider: str | None = None, aiModel: str | None = None, client: ForwardRef('_BaseAPIClient') | None = None)

Bases: BaseModel

__init__(id: str, name: str, key: str, externalId: str, type: str, order: int, sectionId: str, accuracy: float | None = None, version: str | None = None, aiProvider: str | None = None, aiModel: str | None = None, client: _BaseAPIClient | None = None)

accuracy: float | None = None

aiModel: str | None = None

aiProvider: str | None = None

create_version_from_code(code_content: str, note: str | None = None, guidelines: str | None = None) → Dict[str, Any]

Create a new score version from code and guidelines content.

This is the foundational method that only creates a new version if the content has changed from the current champion version.

Args:

code_content: The YAML code content as a string note: Optional version note. Defaults to “Updated via Score.create_version_from_code()” guidelines: Optional guidelines content as a string

Returns:

Dict containing:

success: bool
version_id: str (new version ID if created, existing if no changes)
champion_updated: bool (whether champion version was updated)
message: str (success/error message)
skipped: bool (true if no changes detected)

create_version_from_yaml(yaml_content: str, note: str | None = None) → Dict[str, Any]

Create a new score version from YAML content string.

Args:

yaml_content: The YAML configuration content as a string note: Optional version note.

Returns:

Dict containing:

success: bool
version_id: str (new version ID if created, existing if no changes)
champion_updated: bool (whether champion version was updated)
skipped: bool (true if no changes detected)
error: str (error message if failed)

externalId: str

classmethod fields() → str: Return the GraphQL fields to query for this model

classmethod from_dict(data: Dict[str, Any], client: _BaseAPIClient) → Score: Create an instance from a dictionary of data

classmethod get_by_external_id(external_id: str, scorecard_id: str, client: _BaseAPIClient) → Score | None: Get a score by its external ID

classmethod get_by_id(id: str, client: _BaseAPIClient) → Score

classmethod get_by_key(key: str, scorecard_id: str, client: _BaseAPIClient) → Score | None: Get a score by its key

classmethod get_by_name(name: str, scorecard_id: str, client: _BaseAPIClient) → Score | None: Get a score by its name

classmethod get_by_scorecard_and_external_id(scorecard_id: str, external_id: str, client: _BaseAPIClient) → Dict[str, Any] | None

Get a score by scorecard ID and external ID using the GSI.

This uses the byScorecardIdAndExternalId GSI for efficient lookup.

Args:: scorecard_id: The scorecard ID to filter by external_id: The score external ID to look up client: The API client instance
Returns:: Dict with ‘id’ and ‘name’ keys if found, None otherwise

get_champion_code_yaml() → str | None

Get the raw YAML code string from the champion version.

Returns:: Optional[str]: The raw YAML code string, or None if not found

get_champion_configuration_yaml() → str | None

Get the raw YAML configuration string from the champion version.

Returns:: Optional[str]: The raw YAML configuration string, or None if not found

get_code() → Dict[str, Any] | None

Get the score’s code from its champion version.

Returns:: Optional[Dict[str, Any]]: The parsed YAML code, or None if not found

get_local_code_path(scorecard_name: str | None = None) → Path

Get the local YAML file path for this score’s code.

Args:: scorecard_name: Optional scorecard name. If not provided, will lookup via API.
Returns:: Path: Path to the local YAML code file

get_local_configuration_path(scorecard_name: str | None = None) → Path

Get the local YAML file path for this score’s configuration.

Args:: scorecard_name: Optional scorecard name. If not provided, will lookup via API.
Returns:: Path: Path to the local YAML configuration file

get_local_guidelines_path(scorecard_name: str | None = None) → Path

Get the local file path for this score’s guidelines.

Args:: scorecard_name: Optional scorecard name. If not provided, will lookup via API.
Returns:: Path: Path to the local guidelines file

get_valid_classes() → List[Dict[str, Any]]

Get the list of valid classes from the score’s code.

Looks for a ‘classes’ key in the score’s YAML configuration, where each class has at least a ‘name’ field and may have additional metadata like ‘positive’.

Example YAML format: ```yaml classes:

name: Yes positive: true

name: No

```

Returns:

List[Dict[str, Any]]: List of valid class dictionaries, each containing at least a ‘name’ key: and any additional metadata from the configuration. Returns empty list if not found or on error.

get_valid_classes_count() → int

Get the number of valid classes from the score’s configuration.

This method looks at: 1. The explicit valid_classes list in the configuration 2. Any output values specified in conditions that might add additional classes

Returns:: int: Number of valid classes, defaulting to 2 (binary classification) if not found.

key: str

classmethod list_by_section_id(section_id: str, client: _BaseAPIClient, next_token: str | None = None, limit: int = 100) → Dict[str, Any]

Get all scores for a section with pagination support

Returns:

Dict containing:

items: List of Score objects
nextToken: Token for next page if more results exist

name: str

order: int

pull_code_and_guidelines(scorecard_name: str | None = None) → Dict[str, Any]

Pull the champion version code and guidelines to local files.

Args:

scorecard_name: Optional scorecard name. If not provided, will lookup via API.

Returns:

Dict containing:

success: bool
code_file_path: str (path where code was saved)
guidelines_file_path: str (path where guidelines were saved)
version_id: str (the champion version ID that was pulled)
message: str (success/error message)

pull_configuration(scorecard_name: str | None = None) → Dict[str, Any]

Pull the champion version configuration to local file.

Args:

scorecard_name: Optional scorecard name. If not provided, will lookup via API.

Returns:

Dict containing:

success: bool
file_path: str (path where configuration was saved)
version_id: str (the champion version ID that was pulled)
error: str (error message if failed)

push_code_and_guidelines(scorecard_name: str | None = None, note: str | None = None) → Dict[str, Any]

Push local code and guidelines files as a new score version.

Automatically detects which files have changed by comparing with the current champion version. Only creates a new version if either code or guidelines have changed.

Args:

scorecard_name: Optional scorecard name. If not provided, will lookup via API. note: Optional version note. Defaults to “Updated via Score.push_code_and_guidelines()”

Returns:

Dict containing:

success: bool
version_id: str (new version ID if created, existing if no changes)
champion_updated: bool (whether champion version was updated)
message: str (success/error message)
skipped: bool (true if no changes detected)
changes_detected: dict with ‘code’ and ‘guidelines’ booleans indicating what changed

push_configuration(scorecard_name: str | None = None, note: str | None = None) → Dict[str, Any]

Push local configuration file as a new score version.

Args:

scorecard_name: Optional scorecard name. If not provided, will lookup via API. note: Optional version note.

Returns:

Dict containing:

success: bool
version_id: str (new version ID if created, existing if no changes)
champion_updated: bool (whether champion version was updated)
skipped: bool (true if no changes detected)
error: str (error message if failed)

sectionId: str

type: str

version: str | None = None

Bases: BaseModel

Represents a single classification or scoring result in the Plexus dashboard.

ScoreResult is the core data structure for tracking individual scoring operations, used both for real-time scoring and evaluation. It integrates with the GraphQL API to provide:

Score value tracking with confidence
Metadata storage for debugging and analysis
Links to related entities (items, accounts, scorecards)
Evaluation result tracking
Batch processing support

Common usage patterns: 1. Creating a single score result:

result = ScoreResult.create(
client=client, value=0.95, itemId=”item-123”, accountId=”acc-456”, scorecardId=”card-789”, metadata={“source”: “phone_call”}

)

Batch creation for efficiency:

results = ScoreResult.batch_create(client, [

{
“value”: 0.95, “itemId”: “item-123”, “accountId”: “acc-456”, “scorecardId”: “card-789”

}, {

“value”: 0.82, “itemId”: “item-124”, “accountId”: “acc-456”, “scorecardId”: “card-789”

}

])
Updating with evaluation results:

result.update(
correct=True, evaluationId=”eval-123”, metadata={“human_label”: “Yes”}

)

ScoreResult is commonly used with: - Evaluation for accuracy testing - ScoringJob for batch processing - LangGraphScore for LLM-based classification

__init__(id: str, value: str, itemId: str, accountId: str, scorecardId: str, scoreId: str | None = None, scoreVersionId: str | None = None, score: Dict[str, Any] | None = None, explanation: str | None = None, confidence: float | None = None, metadata: Dict | None = None, trace: Dict | None = None, correct: bool | None = None, cost: Dict | None = None, attachments: List[str] | None = None, scoringJobId: str | None = None, evaluationId: str | None = None, feedbackItemId: str | None = None, code: str | None = '200', type: str | None = 'prediction', status: str | None = 'COMPLETED', createdAt: datetime | None = None, updatedAt: datetime | None = None, client: _BaseAPIClient | None = None)

accountId: str

attachments: List[str] | None = None

classmethod batch_create(client: _BaseAPIClient, items: List[Dict]) → List[ScoreResult]: Create multiple score results in a single API request.

code: str | None = '200'

confidence: float | None = None

correct: bool | None = None

cost: Dict | None = None

classmethod create(client: _BaseAPIClient, value: float, itemId: str, accountId: str, scorecardId: str, scoringJobId: str | None = None, evaluationId: str | None = None, code: str = '200', type: str = 'prediction', status: str = 'COMPLETED', **kwargs) → ScoreResult

Create a new score result.

Args:

client: The API client instance value: Score value (required) itemId: ID of scored item (required) accountId: Account context (required) scorecardId: ID of scorecard used (required) scoringJobId: ID of scoring job (optional) evaluationId: ID of evaluation (optional) code: HTTP response code (defaults to ‘200’ for success) type: Type of score result - “prediction”, “evaluation”, etc. (defaults to ‘prediction’) status: Status of score result (defaults to ‘COMPLETED’) **kwargs: Optional fields:

confidence: float

metadata: dict (will be JSON serialized)

trace: dict (will be JSON serialized)

cost: dict (will be JSON serialized)

correct: bool

attachments: List[str]

Note:

Either scoringJobId or evaluationId should be provided, but not required

createdAt: datetime | None = None

evaluationId: str | None = None

explanation: str | None = None

feedbackItemId: str | None = None

classmethod fields() → str: Return the GraphQL fields to query for this model

classmethod find_by_cache_key(client: _BaseAPIClient, item_id: str, type: str, score_id: str, account_id: str | None = None) → ScoreResult | None

Find the most recent ScoreResult using cache key components.

This method encapsulates the cache lookup logic for finding existing ScoreResults based on the standard cache key: itemId + type + scoreId. It uses the time-based GSI to return the most recent result by updatedAt.

Args:

client: API client for database operations item_id: Item ID (should be resolved DynamoDB ID) type: Type of score result (should be “prediction” or “evaluation”) score_id: Score ID (should be resolved DynamoDB ID) account_id: Optional account ID for additional context/validation

Returns:

Most recent ScoreResult matching the cache key, or None if not found

Example:

# Find cached score result cached_result = ScoreResult.find_by_cache_key(

client=client, item_id=”da270073-83ab-4c43-a1e6-961851c13d92”, type=”prediction”, score_id=”687361f7-44a9-466f-8920-7f9dc351bcd2”

)

classmethod from_dict(data: Dict[str, Any], client: _BaseAPIClient) → ScoreResult: Create a ScoreResult instance from API response data.

classmethod get_by_id(score_result_id: str, client: _BaseAPIClient) → ScoreResult | None

Get a ScoreResult by its ID.

Args:: score_result_id: The ID of the score result to retrieve client: The API client to use
Returns:: ScoreResult object if found, None otherwise

itemId: str

metadata: Dict | None = None

property numeric_value: float | None: Get the numeric value as a float, or None if invalid.

score: Dict[str, Any] | None

scoreId: str | None = None

property scoreName: str | None: Get the score name from the related score object.

scoreVersionId: str | None = None

scorecardId: str

scoringJobId: str | None = None

status: str | None = 'COMPLETED'

trace: Dict | None = None

type: str | None = 'prediction'

update(**kwargs) → ScoreResult: Update this score result with new values.

updatedAt: datetime | None = None

value: str

class plexus.dashboard.api.models.Scorecard(id: str, name: str, key: str, externalId: str, accountId: str, description: str | None = None, client: ForwardRef('_BaseAPIClient') | None = None)

Bases: BaseModel

__init__(id: str, name: str, key: str, externalId: str, accountId: str, description: str | None = None, client: _BaseAPIClient | None = None)

accountId: str

classmethod create(client: _BaseAPIClient, name: str, key: str, externalId: str, accountId: str, description: str | None = None) → Scorecard

Create a new scorecard.

Args:: client: The API client name: Name of the scorecard key: Unique key identifier externalId: External ID (usually from YAML) accountId: ID of the account this scorecard belongs to description: Optional description
Returns:: The created Scorecard instance

description: str | None = None

externalId: str

classmethod fields() → str: Return the GraphQL fields to query for this model

classmethod from_dict(data: Dict[str, Any], client: _BaseAPIClient) → Scorecard: Create an instance from a dictionary of data

classmethod get_by_account_and_external_id(account_id: str, external_id: str, client: _BaseAPIClient) → Scorecard | None

Get a scorecard by account ID and external ID using the GSI.

This uses the byAccountIdAndExternalId GSI for efficient lookup.

Args:: account_id: The account ID to filter by external_id: The scorecard external ID to look up client: The API client instance
Returns:: Scorecard: The found scorecard or None

classmethod get_by_external_id(external_id: str, client: _BaseAPIClient) → Scorecard

classmethod get_by_id(id: str, client: _BaseAPIClient) → Scorecard

classmethod get_by_key(key: str, client: _BaseAPIClient) → Scorecard

classmethod get_by_name(name: str, client: _BaseAPIClient) → Scorecard

key: str

classmethod list_by_external_id(external_id: str, client: _BaseAPIClient) → Scorecard | None

Get a scorecard by its external ID.

Args:: external_id: The scorecard external ID to look up client: The API client instance
Returns:: Scorecard: The found scorecard or None

classmethod list_by_key(key: str, client: _BaseAPIClient) → Scorecard | None

Get a scorecard by its key.

Args:: key: The scorecard key to look up client: The API client instance
Returns:: Scorecard: The found scorecard or None

classmethod list_by_name(name: str, client: _BaseAPIClient) → Scorecard | None

Get a scorecard by its name.

Args:: name: The scorecard name to look up client: The API client instance
Returns:: Scorecard: The found scorecard or None

name: str

class plexus.dashboard.api.models.ScoringJob(id: str, accountId: str, status: str, createdAt: datetime, updatedAt: datetime, scorecardId: str, itemId: str, startedAt: datetime | None = None, completedAt: datetime | None = None, errorMessage: str | None = None, errorDetails: Dict | None = None, metadata: Dict | None = None, evaluationId: str | None = None, scoreId: str | None = None, batchId: str | None = None, client: _BaseAPIClient | None = None)

Bases: BaseModel

Represents an individual scoring operation job.

A ScoringJob tracks the execution of a single scoring operation on an item, managing its lifecycle from creation through completion or failure.

Parameters:

accountId – Identifier for the account owning this job
status – Current status of the scoring job
createdAt – Timestamp when the job was created
updatedAt – Timestamp of the last update
scorecardId – Associated scorecard identifier
itemId – Identifier of the item being scored
startedAt – Timestamp when scoring started
completedAt – Timestamp when scoring completed
errorMessage – Error message if scoring failed
errorDetails – Detailed error information
metadata – Additional metadata about the scoring job
evaluationId – Associated evaluation identifier
scoreId – Associated score identifier
batchId – Identifier of the batch this job belongs to

__init__(id: str, accountId: str, status: str, createdAt: datetime, updatedAt: datetime, scorecardId: str, itemId: str, startedAt: datetime | None = None, completedAt: datetime | None = None, errorMessage: str | None = None, errorDetails: Dict | None = None, metadata: Dict | None = None, evaluationId: str | None = None, scoreId: str | None = None, batchId: str | None = None, client: _BaseAPIClient | None = None)

accountId: str

batchId: str | None = None

completedAt: datetime | None = None

classmethod create(client: _BaseAPIClient, accountId: str, scorecardId: str, itemId: str, parameters: Dict, **kwargs) → ScoringJob

Creates a new scoring job.

Parameters:

client – API client instance for making requests
accountId – Account identifier for the job
scorecardId – Associated scorecard identifier
itemId – Item identifier to be scored
parameters – Additional parameters for job creation
kwargs – Optional parameters (status, metadata, etc.)

Returns:

New ScoringJob instance

Raises:

Exception if creation fails

createdAt: datetime

errorDetails: Dict | None = None

errorMessage: str | None = None

evaluationId: str | None = None

classmethod fields() → str: Return the GraphQL fields to query for this model

classmethod find_by_item_id(item_id: str, client: _BaseAPIClient) → ScoringJob | None

Finds a scoring job by its associated item identifier.

Parameters:

item_id – Item identifier to search for
client – API client instance

Returns:

ScoringJob instance if found, None otherwise

classmethod find_existing_job(client: _BaseAPIClient, item_id: str, scorecard_id: str, score_id: str) → Dict[str, Any] | None

Finds an existing scoring job for the given item/score combination.

This method searches for any ScoringJob (regardless of status) that matches the specified item and score. Useful for preventing duplicate job creation and retrieving job status.

Note: scorecardId parameter is kept for API compatibility but not used in the query since scoreId is globally unique.

Parameters:

client – API client instance
item_id – Item identifier to search for
scorecard_id – Scorecard identifier (unused, kept for compatibility)
score_id – Score identifier to search for

Returns:

Dictionary with ‘id’ and ‘status’ if found, None otherwise

classmethod from_dict(data: Dict[str, Any], client: _BaseAPIClient) → ScoringJob

Creates a ScoringJob instance from dictionary data.

Parameters:

data – Dictionary containing scoring job data
client – API client instance

Returns:

ScoringJob instance

classmethod get_batch_job(scoring_job_id: str, client: _BaseAPIClient) → BatchJob | None

Gets the associated batch job for a scoring job through the join table.

Parameters:

scoring_job_id – ID of the scoring job
client – API client instance

Returns:

BatchJob instance if found, None otherwise

classmethod get_by_id(id: str, client: _BaseAPIClient) → ScoringJob

Retrieves a scoring job by its identifier.

Parameters:

id – Scoring job identifier
client – API client instance

Returns:

ScoringJob instance

Raises:

Exception if scoring job not found

itemId: str

metadata: Dict | None = None

scoreId: str | None = None

scorecardId: str

startedAt: datetime | None = None

status: str

update(**kwargs) → ScoringJob

Updates scoring job fields.

Parameters:: kwargs – Fields to update and their new values
Returns:: Updated ScoringJob instance
Raises:: ValueError if attempting to modify createdAt

updatedAt: datetime

Bases: BaseModel

__init__(id: str, accountId: str, type: str, status: str, target: str, command: str, description: str | None = None, metadata: Dict | None = None, createdAt: datetime | None = None, updatedAt: datetime | None = None, startedAt: datetime | None = None, completedAt: datetime | None = None, estimatedCompletionAt: datetime | None = None, errorMessage: str | None = None, errorDetails: Dict | None = None, stdout: str | None = None, stderr: str | None = None, output: str | None = None, error: str | None = None, attachedFiles: List[str] | None = None, currentStageId: str | None = None, workerNodeId: str | None = None, dispatchStatus: str | None = None, scorecardId: str | None = None, scoreId: str | None = None, client: _BaseAPIClient | None = None)

accountId: str

acquire_lock(timeout=1800)

advance_stage(stage: TaskStage) → None: Update the task’s current stage.

attachedFiles: List[str] | None = None

command: str

complete_processing(): Complete processing of the task.

completedAt: datetime | None = None

classmethod create(client, type: str, target: str, command: str, **kwargs) → Task

Create a new task record.

Args:: client: The API client instance type: Task type target: Task target command: Command string **kwargs: Additional task fields
Returns:: Task: The created task instance

create_stage(name: str, order: int, status: str = 'PENDING', status_message: str | None = None, total_items: int | None = None, estimated_completion_at: str | None = None, **kwargs) → TaskStage: Create a new TaskStage for this Task.

create_stages_batch(stage_configs: List[Dict[str, Any]]) → List[TaskStage]

Create multiple TaskStages in a single GraphQL mutation.

Args:

stage_configs: List of dictionaries containing stage configuration.: Each dict should have keys: name, order, status, and optional fields.

Returns:

List[TaskStage]: List of created TaskStage instances

classmethod create_with_stages(client, type: str, target: str, command: str, stages: list, **kwargs) → Task

Create a new task record with stages in a single GraphQL request.

Args:

client: The API client instance type: Task type target: Task target command: Command string stages: List of stage configurations in the format

[{“name”: str, “order”: int, “status”: str, “statusMessage”: str,
“totalItems”: int, “estimatedCompletionAt”: str}]

**kwargs: Additional task fields

Returns:

Task: The created task instance with stages initialized

createdAt: datetime | None = None

currentStageId: str | None = None

description: str | None = None

dispatchStatus: str | None = None

error: str | None = None

errorDetails: Dict | None = None

errorMessage: str | None = None

estimatedCompletionAt: datetime | None = None

fail_current_stage(error_message: str, error_details: Dict | None = None, current_items: int | None = None) → None

Mark the current running stage as failed with error message and progress.

Args:: error_message: The error message to display error_details: Optional dictionary of additional error details current_items: The number of items processed when the failure occurred

fail_processing(error_message: str, error_details: Dict | None = None, current_items: int | None = None) → None

Mark the task as failed with error information.

Args:: error_message: The error message to display error_details: Optional dictionary of additional error details current_items: The number of items processed when the failure occurred

classmethod fields() → str: Return the GraphQL fields to query for this model

classmethod from_dict(data: Dict[str, Any], client: _BaseAPIClient) → Task: Create an instance from a dictionary of data

classmethod get_by_id(id: str, client: _BaseAPIClient) → Task

get_stages() → List[TaskStage]: Get all stages for this task, handling pagination properly.

classmethod list_tasks(client: _BaseAPIClient, updated_at: str = '', limit: int = 100) → List[Task]

List tasks using the listTaskByUpdatedAt query for proper pagination.

Args:: client: The API client instance updated_at: The updatedAt timestamp to start from (empty string for most recent) limit: Maximum number of tasks to return per page
Returns:: List[Task]: List of Task instances

lock_expires: datetime | None = None

lock_token: str | None = None

metadata: Dict | None = None

output: str | None = None

release_lock()

scoreId: str | None = None

scorecardId: str | None = None

start_processing() → None: Mark the task as started.

startedAt: datetime | None = None

status: str

stderr: str | None = None

stdout: str | None = None

target: str

type: str

update(**kwargs) → Task

update_progress(current_items: int, total_items: int, stage_configs: Dict[str, Dict], estimated_completion_at: datetime | None = None) → List[TaskStage]: Update task progress and stage information.

updatedAt: datetime | None = None

workerNodeId: str | None = None

class plexus.dashboard.api.models.TaskStage(id: str, taskId: str, name: str, order: int, status: str, statusMessage: str | None = None, startedAt: datetime.datetime | None = None, completedAt: datetime.datetime | None = None, estimatedCompletionAt: datetime.datetime | None = None, processedItems: int | None = None, totalItems: int | None = None, client: ForwardRef('_BaseAPIClient') | None = None)

Bases: BaseModel

__init__(id: str, taskId: str, name: str, order: int, status: str, statusMessage: str | None = None, startedAt: datetime | None = None, completedAt: datetime | None = None, estimatedCompletionAt: datetime | None = None, processedItems: int | None = None, totalItems: int | None = None, client: _BaseAPIClient | None = None)

complete_processing() → None: Mark the stage as completed.

completedAt: datetime | None = None

classmethod create(client: _BaseAPIClient, taskId: str, name: str, order: int, status: str, statusMessage: str | None = None, **kwargs) → TaskStage

estimatedCompletionAt: datetime | None = None

fail_processing(error_message: str) → None: Mark the stage as failed with error information.

classmethod fields() → str: Return the GraphQL fields to query for this model

classmethod from_dict(data: Dict[str, Any], client: _BaseAPIClient) → TaskStage: Create an instance from a dictionary of data

classmethod get_by_id(id: str, client: _BaseAPIClient) → TaskStage

name: str

order: int

processedItems: int | None = None

start_processing() → None: Mark the stage as started and update its status.

startedAt: datetime | None = None

status: str

statusMessage: str | None = None

taskId: str

totalItems: int | None = None

update(**kwargs) → TaskStage: Update the task stage.

plexus.dashboard.api.models package

Submodules