add insight aggregations tools #41
Open
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
WalkthroughThe changes introduce aggregation validation utilities and two new aggregation methods in the Insight service. The utilities define valid aggregation functions and provide validation logic. The new methods, Changes
Sequence Diagram(s)sequenceDiagram
participant User
participant Insight
participant Utils
participant API
User->>Insight: aggregate_events(...) or aggregate_transactions(...)
Insight->>Utils: validate_*_aggregation(aggregate)
Utils-->>Insight: Validated aggregation(s)
Insight->>API: _get("events"/"transactions", params)
API-->>Insight: Aggregated results
Insight->>Insight: Clean up response (remove chain_id)
Insight-->>User: Aggregated data
✨ Finishing Touches
🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
SupportNeed help? Create a ticket on our support page for assistance with any issues or questions. Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
CodeRabbit Configuration File (
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (1)
python/thirdweb-ai/src/thirdweb_ai/common/utils.py (1)
110-118: Improve validation logic for edge cases.The validation function handles basic alias parsing well, but could be more robust.
Consider these improvements:
def validate_aggregation(agg: str, valid_aggregations: list[str]) -> str: """Validate an aggregation function string.""" + if not agg or not agg.strip(): + raise ValueError("Aggregation function cannot be empty") + # Handle aliases like "count() as event_count" - base_agg = agg.split(" as ")[0].strip() + base_agg = agg.split(" as ")[0].strip().lower() - if base_agg not in valid_aggregations: + # Case-insensitive comparison for better usability + valid_lower = [v.lower() for v in valid_aggregations] + if base_agg not in valid_lower: raise ValueError(f"Invalid aggregation function: {base_agg}. Valid options: {valid_aggregations}") return agg
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (2)
python/thirdweb-ai/src/thirdweb_ai/common/utils.py(1 hunks)python/thirdweb-ai/src/thirdweb_ai/services/insight.py(2 hunks)
🧰 Additional context used
🪛 Pylint (3.3.7)
python/thirdweb-ai/src/thirdweb_ai/services/insight.py
[refactor] 425-425: Too many arguments (9/5)
(R0913)
[refactor] 425-425: Too many positional arguments (9/5)
(R0917)
[refactor] 502-502: Too many arguments (10/5)
(R0913)
[refactor] 502-502: Too many positional arguments (10/5)
(R0917)
[refactor] 502-502: Too many local variables (16/15)
(R0914)
[refactor] 502-502: Too many branches (13/12)
(R0912)
🔇 Additional comments (4)
python/thirdweb-ai/src/thirdweb_ai/common/utils.py (3)
83-92: LGTM! Comprehensive event aggregation functions defined.The list covers essential aggregation functions for blockchain events including counts, distinct counts, and min/max for key fields.
94-107: LGTM! Well-structured transaction aggregation functions.The transaction aggregations appropriately include value-based calculations (sum, avg, min, max) which are meaningful for transaction data.
121-128: LGTM! Clean wrapper functions.The wrapper functions provide a clean interface for specific validation types.
python/thirdweb-ai/src/thirdweb_ai/services/insight.py (1)
9-18: LGTM! Comprehensive imports for new aggregation functionality.All necessary constants and validation functions are properly imported from the utils module.
Comment on lines
+499
to
+580
| @tool( | ||
| description="Aggregate blockchain transactions with powerful grouping and aggregation options. Use this for transaction counts, volumes, sums, and other analytics grouped by address, block, or time period." | ||
| ) | ||
| def aggregate_transactions( | ||
| self, | ||
| aggregate: Annotated[ | ||
| list[str], | ||
| f"Aggregation functions to apply. Valid options: {', '.join(VALID_TRANSACTION_AGGREGATIONS)}. Can include aliases like 'count() as tx_count' or 'sum(value) as total_value'.", | ||
| ], | ||
| group_by: Annotated[ | ||
| Literal["from_address", "to_address", "block_number"] | None, | ||
| "Field to group transactions by for aggregation. Required when using aggregate functions.", | ||
| ] = None, | ||
| chain_id: Annotated[ | ||
| list[int] | int | None, | ||
| "Chain ID(s) to query (e.g., 1 for Ethereum, 137 for Polygon). Specify multiple IDs as a list for cross-chain queries.", | ||
| ] = None, | ||
| from_address: Annotated[ | ||
| str | None, | ||
| "Filter transactions sent from this address (e.g., '0x1234...'). Useful for tracking outgoing transactions from a wallet.", | ||
| ] = None, | ||
| to_address: Annotated[ | ||
| str | None, | ||
| "Filter transactions sent to this address (e.g., '0x1234...'). Useful for tracking incoming transactions to a contract or wallet.", | ||
| ] = None, | ||
| function_signature: Annotated[ | ||
| str | None, | ||
| "Function signature to filter by (e.g., 'approve(address,uint256)').", | ||
| ] = None, | ||
| value_above: Annotated[ | ||
| int | None, | ||
| "Filter for transactions with value above this amount (in wei - base blockchain units).", | ||
| ] = None, | ||
| limit: Annotated[ | ||
| int | None, | ||
| "Number of aggregated results to return (default 20, max 100).", | ||
| ] = 20, | ||
| page: Annotated[ | ||
| int | None, | ||
| "Page number for paginated results, starting from 0. 20 results are returned per page.", | ||
| ] = None, | ||
| ) -> dict[str, Any]: | ||
| # Validate aggregation functions | ||
| validated_aggregate = [validate_transaction_aggregation(agg) for agg in aggregate] | ||
|
|
||
| params: dict[str, Any] = { | ||
| "sort_by": "block_number", | ||
| "sort_order": "desc", | ||
| "decode": True, | ||
| "aggregate": validated_aggregate, | ||
| } | ||
|
|
||
| if group_by: | ||
| params["group_by"] = group_by | ||
|
|
||
| chain_ids = chain_id if chain_id is not None else self.chain_ids | ||
| if chain_ids: | ||
| params["chain"] = chain_ids | ||
| if from_address: | ||
| params["filter_from_address"] = validate_address(from_address) | ||
| if to_address: | ||
| params["filter_to_address"] = validate_address(to_address) | ||
| if function_signature: | ||
| params["filter_function_signature"] = function_signature | ||
| if value_above: | ||
| params["value_gte"] = value_above | ||
| if limit: | ||
| params["limit"] = limit | ||
| if page: | ||
| params["page"] = page | ||
|
|
||
| out = self._get("transactions", params) | ||
|
|
||
| # Clean up response by removing chain_id from aggregations if present | ||
| if out.get("aggregations"): | ||
| for agg in out["aggregations"]: | ||
| if isinstance(agg, dict): | ||
| for value in agg.values(): | ||
| if isinstance(value, dict) and "chain_id" in value: | ||
| value.pop("chain_id") | ||
|
|
||
| return out |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Extract shared logic and address complexity warnings.
This method duplicates response cleanup logic and has high complexity (flagged by static analysis for too many arguments, variables, and branches).
Apply these changes:
- Use the shared cleanup method:
out = self._get("transactions", params)
- # Clean up response by removing chain_id from aggregations if present
- if out.get("aggregations"):
- for agg in out["aggregations"]:
- if isinstance(agg, dict):
- for value in agg.values():
- if isinstance(value, dict) and "chain_id" in value:
- value.pop("chain_id")
+ return self._cleanup_aggregation_response(out)
-
- return out- Consider splitting parameter building into a helper method to reduce local variables:
def _build_transaction_params(self, validated_aggregate, group_by, chain_ids,
from_address, to_address, function_signature,
value_above, limit, page):
"""Build query parameters for transaction aggregation."""
# Move parameter building logic hereThis addresses the static analysis warnings about complexity while improving maintainability.
🧰 Tools
🪛 Pylint (3.3.7)
[refactor] 502-502: Too many arguments (10/5)
(R0913)
[refactor] 502-502: Too many positional arguments (10/5)
(R0917)
[refactor] 502-502: Too many local variables (16/15)
(R0914)
[refactor] 502-502: Too many branches (13/12)
(R0912)
🤖 Prompt for AI Agents
In python/thirdweb-ai/src/thirdweb_ai/services/insight.py between lines 499 and
580, the aggregate_transactions method has duplicated response cleanup logic and
high complexity due to many arguments and branches. To fix this, extract the
parameter building logic into a new helper method named
_build_transaction_params that takes the validated_aggregate, group_by,
chain_ids, from_address, to_address, function_signature, value_above, limit, and
page as arguments and returns the params dictionary. Also, replace the manual
cleanup of the response's aggregations with a call to the existing shared
cleanup method to avoid duplication and reduce complexity.
Comment on lines
+422
to
+498
| @tool( | ||
| description="Aggregate blockchain events with powerful grouping and aggregation options. Use this to get event counts, sums, or other aggregations grouped by fields like address, block, or time period." | ||
| ) | ||
| def aggregate_events( | ||
| self, | ||
| aggregate: Annotated[ | ||
| list[str], | ||
| f"Aggregation functions to apply. Valid options: {', '.join(VALID_EVENT_AGGREGATIONS)}. Can include aliases like 'count() as event_count'.", | ||
| ], | ||
| group_by: Annotated[ | ||
| Literal["address", "contract_address", "from_address", "block_number", "event_signature"] | None, | ||
| "Field to group events by for aggregation. Required when using aggregate functions.", | ||
| ] = None, | ||
| chain_id: Annotated[ | ||
| list[int] | int | None, | ||
| "Chain ID(s) to query (e.g., 1 for Ethereum Mainnet, 137 for Polygon). Specify multiple IDs as a list for cross-chain queries (max 5).", | ||
| ] = None, | ||
| contract_address: Annotated[ | ||
| str | None, | ||
| "Contract address to filter events by (e.g., '0x1234...'). Only return events emitted by this contract.", | ||
| ] = None, | ||
| transaction_hash: Annotated[ | ||
| str | None, | ||
| "Specific transaction hash to filter events by (e.g., '0xabc123...'). Useful for examining events in a particular transaction.", | ||
| ] = None, | ||
| event_signature: Annotated[ | ||
| str | None, | ||
| "Event signature to filter by (human-readable, e.g., 'Transfer(address,address,uint256)').", | ||
| ] = None, | ||
| limit: Annotated[ | ||
| int | None, | ||
| "Number of aggregated results to return (default 20, max 100).", | ||
| ] = 20, | ||
| page: Annotated[ | ||
| int | None, | ||
| "Page number for paginated results, starting from 0. 20 results are returned per page.", | ||
| ] = None, | ||
| ) -> dict[str, Any]: | ||
| # Validate aggregation functions | ||
| validated_aggregate = [validate_event_aggregation(agg) for agg in aggregate] | ||
|
|
||
| params: dict[str, Any] = { | ||
| "sort_by": "block_number", | ||
| "sort_order": "desc", | ||
| "decode": True, | ||
| "aggregate": validated_aggregate, | ||
| } | ||
|
|
||
| if group_by: | ||
| params["group_by"] = "address" if group_by == "contract_address" else group_by | ||
|
|
||
| chain_ids = chain_id if chain_id is not None else self.chain_ids | ||
| if chain_ids: | ||
| params["chain"] = chain_ids | ||
| if contract_address: | ||
| params["filter_address"] = validate_address(contract_address) | ||
| if transaction_hash: | ||
| params["filter_transaction_hash"] = validate_transaction_hash(transaction_hash) | ||
| if event_signature: | ||
| params["filter_event_signature"] = event_signature | ||
| if limit: | ||
| params["limit"] = limit | ||
| if page: | ||
| params["page"] = page | ||
|
|
||
| out = self._get("events", params) | ||
|
|
||
| # Clean up response by removing chain_id from aggregations if present | ||
| if out.get("aggregations"): | ||
| for agg in out["aggregations"]: | ||
| if isinstance(agg, dict): | ||
| for value in agg.values(): | ||
| if isinstance(value, dict) and "chain_id" in value: | ||
| value.pop("chain_id") | ||
|
|
||
| return out | ||
|
|
There was a problem hiding this comment.
🛠️ Refactor suggestion
Refactor to reduce complexity and improve maintainability.
While the functionality is correct, this method has high complexity (many parameters, branches, and nested logic).
Consider these improvements:
- Extract response cleanup logic to a shared method:
+ def _cleanup_aggregation_response(self, out: dict[str, Any]) -> dict[str, Any]:
+ """Remove chain_id from aggregation results."""
+ if out.get("aggregations"):
+ for agg in out["aggregations"]:
+ if isinstance(agg, dict):
+ for value in agg.values():
+ if isinstance(value, dict) and "chain_id" in value:
+ value.pop("chain_id")
+ return out- Clarify the group_by mapping logic:
if group_by:
- params["group_by"] = "address" if group_by == "contract_address" else group_by
+ # Map contract_address to address for API compatibility
+ mapped_group_by = "address" if group_by == "contract_address" else group_by
+ params["group_by"] = mapped_group_by- Consider using a parameter builder pattern for the complex parameter construction.
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| @tool( | |
| description="Aggregate blockchain events with powerful grouping and aggregation options. Use this to get event counts, sums, or other aggregations grouped by fields like address, block, or time period." | |
| ) | |
| def aggregate_events( | |
| self, | |
| aggregate: Annotated[ | |
| list[str], | |
| f"Aggregation functions to apply. Valid options: {', '.join(VALID_EVENT_AGGREGATIONS)}. Can include aliases like 'count() as event_count'.", | |
| ], | |
| group_by: Annotated[ | |
| Literal["address", "contract_address", "from_address", "block_number", "event_signature"] | None, | |
| "Field to group events by for aggregation. Required when using aggregate functions.", | |
| ] = None, | |
| chain_id: Annotated[ | |
| list[int] | int | None, | |
| "Chain ID(s) to query (e.g., 1 for Ethereum Mainnet, 137 for Polygon). Specify multiple IDs as a list for cross-chain queries (max 5).", | |
| ] = None, | |
| contract_address: Annotated[ | |
| str | None, | |
| "Contract address to filter events by (e.g., '0x1234...'). Only return events emitted by this contract.", | |
| ] = None, | |
| transaction_hash: Annotated[ | |
| str | None, | |
| "Specific transaction hash to filter events by (e.g., '0xabc123...'). Useful for examining events in a particular transaction.", | |
| ] = None, | |
| event_signature: Annotated[ | |
| str | None, | |
| "Event signature to filter by (human-readable, e.g., 'Transfer(address,address,uint256)').", | |
| ] = None, | |
| limit: Annotated[ | |
| int | None, | |
| "Number of aggregated results to return (default 20, max 100).", | |
| ] = 20, | |
| page: Annotated[ | |
| int | None, | |
| "Page number for paginated results, starting from 0. 20 results are returned per page.", | |
| ] = None, | |
| ) -> dict[str, Any]: | |
| # Validate aggregation functions | |
| validated_aggregate = [validate_event_aggregation(agg) for agg in aggregate] | |
| params: dict[str, Any] = { | |
| "sort_by": "block_number", | |
| "sort_order": "desc", | |
| "decode": True, | |
| "aggregate": validated_aggregate, | |
| } | |
| if group_by: | |
| params["group_by"] = "address" if group_by == "contract_address" else group_by | |
| chain_ids = chain_id if chain_id is not None else self.chain_ids | |
| if chain_ids: | |
| params["chain"] = chain_ids | |
| if contract_address: | |
| params["filter_address"] = validate_address(contract_address) | |
| if transaction_hash: | |
| params["filter_transaction_hash"] = validate_transaction_hash(transaction_hash) | |
| if event_signature: | |
| params["filter_event_signature"] = event_signature | |
| if limit: | |
| params["limit"] = limit | |
| if page: | |
| params["page"] = page | |
| out = self._get("events", params) | |
| # Clean up response by removing chain_id from aggregations if present | |
| if out.get("aggregations"): | |
| for agg in out["aggregations"]: | |
| if isinstance(agg, dict): | |
| for value in agg.values(): | |
| if isinstance(value, dict) and "chain_id" in value: | |
| value.pop("chain_id") | |
| return out | |
| @tool( | |
| description="Aggregate blockchain events with powerful grouping and aggregation options. Use this to get event counts, sums, or other aggregations grouped by fields like address, block, or time period." | |
| ) | |
| def aggregate_events( | |
| self, | |
| aggregate: Annotated[ | |
| list[str], | |
| f"Aggregation functions to apply. Valid options: {', '.join(VALID_EVENT_AGGREGATIONS)}. Can include aliases like 'count() as event_count'.", | |
| ], | |
| group_by: Annotated[ | |
| Literal["address", "contract_address", "from_address", "block_number", "event_signature"] | None, | |
| "Field to group events by for aggregation. Required when using aggregate functions.", | |
| ] = None, | |
| chain_id: Annotated[ | |
| list[int] | int | None, | |
| "Chain ID(s) to query (e.g., 1 for Ethereum Mainnet, 137 for Polygon). Specify multiple IDs as a list for cross-chain queries (max 5).", | |
| ] = None, | |
| contract_address: Annotated[ | |
| str | None, | |
| "Contract address to filter events by (e.g., '0x1234...'). Only return events emitted by this contract.", | |
| ] = None, | |
| transaction_hash: Annotated[ | |
| str | None, | |
| "Specific transaction hash to filter events by (e.g., '0xabc123...'). Useful for examining events in a particular transaction.", | |
| ] = None, | |
| event_signature: Annotated[ | |
| str | None, | |
| "Event signature to filter by (human-readable, e.g., 'Transfer(address,address,uint256)').", | |
| ] = None, | |
| limit: Annotated[ | |
| int | None, | |
| "Number of aggregated results to return (default 20, max 100).", | |
| ] = 20, | |
| page: Annotated[ | |
| int | None, | |
| "Page number for paginated results, starting from 0. 20 results are returned per page.", | |
| ] = None, | |
| ) -> dict[str, Any]: | |
| # Validate aggregation functions | |
| validated_aggregate = [validate_event_aggregation(agg) for agg in aggregate] | |
| params: dict[str, Any] = { | |
| "sort_by": "block_number", | |
| "sort_order": "desc", | |
| "decode": True, | |
| "aggregate": validated_aggregate, | |
| } | |
| if group_by: | |
| + # Map contract_address to address for API compatibility | |
| + mapped_group_by = "address" if group_by == "contract_address" else group_by | |
| + params["group_by"] = mapped_group_by | |
| chain_ids = chain_id if chain_id is not None else self.chain_ids | |
| if chain_ids: | |
| params["chain"] = chain_ids | |
| if contract_address: | |
| params["filter_address"] = validate_address(contract_address) | |
| if transaction_hash: | |
| params["filter_transaction_hash"] = validate_transaction_hash(transaction_hash) | |
| if event_signature: | |
| params["filter_event_signature"] = event_signature | |
| if limit: | |
| params["limit"] = limit | |
| if page: | |
| params["page"] = page | |
| out = self._get("events", params) | |
| # Clean up response by removing chain_id from aggregations if present | |
| if out.get("aggregations"): | |
| for agg in out["aggregations"]: | |
| if isinstance(agg, dict): | |
| for value in agg.values(): | |
| if isinstance(value, dict) and "chain_id" in value: | |
| value.pop("chain_id") | |
| return out | |
| + def _cleanup_aggregation_response(self, out: dict[str, Any]) -> dict[str, Any]: | |
| + """Remove chain_id from aggregation results.""" | |
| + if out.get("aggregations"): | |
| + for agg in out["aggregations"]: | |
| + if isinstance(agg, dict): | |
| + for value in agg.values(): | |
| + if isinstance(value, dict) and "chain_id" in value: | |
| + value.pop("chain_id") | |
| + return out |
🧰 Tools
🪛 Pylint (3.3.7)
[refactor] 425-425: Too many arguments (9/5)
(R0913)
[refactor] 425-425: Too many positional arguments (9/5)
(R0917)
🤖 Prompt for AI Agents
In python/thirdweb-ai/src/thirdweb_ai/services/insight.py around lines 422 to
498, the aggregate_events method is complex due to many parameters and nested
logic. Refactor by extracting the response cleanup logic that removes "chain_id"
from aggregations into a separate helper method to improve readability and
reuse. Simplify the group_by assignment by clarifying or encapsulating the
mapping logic, for example by using a dictionary mapping instead of inline
conditional. Finally, implement a parameter builder pattern or helper function
to construct the params dictionary step-by-step, which will reduce branching and
improve maintainability.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Summary by CodeRabbit