plexus.reports.blocks.feedback_utils module

Shared utilities for identifying scorecards and scores with feedback items.

This module provides reusable functions for: - Fetching scorecards from the account - Fetching scores for a scorecard - Fetching feedback items for a score in a time range - Identifying which scorecards/scores have feedback in a given time range

These utilities are used by both the FeedbackAnalysis report block and the feedback evaluation CLI commands.

async plexus.reports.blocks.feedback_utils.fetch_all_scorecards(api_client, account_id: str) → List[Scorecard]

Fetches all scorecards for the given account.

Args:

api_client: The API client for making GraphQL queries account_id: The account ID to fetch scorecards for

Returns:

List of Scorecard objects

async plexus.reports.blocks.feedback_utils.fetch_feedback_items_for_score(api_client, account_id: str, scorecard_id: str, score_id: str, start_date: datetime, end_date: datetime) → List[FeedbackItem]

Fetches FeedbackItem records for a specific score on a scorecard within a date range.

Uses the optimized GSI query for efficient retrieval.

Args:

api_client: The API client for making GraphQL queries account_id: The account ID scorecard_id: The ID of the scorecard score_id: The ID of the score start_date: Start date for filtering (UTC aware) end_date: End date for filtering (UTC aware)

Returns:

List of FeedbackItem objects

async plexus.reports.blocks.feedback_utils.fetch_scores_for_scorecard(api_client, scorecard_id: str) → List[Dict[str, Any]]

Fetches all scores with externalIds for a given scorecard.

Args:

api_client: The API client for making GraphQL queries scorecard_id: The ID of the scorecard to fetch scores for

Returns:

List of score info dictionaries with structure: [

{

‘plexus_score_id’: str, ‘plexus_score_name’: str, ‘cc_question_id’: str # The externalId

}

] Sorted by section order and score order within sections.

async plexus.reports.blocks.feedback_utils.get_feedback_counts_by_score(api_client, account_id: str, scorecard_id: str, score_ids: List[str], start_date: datetime, end_date: datetime) → Dict[str, int]

Gets feedback counts for each score in a scorecard efficiently. Instead of fetching all items, we fetch just enough to know which scores have feedback.

Args:

api_client: The API client for making GraphQL queries account_id: The account ID scorecard_id: The ID of the scorecard score_ids: List of score IDs to check start_date: Start date for filtering (UTC aware) end_date: End date for filtering (UTC aware)

Returns:

Dict mapping score_id to feedback count

async plexus.reports.blocks.feedback_utils.identify_scorecards_with_feedback(api_client, account_id: str, start_date: datetime, end_date: datetime, max_concurrent: int = 10) → List[Dict[str, Any]]

Identifies all scorecards and scores that have feedback items in the given time range.

This is the main orchestration function that: 1. Fetches all scorecards for the account 2. For each scorecard, fetches its scores (in parallel) 3. For each score, checks if feedback exists in the time range (in parallel) 4. Returns only scorecards/scores with feedback

Args:

api_client: The API client for making GraphQL queries account_id: The account ID start_date: Start date for filtering (UTC aware) end_date: End date for filtering (UTC aware) max_concurrent: Maximum number of scorecards to check concurrently (default: 10)

Returns:

List of dictionaries with structure: [

{

“scorecard”: Scorecard, “scores_with_feedback”: [

{

“score_id”: str, “score_name”: str, “external_id”: str, “feedback_count”: int

}

]

}

] Only includes scorecards that have at least one score with feedback.