plexus.metrics.calculator module

Core metrics calculation logic for items and score results. This module provides reusable functions for calculating metrics that can be used by both Lambda functions and CLI commands.

class plexus.metrics.calculator.MetricsCalculator(graphql_endpoint: str, api_key: str, cache_bucket_minutes: int = 5)

Bases: object

A utility class for calculating items and score results metrics over time periods.

Uses API key authentication to connect to the Plexus GraphQL API and provides methods to count items and score results within specific timeframes, generate time buckets, and calculate hourly rates and statistics.

Initialize the MetricsCalculator.

Args:

graphql_endpoint: The GraphQL API endpoint URL api_key: The API key for authentication cache_bucket_minutes: The width of the cache buckets in minutes.

__init__(graphql_endpoint: str, api_key: str, cache_bucket_minutes: int = 5)

Initialize the MetricsCalculator.

Args:

graphql_endpoint: The GraphQL API endpoint URL api_key: The API key for authentication cache_bucket_minutes: The width of the cache buckets in minutes.

count_items_in_timeframe(account_id: str, start_time: datetime, end_time: datetime) → int

Count items created within a specific timeframe using the GSI.

Args:

account_id: The account ID to filter by start_time: Start of the time range end_time: End of the time range

Returns:

Total count of items in the timeframe

count_score_results_in_timeframe(account_id: str, start_time: datetime, end_time: datetime) → int

Count score results updated within a specific timeframe using the GSI.

Args:

account_id: The account ID to filter by start_time: Start of the time range end_time: End of the time range

Returns:

Total count of score results in the timeframe

get_clock_aligned_buckets(start_time: datetime, end_time: datetime, bucket_size_minutes: int) → List[Tuple[datetime, datetime]]

Calculates the clock-aligned buckets that are fully contained within the given time range.

get_count_with_caching(account_id: str, start_time: datetime, end_time: datetime, count_function) → int

Get count for a time range, using clock-aligned cache buckets.

get_items_summary(account_id: str, hours: int = 24) → Dict[str, Any]

Calculate key metrics for items over the last N hours.

Args:

account_id: The account ID to filter by hours: The number of hours to look back from the current time

Returns:

A dictionary of calculated metrics for items

get_score_results_summary(account_id: str, hours: int = 24) → Dict[str, Any]

Calculate key metrics for score results over the last N hours.

Args:

account_id: The account ID to filter by hours: The number of hours to look back from the current time

Returns:

A dictionary of calculated metrics for score results

get_time_buckets(start_time: datetime, end_time: datetime, bucket_size_minutes: int) → List[Tuple[datetime, datetime]]

Generate time buckets of a specific size between a start and end time.

make_graphql_request(query: str, variables: Dict[str, Any]) → Dict[str, Any]

Make a GraphQL request to the API using API key authentication.

Args:

query: The GraphQL query string variables: Variables for the GraphQL query

Returns:

The response data from the API

Raises:

Exception: If the request fails or returns errors

class plexus.metrics.calculator.SQLiteCache(db_name='plexus-record-count-cache.db')

Bases: object

A simple SQLite-based cache for storing metric counts.

__init__(db_name='plexus-record-count-cache.db')

close()

create_table()

get(key: str) → int | None

set(key: str, value: int)

plexus.metrics.calculator.create_calculator_from_env(cache_bucket_minutes: int = 15) → MetricsCalculator

Create a MetricsCalculator instance from environment variables.

Uses standard PLEXUS_API_URL and PLEXUS_API_KEY environment variables.