Spaces:
Build error
Build error
File size: 3,218 Bytes
4585d4c |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 |
import requests
import logging
from typing import Any, Dict, List, Optional
from aptos_sdk.async_client import RestClient as AsyncRestClient
from utils.nest_runner import run_async, run_coroutine, async_to_sync
def _run_coro_sync(coro):
"""Run coroutine synchronously, using the clean nest_asyncio implementation.
This function is kept for backward compatibility and delegates to the more
clean and robust nest_runner utilities.
"""
return async_to_sync(coro)
class RestClientSync:
"""A tiny sync wrapper around aptos_sdk.async_client.RestClient.
This runs async calls via a safe sync runner to provide a synchronous API
suitable for Streamlit scripts. Add more proxy methods as needed.
"""
def __init__(self, node_url: str):
self._client = AsyncRestClient(node_url)
def account(self, address: str) -> Any:
return _run_coro_sync(self._client.account(address))
def account_resources(self, address: str) -> Any:
# Use a completely fresh call each time to avoid event loop issues
try:
return _run_coro_sync(self._client.account_resources(address))
except Exception as e:
if "Event loop is closed" in str(e):
# If we get the specific error, recreate client and retry
self._client = AsyncRestClient(self._client.base_url)
return _run_coro_sync(self._client.account_resources(address))
raise
def create_transaction(self, sender: str, payload: Any) -> Any:
return _run_coro_sync(self._client.create_transaction(sender, payload))
def submit_transaction(self, signed_txn: Any) -> Any:
return _run_coro_sync(self._client.submit_transaction(signed_txn))
def wait_for_transaction(self, txn_hash: str, timeout: int = 30) -> Any:
return _run_coro_sync(self._client.wait_for_transaction(txn_hash, timeout))
def get_account_transactions(self, address: str, limit: int = 20) -> List[Dict[str, Any]]:
"""Fetch transaction history for an account
This method makes a direct HTTP request since AsyncRestClient doesn't have this method.
"""
try:
# Extract base URL from client
base_url = self._client.base_url
if base_url.endswith('/'):
base_url = base_url[:-1]
# Construct the API endpoint URL
url = f"{base_url}/accounts/{address}/transactions"
# Set up query parameters
params = {
'limit': limit
}
# Make the HTTP request
response = requests.get(url, params=params)
# Check if the request was successful
if response.status_code == 200:
return response.json()
else:
logging.error(f"Error fetching transactions: HTTP {response.status_code}: {response.text}")
return []
except Exception as e:
logging.error(f"Error in get_account_transactions: {str(e)}")
return []
def __getattr__(self, name: str):
# Fallback to underlying client attributes if needed
return getattr(self._client, name)
|