plexus.dashboard.api.client module

GraphQL API Client - Handles authentication, API communication, and background logging.

This client provides three main functionalities: 1. GraphQL API communication with authentication 2. Fire-and-forget score logging with configurable batching 3. Background thread management for async operations

Authentication:

Uses API key authentication configured through environment variables: - PLEXUS_API_URL: The AppSync API endpoint - PLEXUS_API_KEY: The API key for authentication

Background Logging:

The client maintains a background thread for efficient score logging with these features: - Configurable batch sizes (default: 10 items) - Configurable timeouts (default: 1 second) - Immediate logging option for urgent data - Automatic batch flushing on shutdown - Error resilience (errors are logged but don’t affect main thread)

Example usage:

# Standard batched logging client.log_score(0.95, “item-123”,

accountId=”acc-123”, scoringJobId=”job-123”, scorecardId=”card-123”)

# Immediate logging (no batching) client.log_score(0.95, “item-123”,

immediate=True, accountId=”acc-123”, scoringJobId=”job-123”, scorecardId=”card-123”)

# Custom batch configuration client.log_score(0.95, “item-123”,

batch_size=50, batch_timeout=5.0, accountId=”acc-123”, scoringJobId=”job-123”, scorecardId=”card-123”)

Implementation Details:

  • Uses a Queue for thread-safe communication

  • Maintains separate batches for different batch_size/timeout configurations

  • Daemon thread ensures cleanup on program exit

  • Graceful shutdown with flush() method

  • Automatic JSON serialization of metadata fields

Error Handling:

  • API errors are caught and logged (fire-and-forget)

  • Background thread errors don’t affect main application

  • GraphQL query errors are propagated for direct API calls

class plexus.dashboard.api.client.ClientContext(account_key: str | None = None, account_id: str | None = None, scorecard_key: str | None = None, scorecard_id: str | None = None, score_name: str | None = None, score_id: str | None = None)

Bases: object

__init__(account_key: str | None = None, account_id: str | None = None, scorecard_key: str | None = None, scorecard_id: str | None = None, score_name: str | None = None, score_id: str | None = None) → None
account_id: str | None = None
account_key: str | None = None
score_id: str | None = None
score_name: str | None = None
scorecard_id: str | None = None
scorecard_key: str | None = None
class plexus.dashboard.api.client.PlexusDashboardClient(api_url: str | None = None, api_key: str | None = None, context: ClientContext | None = None)

Bases: _BaseAPIClient

Client for the Plexus Dashboard API.

Provides access to all API resources with schema-matching namespaces:

client.ScoreResult.create(…) client.ScoreResult.batch_create(…) client.Scorecard.get_by_key(…) etc.

Features: - Background processing with configurable batching - ID resolution and caching - Thread-safe operations - Context management for accounts, scorecards, and scores

__init__(api_url: str | None = None, api_key: str | None = None, context: ClientContext | None = None)
async create_chat_message(message_data: Dict[str, Any]) → Dict[str, Any]

Create a new chat message.

async create_chat_session(session_data: Dict[str, Any]) → Dict[str, Any]

Create a new chat session.

classmethod for_account(account_key: str) → PlexusDashboardClient

Create a client initialized with account context

classmethod for_scorecard(account_key: str, scorecard_key: str, score_name: str | None = None) → PlexusDashboardClient

Create a client initialized with full scoring context

async get_chat_session(session_id: str) → Dict[str, Any] | None

Get a chat session by ID.

async list_chat_messages(session_id: str = None, experiment_id: str = None, limit: int = 50, sort_by: str = 'createdAt') → List[Dict[str, Any]]

List chat messages for a session or experiment.

async list_chat_sessions(account_id: str = None, experiment_id: str = None, limit: int = 20) → List[Dict[str, Any]]

List chat sessions for an account or experiment.

async update_chat_session(update_data: Dict[str, Any]) → Dict[str, Any]

Update an existing chat session.