cryo_freeze 0.3.2

cryo is the easiest way to extract blockchain data to parquet, csv, or json
Documentation 
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
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109

use crate::*;
use ethers::prelude::*;
use polars::prelude::*;

/// columns for transactions
#[cryo_to_df::to_df(Datatype::NativeTransfers)]
#[derive(Default)]
pub struct NativeTransfers {
    n_rows: u64,
    block_number: Vec<u32>,
    block_hash: Vec<Vec<u8>>,
    transaction_index: Vec<Option<u32>>,
    transfer_index: Vec<u32>,
    transaction_hash: Vec<Option<Vec<u8>>>,
    from_address: Vec<Vec<u8>>,
    to_address: Vec<Vec<u8>>,
    value: Vec<U256>,
    chain_id: Vec<u64>,
}

#[async_trait::async_trait]
impl Dataset for NativeTransfers {
    fn optional_parameters() -> Vec<Dim> {
        vec![Dim::FromAddress, Dim::ToAddress]
    }
}

#[async_trait::async_trait]
impl CollectByBlock for NativeTransfers {
    type Response = Vec<Trace>;

    async fn extract(request: Params, source: Arc<Source>, _: Arc<Query>) -> R<Self::Response> {
        let traces = source.trace_block(request.block_number()?.into()).await?;
        Ok(filter_traces_by_from_to_addresses(traces, &request.from_address, &request.to_address))
    }

    fn transform(response: Self::Response, columns: &mut Self, query: &Arc<Query>) -> R<()> {
        let traces =
            if query.exclude_failed { traces::filter_failed_traces(response) } else { response };
        process_native_transfers(&traces, columns, &query.schemas)
    }
}

#[async_trait::async_trait]
impl CollectByTransaction for NativeTransfers {
    type Response = Vec<Trace>;

    async fn extract(request: Params, source: Arc<Source>, _: Arc<Query>) -> R<Self::Response> {
        let traces = source.trace_transaction(request.ethers_transaction_hash()?).await?;
        Ok(filter_traces_by_from_to_addresses(traces, &request.from_address, &request.to_address))
    }

    fn transform(response: Self::Response, columns: &mut Self, query: &Arc<Query>) -> R<()> {
        let traces =
            if query.exclude_failed { traces::filter_failed_traces(response) } else { response };
        process_native_transfers(&traces, columns, &query.schemas)
    }
}

/// process block into columns
pub(crate) fn process_native_transfers(
    traces: &[Trace],
    columns: &mut NativeTransfers,
    schemas: &Schemas,
) -> R<()> {
    let schema = schemas.get(&Datatype::NativeTransfers).ok_or(err("schema not provided"))?;
    for (transfer_index, trace) in traces.iter().enumerate() {
        columns.n_rows += 1;
        store!(schema, columns, block_number, trace.block_number as u32);
        store!(schema, columns, transaction_index, trace.transaction_position.map(|x| x as u32));
        store!(schema, columns, block_hash, trace.block_hash.as_bytes().to_vec());
        store!(schema, columns, transfer_index, transfer_index as u32);
        store!(
            schema,
            columns,
            transaction_hash,
            trace.transaction_hash.map(|x| x.as_bytes().to_vec())
        );

        match &trace.action {
            Action::Call(action) => {
                store!(schema, columns, from_address, action.from.as_bytes().to_vec());
                store!(schema, columns, to_address, action.to.as_bytes().to_vec());
                store!(schema, columns, value, action.value);
            }
            Action::Create(action) => {
                store!(schema, columns, from_address, action.from.as_bytes().to_vec());
                match &trace.result.as_ref() {
                    Some(Res::Create(res)) => {
                        store!(schema, columns, to_address, res.address.0.into())
                    }
                    _ => store!(schema, columns, to_address, vec![0; 32]),
                }
                store!(schema, columns, value, action.value);
            }
            Action::Suicide(action) => {
                store!(schema, columns, from_address, action.address.as_bytes().to_vec());
                store!(schema, columns, to_address, action.refund_address.as_bytes().to_vec());
                store!(schema, columns, value, action.balance);
            }
            Action::Reward(action) => {
                store!(schema, columns, from_address, vec![0; 20]);
                store!(schema, columns, to_address, action.author.as_bytes().to_vec());
                store!(schema, columns, value, action.value);
            }
        }
    }
    Ok(())
}