Manager

Blockchain data collection Index / src / data_collection / App / Db / Manager

Auto-generated documentation for src.data_collection.app.db.manager module.

DatabaseManager

Manage the PostgreSQL connection and CRUD operations.

https://magicstack.github.io/asyncpg/current/usage.html

Signature

class DatabaseManager:
    def __init__(self, postgresql_dsn: str, node_name: str) -> None:
        ...

DatabaseManager().connect

Show source in manager.py:31

Connects to the PostgreSQL database.

Signature

async def connect(self):
    ...

DatabaseManager().disconnect

Show source in manager.py:37

Disconnect from PostgreSQL

Signature

async def disconnect(self):
    ...

DatabaseManager().get_block

Show source in manager.py:325

Get the block data with a given identifier from the block table.

Notes

If block_identifier is None, returns the latest block data in the table.

Returns

Optional[dict[str, Any]] - if available returns block data as a dict, otherwise returns None

Signature

async def get_block(
    self, block_identifier: Optional[Union[str, int]] = None
) -> Optional[dict[str, Any]]:
    ...

DatabaseManager().insert_block

Show source in manager.py:42

Insert block data into _block table.

Signature

async def insert_block(
    self,
    block_number: int,
    block_hash: str,
    nonce: str,
    difficulty: int,
    gas_limit: int,
    gas_used: int,
    timestamp: datetime,
    miner: str,
    parent_hash: str,
    block_reward: float,
    uncles: list[str],
):
    ...

DatabaseManager().insert_contract

Show source in manager.py:211

Insert contract data into _contract table.

Signature

async def insert_contract(
    self, address: str, transaction_hash: str, is_pair_contract: bool
):
    ...

DatabaseManager().insert_contract_supply_change

Show source in manager.py:258

Insert token contract supply change data into _token_contract_supply_change table.

Signature

async def insert_contract_supply_change(
    self, address: str, amount_changed: int, transaction_hash: str
):
    ...

DatabaseManager().insert_internal_transaction

Show source in manager.py:122

Insert internal transaction data into _internal_transaction table.

Signature

async def insert_internal_transaction(
    self,
    transaction_hash: str,
    from_address: str,
    to_address: str,
    value: float,
    gas_limit: float,
    gas_used: int,
    input_data: str,
    call_type: str,
):
    ...

DatabaseManager().insert_nft_transfer

Show source in manager.py:182

Insert nft transfer data into _nft_transfer table.

Signature

async def insert_nft_transfer(
    self,
    transaction_hash: str,
    log_index: int,
    address: str,
    from_address: str,
    to_address: str,
    token_id: int,
):
    ...

DatabaseManager().insert_pair_contract

Show source in manager.py:277

Insert pair contract data into _pair_contract table.

Signature

async def insert_pair_contract(
    self,
    address: str,
    token0_address: str,
    token1_address: str,
    reserve0: int,
    reserve1: int,
    factory: str,
):
    ...

DatabaseManager().insert_pair_liquidity_change

Show source in manager.py:305

Insert pair liquidity change data into _token_contract_supply_change table.

Signature

async def insert_pair_liquidity_change(
    self, address: str, amount0: int, amount1: int, transaction_hash: str
):
    ...

DatabaseManager().insert_token_contract

Show source in manager.py:230

Insert token contract data into _token_contract table.

Signature

async def insert_token_contract(
    self,
    address: str,
    symbol: str,
    name: str,
    decimals: int,
    total_supply: int,
    token_category: str,
):
    ...

DatabaseManager().insert_transaction

Show source in manager.py:81

Insert transaction data into _transaction table.

Signature

async def insert_transaction(
    self,
    transaction_hash: str,
    block_number: int,
    from_address: str,
    to_address: str,
    value: float,
    transaction_fee: float,
    gas_price: float,
    gas_limit: int,
    gas_used: int,
    is_token_tx: bool,
    input_data: str,
):
    ...

DatabaseManager().insert_transaction_logs

Show source in manager.py:153

Insert transaction logs data into _transaction_logs table.

Signature

async def insert_transaction_logs(
    self,
    transaction_hash: str,
    address: str,
    log_index: int,
    data: str,
    removed: bool,
    topics: list[str],
):
    ...