def get_channel_events(self, channel_address, from_block, to_block=None):
if not isaddress(channel_address):
raise InvalidAddress(
'Expected binary address format for channel in get_channel_events'
)
returned_events = get_all_netting_channel_events(
self.raiden.chain,
channel_address,
events=ALL_EVENTS,
from_block=from_block,
to_block=to_block,
)
raiden_events = self.raiden.transaction_log.get_events_in_block_range(
from_block=from_block,
to_block=to_block
)
# Here choose which raiden internal events we want to expose to the end user
for event in raiden_events:
is_user_transfer_event = isinstance(event.event_object, (
EventTransferSentSuccess,
EventTransferSentFailed,
EventTransferReceivedSuccess
))
if is_user_transfer_event:
new_event = {
'block_number': event.block_number,
'_event_type': type(event.event_object).__name__,
}
new_event.update(event.event_object.__dict__)
returned_events.append(new_event)
return returned_events
评论列表
文章目录