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

use crate::*;
use polars::prelude::*;
use std::collections::HashMap;

/// StateDiffs
#[derive(Default)]
pub struct StateDiffs(
    balance_diffs::BalanceDiffs,
    code_diffs::CodeDiffs,
    nonce_diffs::NonceDiffs,
    storage_diffs::StorageDiffs,
);

type BlockTxsTraces = (Option<u32>, Vec<Option<Vec<u8>>>, Vec<ethers::types::BlockTrace>);

impl ToDataFrames for StateDiffs {
    fn create_dfs(
        self,
        schemas: &HashMap<Datatype, Table>,
        chain_id: u64,
    ) -> R<HashMap<Datatype, DataFrame>> {
        let StateDiffs(balances, codes, nonces, storages) = self;
        let mut output = HashMap::new();
        output.extend(balances.create_dfs(schemas, chain_id)?);
        output.extend(codes.create_dfs(schemas, chain_id)?);
        output.extend(nonces.create_dfs(schemas, chain_id)?);
        output.extend(storages.create_dfs(schemas, chain_id)?);
        Ok(output)
    }
}

#[async_trait::async_trait]
impl CollectByBlock for StateDiffs {
    type Response = BlockTxsTraces;

    async fn extract(request: Params, source: Arc<Source>, query: Arc<Query>) -> R<Self::Response> {
        let include_txs = query.schemas.values().any(|x| x.has_column("transaction_hash"));
        source.trace_block_state_diffs(request.block_number()? as u32, include_txs).await
    }

    fn transform(response: Self::Response, columns: &mut Self, query: &Arc<Query>) -> R<()> {
        process_state_diffs(response, columns, &query.schemas)
    }
}

#[async_trait::async_trait]
impl CollectByTransaction for StateDiffs {
    type Response = BlockTxsTraces;

    async fn extract(request: Params, source: Arc<Source>, _: Arc<Query>) -> R<Self::Response> {
        source.trace_transaction_state_diffs(request.transaction_hash()?).await
    }

    fn transform(response: Self::Response, columns: &mut Self, query: &Arc<Query>) -> R<()> {
        process_state_diffs(response, columns, &query.schemas)
    }
}

fn process_state_diffs(
    response: BlockTxsTraces,
    columns: &mut StateDiffs,
    schemas: &HashMap<Datatype, Table>,
) -> R<()> {
    let StateDiffs(balances, codes, nonces, storages) = columns;
    balance_diffs::process_balance_diffs(&response, balances, schemas)?;
    code_diffs::process_code_diffs(&response, codes, schemas)?;
    nonce_diffs::process_nonce_diffs(&response, nonces, schemas)?;
    storage_diffs::process_storage_diffs(&response, storages, schemas)?;
    Ok(())
}