Coverage for app / services / blockchain.py: 35%

1"""Blockchain API service for querying balances and transactions.""" 

2import httpx 

3from typing import Optional, List, Dict, Any 

4from app.core.config import settings 

5 

6 

7class BlockchainAPIError(Exception): 

8 """Raised when blockchain API calls fail.""" 

9 pass 

10 

11 

12class BlockchainService: 

13 """Service for interacting with blockchain APIs (BlockCypher, Helius).""" 

14 

15 def __init__(self): 

16 self.blockcypher_base = "https://api.blockcypher.com/v1" 

17 self.helius_base = "https://api.helius.xyz/v0" 

18 self.blockcypher_token = settings.blockcypher_api_token 

19 self.helius_key = settings.helius_api_key 

20 

21 def _get_blockcypher_url(self, chain: str, endpoint: str) -> str: 

22 """Build BlockCypher API URL.""" 

23 url = f"{self.blockcypher_base}/{chain}/main/{endpoint}" 

24 if self.blockcypher_token: 

25 url += f"?token={self.blockcypher_token}" 

26 return url 

27 

28 async def get_btc_balance(self, address: str) -> int: 

29 """Get BTC balance in satoshi.""" 

30 try: 

31 url = self._get_blockcypher_url("btc", f"addrs/{address}/balance") 

32 async with httpx.AsyncClient(timeout=10.0) as client: 

33 response = await client.get(url) 

34 response.raise_for_status() 

35 data = response.json() 

36 return data.get("balance", 0) 

37 except httpx.HTTPError as e: 

38 raise BlockchainAPIError(f"Failed to fetch BTC balance: {e}") 

39 

40 async def get_eth_balance(self, address: str) -> int: 

41 """Get ETH balance in wei.""" 

42 try: 

43 url = self._get_blockcypher_url("eth", f"addrs/{address}/balance") 

44 async with httpx.AsyncClient(timeout=10.0) as client: 

45 response = await client.get(url) 

46 response.raise_for_status() 

47 data = response.json() 

48 return data.get("balance", 0) 

49 except httpx.HTTPError as e: 

50 raise BlockchainAPIError(f"Failed to fetch ETH balance: {e}") 

51 

52 async def get_doge_balance(self, address: str) -> int: 

53 """Get DOGE balance in satoshi (smallest unit).""" 

54 try: 

55 url = self._get_blockcypher_url("doge", f"addrs/{address}/balance") 

56 async with httpx.AsyncClient(timeout=10.0) as client: 

57 response = await client.get(url) 

58 response.raise_for_status() 

59 data = response.json() 

60 return data.get("balance", 0) 

61 except httpx.HTTPError as e: 

62 raise BlockchainAPIError(f"Failed to fetch DOGE balance: {e}") 

63 

64 async def get_sol_balance(self, address: str) -> int: 

65 """Get SOL balance in lamports.""" 

66 if not self.helius_key: 

67 raise BlockchainAPIError("Helius API key not configured") 

68 

69 try: 

70 url = f"{self.helius_base}/addresses/{address}/balances" 

71 params = {"api-key": self.helius_key} 

72 async with httpx.AsyncClient(timeout=10.0) as client: 

73 response = await client.get(url, params=params) 

74 response.raise_for_status() 

75 data = response.json() 

76 # Helius returns nativeBalance in lamports 

77 native_balance = data.get("nativeBalance", 0) 

78 return int(native_balance) 

79 except httpx.HTTPError as e: 

80 raise BlockchainAPIError(f"Failed to fetch SOL balance: {e}") 

81 

82 async def get_btc_transactions(self, address: str, limit: int = 50) -> List[Dict[str, Any]]: 

83 """Get BTC transactions for an address.""" 

84 try: 

85 url = self._get_blockcypher_url("btc", f"addrs/{address}/txs") 

86 params = {"limit": limit} 

87 async with httpx.AsyncClient(timeout=10.0) as client: 

88 response = await client.get(url, params=params) 

89 response.raise_for_status() 

90 data = response.json() 

91 return data.get("txs", []) 

92 except httpx.HTTPError as e: 

93 raise BlockchainAPIError(f"Failed to fetch BTC transactions: {e}") 

94 

95 async def get_eth_transactions(self, address: str, limit: int = 50) -> List[Dict[str, Any]]: 

96 """Get ETH transactions for an address.""" 

97 try: 

98 url = self._get_blockcypher_url("eth", f"addrs/{address}/txs") 

99 params = {"limit": limit} 

100 async with httpx.AsyncClient(timeout=10.0) as client: 

101 response = await client.get(url, params=params) 

102 response.raise_for_status() 

103 data = response.json() 

104 return data.get("txs", []) 

105 except httpx.HTTPError as e: 

106 raise BlockchainAPIError(f"Failed to fetch ETH transactions: {e}") 

107 

108 async def get_doge_transactions(self, address: str, limit: int = 50) -> List[Dict[str, Any]]: 

109 """Get DOGE transactions for an address.""" 

110 try: 

111 url = self._get_blockcypher_url("doge", f"addrs/{address}/txs") 

112 params = {"limit": limit} 

113 async with httpx.AsyncClient(timeout=10.0) as client: 

114 response = await client.get(url, params=params) 

115 response.raise_for_status() 

116 data = response.json() 

117 return data.get("txs", []) 

118 except httpx.HTTPError as e: 

119 raise BlockchainAPIError(f"Failed to fetch DOGE transactions: {e}") 

120 

121 async def get_sol_transactions(self, address: str, limit: int = 50) -> List[Dict[str, Any]]: 

122 """Get SOL transactions for an address.""" 

123 if not self.helius_key: 

124 raise BlockchainAPIError("Helius API key not configured") 

125 

126 try: 

127 url = f"{self.helius_base}/addresses/{address}/transactions" 

128 params = { 

129 "api-key": self.helius_key, 

130 "limit": limit, 

131 } 

132 async with httpx.AsyncClient(timeout=10.0) as client: 

133 response = await client.get(url, params=params) 

134 response.raise_for_status() 

135 data = response.json() 

136 return data if isinstance(data, list) else [] 

137 except httpx.HTTPError as e: 

138 raise BlockchainAPIError(f"Failed to fetch SOL transactions: {e}")