DataBase
- class ScanWatch.storage.ScanDataBase.ScanDataBase(name: str = 'scan_db')[source]
Handles the recording of the address transactions in a local database
- __init__(name: str = 'scan_db')[source]
Initialise a Scan database instance
- Parameters
name (str) – name of the database
- add_transactions(address: str, nt_type: ScanWatch.utils.enums.NETWORK, net: str, tr_type: ScanWatch.utils.enums.TRANSACTION, transactions: List[Dict])[source]
Add a list of transactions to the database
- Parameters
address (str) – address involved in the transaction
nt_type (NETWORK) – type of network
net (str) – name of the network, used to differentiate main and test nets
tr_type (TRANSACTION) – type of the transaction to record
transactions (List[Dict]) – list of the transaction to record
- Returns
None
- Return type
None
- get_last_block_number(address: str, nt_type: ScanWatch.utils.enums.NETWORK, net: str, tr_type: ScanWatch.utils.enums.TRANSACTION) int[source]
Return the last block number seen in recorded transactions (per address, type of transaction and network) If None are found, return 0
- Parameters
address (str) – address involved in the transactions
nt_type (NETWORK) – type of network
net (str) – name of the network, used to differentiate main and test nets
tr_type (TRANSACTION) – type of the transaction to fetch
- Returns
last block number
- Return type
int
- get_transactions(address: str, nt_type: ScanWatch.utils.enums.NETWORK, net: str, tr_type: ScanWatch.utils.enums.TRANSACTION) List[Dict][source]
Return the List of the transactions recorded in the database
- Parameters
address (str) – address involved in the transactions
nt_type (NETWORK) – type of network
net (str) – name of the network, used to differentiate main and test nets
tr_type (TRANSACTION) – type of the transaction to fetch
- Returns
list of the transaction recorded
- Return type
List[Dict]
- class ScanWatch.storage.DataBase.DataBase(name: str)[source]
This class will be used to interact with sqlite3 databases without having to generates sqlite commands
- __init__(name: str)[source]
Initialise a DataBase instance
- Parameters
name (str) – name of the database
- add_row(table: ScanWatch.storage.tables.Table, row: Tuple, auto_commit: bool = True, update_if_exists: bool = False)[source]
Add a row to a table
- Parameters
table (Table) – table to add a row to
row (Tuple) – values to add to the database
auto_commit (bool) – if the database state should be saved after the changes
update_if_exists (bool) – if an integrity error is raised and this parameter is true, will update the existing row
- Returns
None
- Return type
None
- add_rows(table: ScanWatch.storage.tables.Table, rows: List[Tuple], auto_commit: bool = True, update_if_exists: bool = False)[source]
Add several rows to a table
- Parameters
table (Table) – table to add a row to
rows (List[Tuple]) – list of values to add to the database
auto_commit (bool) – if the database state should be saved after the changes
update_if_exists (bool) – if an integrity error is raised and this parameter is true, will update the existing row
- Returns
None
- Return type
None
- create_table(table: ScanWatch.storage.tables.Table)[source]
Create a table in the database
- Parameters
table (Table) – Table instance with the config of the table to create
- Returns
None
- Return type
None
- drop_all_tables()[source]
Drop all the tables existing in the database
- Returns
None
- Return type
None
- drop_table(table: Union[ScanWatch.storage.tables.Table, str])[source]
Delete a table from the database
- Parameters
table (Union[Table, str]) – table or table name to drop
- Returns
None
- Return type
None
- get_all_rows(table: ScanWatch.storage.tables.Table) List[Tuple][source]
Get all the rows of a table
- Parameters
table (Table) – table to get the rows from
- Returns
all the rows of the table
- Return type
List[Tuple]
- get_all_tables() List[Tuple][source]
Return all the tables existing in the database
- Returns
tables descriptions
- Return type
List[Tuple]
- get_conditions_rows(table: ScanWatch.storage.tables.Table, selection: Union[str, List[str]] = '*', conditions_list: Optional[List[Tuple[str, ScanWatch.storage.DataBase.SQLConditionEnum, Any]]] = None, order_list: Optional[List[str]] = None) List[Tuple][source]
Select rows with optional conditions and optional order
- Parameters
table (Table) – table to select the rows from
selection (Union[str, List[str]]) – list of column or SQL type selection
conditions_list (Optional[List[Tuple[str, SQLConditionEnum, Any]]]) – list of conditions to select the row
order_list (Optional[List[str]]) – List of SQL type order by
- Returns
the selected rows
- Return type
List[Tuple]
- static get_create_cmd(table: ScanWatch.storage.tables.Table) str[source]
Return the command in string format to create a table in the database
- Parameters
table (Table) – Table instance with the config if the table to create
- Returns
execution command for the table creation
- Return type
str
- get_row_by_key(table: ScanWatch.storage.tables.Table, key_value) Optional[Tuple][source]
Get the row identified by a primary key value from a table
- Parameters
table (Table) – table to fetch the key from
key_value (Any) – value of the primary key
- Returns
the raw row of of the table
- Return type
Optional[Tuple]
- update_row(table: ScanWatch.storage.tables.Table, row: Tuple, auto_commit=True)[source]
Update the value of a row in a table
- Parameters
table (Table) – table to get updated
row (Tuple) – values to update
auto_commit (bool) – if the database state should be saved after the changes
- Returns
None
- Return type
None