Added script for resetting balances on Anvil #646
Merged
Conversation
This file contains 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 pull request introduces a new Python script Changes
Sequence DiagramsequenceDiagram
participant CLI as Command Line
participant Script as reset_balance_anvil.py
participant RPC as Anvil RPC Interface
CLI->>Script: Provide RPC URL, agent balance, treasury balance
Script->>Script: Convert balances to wei
Script->>RPC: Send balance set requests for each address
RPC-->>Script: Confirm balance updates
Script-->>CLI: Report operation completion
Finishing Touches
Thank you for using CodeRabbit. We offer it for free to the OSS community and would appreciate your support in helping us grow. If you find it useful, would you consider giving us a shout-out on your favorite social media? 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
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: 3
🧹 Nitpick comments (2)
scripts/reset_balance_anvil.py (2)
32-46: Consider parallel processing for better performance.When dealing with multiple agents, sequential processing might be slow. Consider using async/await or multiprocessing to parallelize the balance updates.
+import asyncio +import aiohttp +from typing import List + +async def set_balance_async(session: aiohttp.ClientSession, rpc_url: str, address: str, balance: int) -> None: + # Similar to set_balance but using aiohttp + ... + +async def set_multiple_balances(rpc_url: str, addresses: List[str], balance: int) -> None: + async with aiohttp.ClientSession() as session: + tasks = [set_balance_async(session, rpc_url, addr, balance) for addr in addresses] + await asyncio.gather(*tasks) + def main( rpc_url: str, new_balance_agents_xdai: int, new_balance_treasury_xdai: int ) -> None: - for agent in DEPLOYED_NFT_AGENTS: - set_balance( - rpc_url=rpc_url, - address=agent.wallet_address, - balance=new_balance_agents_xdai, - ) + agent_addresses = [agent.wallet_address for agent in DEPLOYED_NFT_AGENTS] + asyncio.run(set_multiple_balances(rpc_url, agent_addresses, new_balance_agents_xdai))
49-50: Enhance CLI interface with better documentation and constraints.The CLI could be more user-friendly with proper argument descriptions, help text, and value constraints.
+from typing import Annotated + def main( - rpc_url: str, new_balance_agents_xdai: int, new_balance_treasury_xdai: int + rpc_url: Annotated[str, typer.Argument(help="Anvil RPC endpoint URL")], + new_balance_agents_xdai: Annotated[int, typer.Argument( + help="New balance for NFT agents in xDAI", + min=0, + max=1000000 + )], + new_balance_treasury_xdai: Annotated[int, typer.Argument( + help="New balance for treasury in xDAI", + min=0, + max=1000000 + )] ) -> None: + """Reset balances for NFT agents and treasury on Anvil network."""
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
scripts/reset_balance_anvil.py(1 hunks)
⏰ Context from checks skipped due to timeout of 90000ms (4)
- GitHub Check: test-build-image
- GitHub Check: pytest-docker
- GitHub Check: pytest
- GitHub Check: mypy
Comment on lines
+1
to
+12
| import requests | ||
| import typer | ||
| from prediction_market_agent_tooling.gtypes import xdai_type | ||
| from prediction_market_agent_tooling.tools.web3_utils import xdai_to_wei | ||
| from web3 import Web3 | ||
|
|
||
| from prediction_market_agent.agents.microchain_agent.nft_treasury_game.constants_nft_treasury_game import ( | ||
| TREASURY_ADDRESS, | ||
| ) | ||
| from prediction_market_agent.agents.microchain_agent.nft_treasury_game.deploy_nft_treasury_game import ( | ||
| DEPLOYED_NFT_AGENTS, | ||
| ) |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Add input validation for the RPC URL.
The script makes HTTP requests to an RPC endpoint without validating the URL format or checking for potential security risks (e.g., SSRF). Consider adding validation to ensure the RPC URL is safe and points to a local Anvil instance.
+def validate_rpc_url(url: str) -> None:
+ """Validate that the RPC URL is safe and points to a local Anvil instance."""
+ if not url.startswith(('http://127.0.0.1:', 'http://localhost:')):
+ raise ValueError("RPC URL must point to a local Anvil instance")
def main(
rpc_url: str, new_balance_agents_xdai: int, new_balance_treasury_xdai: int
) -> None:
+ validate_rpc_url(rpc_url)Committable suggestion skipped: line range outside the PR's diff.
Comment on lines
+15
to
+30
| def set_balance(rpc_url: str, address: str, balance: int) -> None: | ||
| balance_wei = xdai_to_wei(xdai_type(balance)) | ||
| data = { | ||
| "jsonrpc": "2.0", | ||
| "method": "anvil_setBalance", | ||
| "params": [ | ||
| Web3.to_checksum_address(address), | ||
| str(balance_wei), | ||
| ], | ||
| "id": 1, | ||
| } | ||
|
|
||
| response = requests.post(rpc_url, json=data) | ||
| response.raise_for_status() | ||
| print(f"Set balance {balance} xDAI for address {address}") | ||
|
|
There was a problem hiding this comment.
🛠️ Refactor suggestion
Enhance error handling and input validation.
The function needs improvements in several areas:
- Balance validation to prevent negative values
- Retry mechanism for transient RPC failures
- Explicit response success verification
- Structured logging instead of print statements
+import logging
+from typing import Union
+from tenacity import retry, stop_after_attempt, wait_exponential
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+def validate_balance(balance: Union[int, float]) -> None:
+ if balance < 0:
+ raise ValueError("Balance cannot be negative")
+
+@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def set_balance(rpc_url: str, address: str, balance: int) -> None:
+ validate_balance(balance)
balance_wei = xdai_to_wei(xdai_type(balance))
data = {
"jsonrpc": "2.0",
"method": "anvil_setBalance",
"params": [
Web3.to_checksum_address(address),
str(balance_wei),
],
"id": 1,
}
response = requests.post(rpc_url, json=data)
response.raise_for_status()
- print(f"Set balance {balance} xDAI for address {address}")
+ result = response.json()
+ if "error" in result:
+ raise ValueError(f"RPC error: {result['error']}")
+ logger.info(f"Set balance {balance} xDAI for address {address}")Committable suggestion skipped: line range outside the PR's diff.
Comment on lines
+1
to
+50
| import requests | ||
| import typer | ||
| from prediction_market_agent_tooling.gtypes import xdai_type | ||
| from prediction_market_agent_tooling.tools.web3_utils import xdai_to_wei | ||
| from web3 import Web3 | ||
|
|
||
| from prediction_market_agent.agents.microchain_agent.nft_treasury_game.constants_nft_treasury_game import ( | ||
| TREASURY_ADDRESS, | ||
| ) | ||
| from prediction_market_agent.agents.microchain_agent.nft_treasury_game.deploy_nft_treasury_game import ( | ||
| DEPLOYED_NFT_AGENTS, | ||
| ) | ||
|
|
||
|
|
||
| def set_balance(rpc_url: str, address: str, balance: int) -> None: | ||
| balance_wei = xdai_to_wei(xdai_type(balance)) | ||
| data = { | ||
| "jsonrpc": "2.0", | ||
| "method": "anvil_setBalance", | ||
| "params": [ | ||
| Web3.to_checksum_address(address), | ||
| str(balance_wei), | ||
| ], | ||
| "id": 1, | ||
| } | ||
|
|
||
| response = requests.post(rpc_url, json=data) | ||
| response.raise_for_status() | ||
| print(f"Set balance {balance} xDAI for address {address}") | ||
|
|
||
|
|
||
| def main( | ||
| rpc_url: str, new_balance_agents_xdai: int, new_balance_treasury_xdai: int | ||
| ) -> None: | ||
| for agent in DEPLOYED_NFT_AGENTS: | ||
| set_balance( | ||
| rpc_url=rpc_url, | ||
| address=agent.wallet_address, | ||
| balance=new_balance_agents_xdai, | ||
| ) | ||
|
|
||
| set_balance( | ||
| rpc_url=rpc_url, | ||
| address=TREASURY_ADDRESS, | ||
| balance=new_balance_treasury_xdai, | ||
| ) | ||
|
|
||
|
|
||
| if __name__ == "__main__": | ||
| typer.run(main) |
There was a problem hiding this comment.
Add safeguards for production environments.
This script modifies blockchain state and handles financial values. Add checks to prevent accidental execution in production environments.
+import os
+from enum import Enum
+
+class Environment(str, Enum):
+ LOCAL = "local"
+ TEST = "test"
+
+def validate_environment() -> None:
+ """Ensure script runs only in safe environments."""
+ env = os.getenv("ENVIRONMENT", "").lower()
+ if env not in (Environment.LOCAL, Environment.TEST):
+ raise RuntimeError("This script should only run in local or test environments")
+
def main(
rpc_url: str, new_balance_agents_xdai: int, new_balance_treasury_xdai: int
) -> None:
+ validate_environment()📝 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
| import requests | |
| import typer | |
| from prediction_market_agent_tooling.gtypes import xdai_type | |
| from prediction_market_agent_tooling.tools.web3_utils import xdai_to_wei | |
| from web3 import Web3 | |
| from prediction_market_agent.agents.microchain_agent.nft_treasury_game.constants_nft_treasury_game import ( | |
| TREASURY_ADDRESS, | |
| ) | |
| from prediction_market_agent.agents.microchain_agent.nft_treasury_game.deploy_nft_treasury_game import ( | |
| DEPLOYED_NFT_AGENTS, | |
| ) | |
| def set_balance(rpc_url: str, address: str, balance: int) -> None: | |
| balance_wei = xdai_to_wei(xdai_type(balance)) | |
| data = { | |
| "jsonrpc": "2.0", | |
| "method": "anvil_setBalance", | |
| "params": [ | |
| Web3.to_checksum_address(address), | |
| str(balance_wei), | |
| ], | |
| "id": 1, | |
| } | |
| response = requests.post(rpc_url, json=data) | |
| response.raise_for_status() | |
| print(f"Set balance {balance} xDAI for address {address}") | |
| def main( | |
| rpc_url: str, new_balance_agents_xdai: int, new_balance_treasury_xdai: int | |
| ) -> None: | |
| for agent in DEPLOYED_NFT_AGENTS: | |
| set_balance( | |
| rpc_url=rpc_url, | |
| address=agent.wallet_address, | |
| balance=new_balance_agents_xdai, | |
| ) | |
| set_balance( | |
| rpc_url=rpc_url, | |
| address=TREASURY_ADDRESS, | |
| balance=new_balance_treasury_xdai, | |
| ) | |
| if __name__ == "__main__": | |
| typer.run(main) | |
| import os | |
| import requests | |
| import typer | |
| from enum import Enum | |
| from prediction_market_agent_tooling.gtypes import xdai_type | |
| from prediction_market_agent_tooling.tools.web3_utils import xdai_to_wei | |
| from web3 import Web3 | |
| from prediction_market_agent.agents.microchain_agent.nft_treasury_game.constants_nft_treasury_game import ( | |
| TREASURY_ADDRESS, | |
| ) | |
| from prediction_market_agent.agents.microchain_agent.nft_treasury_game.deploy_nft_treasury_game import ( | |
| DEPLOYED_NFT_AGENTS, | |
| ) | |
| class Environment(str, Enum): | |
| LOCAL = "local" | |
| TEST = "test" | |
| def validate_environment() -> None: | |
| """Ensure script runs only in safe environments.""" | |
| env = os.getenv("ENVIRONMENT", "").lower() | |
| if env not in (Environment.LOCAL, Environment.TEST): | |
| raise RuntimeError("This script should only run in local or test environments") | |
| def set_balance(rpc_url: str, address: str, balance: int) -> None: | |
| balance_wei = xdai_to_wei(xdai_type(balance)) | |
| data = { | |
| "jsonrpc": "2.0", | |
| "method": "anvil_setBalance", | |
| "params": [ | |
| Web3.to_checksum_address(address), | |
| str(balance_wei), | |
| ], | |
| "id": 1, | |
| } | |
| response = requests.post(rpc_url, json=data) | |
| response.raise_for_status() | |
| print(f"Set balance {balance} xDAI for address {address}") | |
| def main( | |
| rpc_url: str, new_balance_agents_xdai: int, new_balance_treasury_xdai: int | |
| ) -> None: | |
| validate_environment() | |
| for agent in DEPLOYED_NFT_AGENTS: | |
| set_balance( | |
| rpc_url=rpc_url, | |
| address=agent.wallet_address, | |
| balance=new_balance_agents_xdai, | |
| ) | |
| set_balance( | |
| rpc_url=rpc_url, | |
| address=TREASURY_ADDRESS, | |
| balance=new_balance_treasury_xdai, | |
| ) | |
| if __name__ == "__main__": | |
| typer.run(main) |
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.
No description provided.