(
self, token_address: TokenAddress = None, partner_address: Address = None
)
| 1264 | return self.raiden.wal.storage.get_events_with_timestamps(limit=limit, offset=offset) |
| 1265 | |
| 1266 | def get_pending_transfers( |
| 1267 | self, token_address: TokenAddress = None, partner_address: Address = None |
| 1268 | ) -> List[Dict[str, Any]]: |
| 1269 | chain_state = views.state_from_raiden(self.raiden) |
| 1270 | transfer_tasks = views.get_all_transfer_tasks(chain_state) |
| 1271 | channel_id = None |
| 1272 | confirmed_block_identifier = chain_state.block_hash |
| 1273 | if token_address is not None: |
| 1274 | token_network = self.raiden.default_registry.get_token_network( |
| 1275 | token_address=token_address, block_identifier=confirmed_block_identifier |
| 1276 | ) |
| 1277 | if token_network is None: |
| 1278 | raise UnknownTokenAddress(f"Token {to_checksum_address(token_address)} not found.") |
| 1279 | if partner_address is not None: |
| 1280 | partner_channel = views.get_channelstate_for( |
| 1281 | chain_state=chain_state, |
| 1282 | token_network_registry_address=self.raiden.default_registry.address, |
| 1283 | token_address=token_address, |
| 1284 | partner_address=partner_address, |
| 1285 | ) |
| 1286 | if not partner_channel: |
| 1287 | raise ChannelNotFound("Channel with partner `partner_address not found.`") |
| 1288 | channel_id = partner_channel.identifier |
| 1289 | |
| 1290 | return transfer_tasks_view(transfer_tasks, token_address, channel_id) |
| 1291 | |
| 1292 | def set_total_udc_deposit(self, new_total_deposit: TokenAmount) -> TransactionHash: |
| 1293 | """Set the `total_deposit` in the UserDeposit contract by sending an on-chain transaction. |
nothing calls this directly
no test coverage detected