pawnlib.utils package

from pawnlib.utils import *

Submodules

pawnlib.utils.http module

from pawnlib.utils import http

class StrEnum(value)[source]

Bases: str, Enum

An enumeration.

class AllowsHttpMethod(value)[source]

Bases: StrEnum

An enumeration.

get = 'get'

post = 'post'

patch = 'patch'

delete = 'delete'

class AllowsKey(value)[source]

Bases: StrEnum

An enumeration.

status_code = 'status_code'

headers = 'headers'

raw = 'raw'

url = 'url'

reason = 'reason'

http_version = 'http_version'

r_headers = 'r_headers'

result = 'result'

elapsed = 'elapsed'

response_time = 'response_time'

class NetworkInfo(network_name='', platform='icon', force=False, network_api='', planet_api='', nid='', network='', endpoint='', tracker='', symbol='', valid_network=False)[source]

Bases: object

network_name: str = ''

platform: str = 'icon'

force: bool = False

network_api: str = ''

planet_api: str = ''

nid: str = ''

network: str = ''

endpoint: str = ''

tracker: str = ''

symbol: str = ''

valid_network: bool = False

MANDATORY_KEYS = ['platform', 'nid']

STATIC_VALUES = ['nid', 'network_api', 'endpoint']

is_set_static_values()[source]

set_network(network_name=None, platform='icon')[source]

reset_static_values()[source]

list()[source]

Return type:: list

get_platform_list()[source]

Return type:: list

get_platform_info()[source]

Return type:: dict

update_platform_info(data={})[source]

Return type:: dict

add_network(platform, network_name, network_api, nid)[source]

get_network_list(platform='')[source]

Return type:: list

tuple()[source]

Return type:: tuple

to_dict()[source]

find_network_by_platform_and_nid(platform, nid)[source]

Find network and endpoint information based on platform and nid.

Parameters:

platform (str) – The platform name (e.g., ‘icon’, ‘havah’).
nid (str) – The network ID in hexadecimal format (e.g., ‘0x1’).

Returns:

Network information including endpoint, network_api, and tracker.

Return type:

dict

Raises:

ValueError – If the platform or nid is not found in the platform info.

fetch_network_info()[source]

class IconRpcTemplates(category=None, method=None)[source]

Bases: object

requires_sign_method = ['icx_sendTransaction', 'icx_sendTransaction(SCORE)', 'icx_call']

templates = {'IISS': {'setStake': {'method': 'icx_sendTransaction', 'params': {'method': 'setStake', 'params': {'value': ''}}}}, 'main_api': {'debug_estimateStep': {'params': {}}, 'debug_getTrace': {'params': {'txHash': ''}}, 'icx_call': {'params': ''}, 'icx_getBalance': {'params': {'address': ''}}, 'icx_getBlockByHash': {'params': {'hash': ''}}, 'icx_getBlockByHeight': {'params': {'height': ''}}, 'icx_getLastBlock': {}, 'icx_getNetworkInfo': {}, 'icx_getScoreApi': {'params': {'address': ''}}, 'icx_getScoreStatus': {'params': {'address': ''}}, 'icx_getTotalSupply': {}, 'icx_getTransactionByHash': {'params': {'txHash': ''}}, 'icx_getTransactionResult': {'params': {'txHash': ''}}, 'icx_sendTransaction': {'params': {'from': '', 'nid': '', 'nonce': '0x23', 'stepLimit': '', 'to': '', 'value': '', 'version': '0x3'}}, 'icx_sendTransaction(SCORE)': {'method': 'icx_sendTransaction'}}}

update_template(new_template)[source]

get_category()[source]

get_methods(category=None)[source]

create_rpc(params={}, method=None)[source]

is_required_sign()[source]

load_template()[source]

get_rpc(category=None, method=None, params=None)[source]

get_required_params()[source]

get_params_hint()[source]

class IconRpcHelper(url='', wallet=None, network_info=None, raise_on_failure=True, debug=False, required_sign_methods=None, wait_sleep=1, tx_method='icx_getTransactionResult', logger=None, margin_steps=0, verbose=0, use_hex_value=False, **kwargs)[source]

Bases: LoggerMixinVerbose

set_use_hex_value(value)[source]

test_logger()[source]

initialize()[source]

initialize_wallet(wallet=None)[source]

get_wallet_address(wallet=None)[source]

rpc_call(url=None, method=None, timeout=5000, params={}, payload={}, print_error=False, reset_error=True, raise_on_failure=False, store_request_payload=True, return_key=None, http_method='post')[source]

Return type:: dict

get_elapsed(mode='elapsed')[source]

get_total_elapsed()[source]

print_error_message(print_error=True)[source]

print_response(hex_to_int=False, message='')[source]

print_request(message='')[source]

make_params(method=None, params={})[source]

create_governance_payload(method, params={})[source]

governance_call(url=None, method=None, params={}, governance_address=None, sign=None, store_request_payload=True, is_wait=True, value='0x0', step_limit=None, return_key='result')[source]

create_deploy_payload(src='', params={}, governance_address=None)[source]

deploy_score(src='', params={}, step_limit=None, governance_address=None, is_wait=True, is_confirm_send=False)[source]

get_step_price()[source]

get_step_cost(step_kind='input')[source]

get_estimate_step(url=None, tx=None)[source]

get_step_limit(url=None, tx=None, step_kind=None)[source]

get_fee(tx=None, symbol=False, step_kind=None)[source]

get_score_api(address='', url=None)[source]

Return type:: list

static name_to_params(list_data)[source]

static make_params_hint(list_data)[source]

get_governance_api(url=None, return_method_only=False)[source]

get_balance(url=None, address=None, is_comma=False, is_tint=True, return_as_hex=False, return_as_decimal=False, use_hex_value=None)[source]

Retrieve the balance of the given address.

Parameters:

url (optional) – RPC URL. |default| None
address (optional) – Address to get the balance for. |default| None
is_comma (optional) – Whether to format the number with commas. |default| False
is_tint (optional) – Whether to return as TINT. |default| True
return_as_hex (optional) – Whether to return the balance in its raw hex form. |default| False
return_as_decimal (optional) – If True, returns the balance as a floating-point number in ICX. |default| False
use_hex_value (optional) – If True, returns a HexValue object instead of a raw balance. |default| None

Returns:

Balance in the requested format.

analyze_tx_block_time(transaction)[source]

get_block(block_height=None, return_key=None)[source]

get_stake(address=None, return_as_hex=False, return_key='result.stake', use_hex_value=None)[source]

get_bond(address=None, return_as_hex=False, return_key='result.bonds', use_hex_value=None)[source]

Return type:: list

get_delegation(address=None, return_as_hex=False, return_key='result.delegations', use_hex_value=None)[source]

get_iscore(address=None, return_as_hex=False, return_key='result.estimatedICX', use_hex_value=None)[source]

claim_iscore(url=None, step_limit=None, is_wait=True, return_key='result')[source]

unjail(url=None, step_limit=None, is_wait=True, return_key='result')[source]

handle_response_with_key(response=None, return_key=None)[source]

get_tx(tx_hash=None, url=None, return_key=None, use_hex_value=None)[source]

get_tx_wait(tx_hash=None, url=None, is_compact=True, is_block_time=False, max_attempts=None)[source]

check_transaction_loop(tx_hash=None, url=None, is_compact=True, status=None, is_block_time=False, max_attempts=None)[source]

Check the transaction loop until a result is received or max_attempts is reached.

Parameters:

tx_hash (optional) – The transaction hash. |default| None
url (optional) – The URL for the transaction. |default| None
is_compact (optional) – Whether to display a compact result. |default| True
status (optional) – Status object for logging progress. |default| None
is_block_time (optional) – Whether to include block elapsed time in the result. |default| False
max_attempts (optional) – Maximum number of attempts before exiting the loop (None for unlimited). |default| None

Returns:

The final response.

get_debug_trace(tx_hash, print_table=True, reset_error=True)[source]

parse_tx_var(tx=None)[source]

auto_fill_parameter(tx=None, is_force_from_addr=False)[source]

auto_fill_step_limit(tx=None, step_limit=None)[source]

sign_tx(wallet=None, payload=None, check_balance=True, step_limit=None)[source]

exit_on_failure(exception, force_exit=False)[source]

sign_send(is_wait=True, is_compact=False, is_block_time=False, max_attempts=None)[source]

preview_transaction_fee(payload, margin_steps=0, return_in_loop=False, use_decimal=False, use_hex_value=False)[source]

Estimate the transaction fee for a given payload.

Parameters:

payload (dict) – The transaction payload for which the fee is to be estimated.
margin_steps (int) – Additional steps to add to the estimated step limit (default: 0). |default| 0
return_in_loop (bool) – If True, returns the transaction fee in LOOP (smallest unit). Defaults to False (ICX). |default| False
use_decimal (bool) – If True, use Decimal for calculations to maintain precision. Defaults to False. |default| False
use_hex_value (optional) – If True, returns a HexValue object instead of a raw balance. |default| False

Returns:

Estimated transaction fee in ICX (or LOOP if return_in_loop is True).

Return type:

Union[float, Decimal, int]

Raises:

ValueError – If the payload is not properly formatted.

Example usage:

from pawnlib.uitils.http import IconRpcHelper

# Initialize the helper
icon_rpc_helper = IconRpcHelper(wallet="your_wallet_private_key")

# Prepare transaction payload
payload = {
    "from": "hx123...",
    "to": "hx456...",
    "value": "0x1bc16d674ec80000",  # Example value in LOOP (1 ICX)
    "stepLimit": "0xf4240"
}

# Preview the transaction fee in ICX
estimated_fee = icon_rpc_helper.preview_transaction_fee(payload, margin_steps=500, use_decimal=True)
print(f"Estimated Transaction Fee: {estimated_fee} ICX")

send_all_icx_with_decimal(to_address, fee=None, step_limit=None, min_balance=0, margin_steps=1000)[source]

Transfer all available ICX from the wallet to the specified address, accounting for transaction fees.

Parameters:

to_address (str) – The recipient address where the ICX will be sent.
fee (float or None) – Optional. The transaction fee in ICX. If not provided, it will be automatically calculated. |default| None
step_limit (int or None) – Optional. Custom step limit for the transaction. If not provided, it will be calculated. |default| None
min_balance (float) – Optional. The minimum balance to keep in the wallet after the transfer. Defaults to 0 ICX. |default| 0
margin_steps (int) – The additional margin to apply to the estimated step limit. Defaults to 1000. |default| 1000

Returns:

None. Sends all available ICX except for the transaction fee and minimum balance.

Return type:

None

Raises:

ValueError – If the calculated transfer amount is less than or equal to zero.
Exception – If the transaction fails.

Example usage:

from pawnlib.utils.http import IconRpcHelper

# Initialize the helper with the sender's wallet
icon_rpc_helper = IconRpcHelper(wallet="your_wallet_private_key")

# Transfer all ICX to the safety wallet
icon_rpc_helper.send_all_icx(
    to_address="hx8095412a43d07ed9869e55501c044849586ed671",
    fee=0.00125,  # Optional fee
    step_limit=100000,  # Optional step limit
    min_balance=0,  # Keep minimum balance of 0 ICX
    margin_steps=500  # Additional step margin for safety
)

send_all_icx(to_address, fee=None, step_limit=None, min_balance=0, margin_steps=0, max_attempts=None, dry_run=False)[source]

Transfer all available ICX from the wallet to the specified address, accounting for transaction fees. This version uses float instead of Decimal for comparison and calculations.

Parameters:

to_address (str) – The recipient address where the ICX will be sent.
fee (float or None) – Optional. The transaction fee in ICX. If not provided, it will be automatically calculated. |default| None
step_limit (int or None) – Optional. Custom step limit for the transaction. If not provided, it will be calculated. |default| None
min_balance (float) – Optional. The minimum balance to keep in the wallet after the transfer. Defaults to 0 ICX. |default| 0
margin_steps (int) – The additional margin to apply to the estimated step limit. Defaults to 0. |default| 0

Returns:

None. Sends all available ICX except for the transaction fee and minimum balance.

Return type:

None

Raises:

ValueError – If the calculated transfer amount is less than or equal to zero.
Exception – If the transaction fails.

Example usage:

from pawnlib.utils.http import IconRpcHelper

# Initialize the helper with the sender's wallet
icon_rpc_helper = IconRpcHelper(wallet="your_wallet_private_key")

# Transfer all ICX to the safety wallet
icon_rpc_helper.send_all_icx_without_decimal(
    to_address="hx8095412a43d07ed9869e55501c044849586ed671",
    fee=0.00125,  # Optional fee
    step_limit=100000,  # Optional step limit
    min_balance=0,  # Keep minimum balance of 0 ICX
    margin_steps=500  # Additional step margin for safety
)

set_stake(url=None, step_limit=None, is_wait=True, value='', return_key='result')[source]

delegate_all_icx(fee=None, step_limit=None, min_balance=0, margin_steps=0)[source]

stake_all_icx(fee=None, step_limit=None, min_balance=0, margin_steps=0, dry_run=False)[source]

Stake all available ICX, accounting for transaction fees, using float for calculations.

Parameters:

to_address (str) – The staking contract address where ICX will be staked.
fee (float or None) – Optional. The transaction fee in ICX. If not provided, it will be automatically calculated. |default| None
step_limit (int or None) – Optional. Custom step limit for the transaction. If not provided, it will be calculated. |default| None
min_balance (float) – Optional. The minimum balance to keep in the wallet after the transfer. Defaults to 0 ICX. |default| 0
margin_steps (int) – The additional margin to apply to the estimated step limit. Defaults to 1000. |default| 0

Returns:

None. Stakes all available ICX except for the transaction fee and minimum balance.

Return type:

None

Raises:

ValueError – If the calculated staking amount is less than or equal to zero.
Exception – If the transaction fails.

Example usage:

from icon_rpc_helper import IconRpcHelper

# Initialize the helper with the sender's wallet
icon_rpc_helper = IconRpcHelper(wallet="your_wallet_private_key")

# Stake all available ICX
icon_rpc_helper.stake_all_available_icx(
    to_address="hx8095412a43d07ed9869e55501c044849586ed671",
    fee=0.00125,  # Optional fee
    step_limit=100000,  # Optional step limit
    min_balance=0,  # Keep minimum balance of 0 ICX
    margin_steps=500  # Additional step margin for safety
)

class JsonRequest[source]

Bases: object

TODO: It will be generated the JSON or JSON-RPC request

__init__()[source]: TODO: It will be generated the JSON or JSON-RPC request

disable_ssl_warnings()[source]

append_scheme(url, default_scheme='http')[source]

Appends the specified scheme (e.g., http, https) to the URL if it is missing.

Parameters:

url (str) – The URL to validate and modify.
default_scheme (str, optional) – The scheme to prepend if missing. Default is “http”. |default| 'http'

Return type:

str

Returns:

A URL with a scheme.

append_http(url)[source]

Add http:// if it doesn’t exist in the URL

Parameters:: url –
Returns:

append_s3(url)[source]

Add http:// if it doesn’t exist in the URL

Parameters:: url –
Returns:

append_ws(url)[source]

Add ws:// if it doesn’t exist in the URL

Parameters:: url –
Returns:

append_api_v3(url)[source]

remove_http(url)[source]

Remove the r’https?://’ string

Parameters:: url –
Returns:

get_operator_truth(inp, relate, cut)[source]

guess_key(find_key, data)[source]

class SuccessCriteria(target='', operator='', expected='')[source]

Bases: object

to_dict()[source]

class SuccessResponse(target_key='', operator='', expected='', target={})[source]: Bases: SuccessCriteria

class CallHttp(url=None, method='get', payload={}, timeout=3000, ignore_ssl=False, verify=False, verbose=0, success_criteria='__DEFAULT__', success_operator='and', success_syntax='auto', raise_on_failure=False, auto_run=True, **kwargs)[source]

Bases: object

SHORTEN_MESSAGE_DICT = {<class 'requests.exceptions.Timeout'>: {'message': 'Timeout Error', 'params_message': 'timeout={}'}, <class 'requests.exceptions.HTTPError'>: 'HTTP Error', <class 'requests.exceptions.ConnectionError'>: 'DNS lookup Error', <class 'requests.exceptions.RequestException'>: 'OOps: Something Else'}

exit_on_failure(exception)[source]

run()[source]

get_response()[source]

Return type:: HttpResponse

print_http_response(response=None)[source]

fetch_response()[source]

fetch_criteria(success_criteria=None, success_operator='and')[source]

is_success()[source]

class HttpInspect(url, method='GET', headers=None, auth=None, timeout=10, max_redirects=5, verify=True, data=None, max_response_length=700, output=None, dns_server=None, debug=False)[source]

Bases: object

Initialize the HttpInspect class for HTTP request inspection.

This class provides functionality to inspect HTTP requests and responses, including DNS resolution, timing analysis, and response handling.

Parameters:

url (str) – The target URL to inspect. Will be prefixed with ‘http://’ if no scheme is provided.
method (str, optional) – HTTP method to use. Defaults to ‘GET’. |default| 'GET'
headers (dict, optional) – Custom HTTP headers. Defaults to None. |default| None
auth (tuple, optional) – Authentication credentials. Defaults to None. |default| None
timeout (int, optional) – Request timeout in seconds. Defaults to 10. |default| 10
max_redirects (int, optional) – Maximum number of redirects to follow. Defaults to 5. |default| 5
verify (bool, optional) – Whether to verify SSL certificates. Defaults to True. |default| True
data (dict, optional) – Request body data. Defaults to None. |default| None
max_response_length (int, optional) – Maximum length of response to display. Defaults to 700. |default| 700
output (str, optional) – Output format specification. Defaults to None. |default| None
dns_server (str, optional) – DNS server to use. Defaults to None. |default| None
debug (bool, optional) – Whether to enable debug mode. Defaults to False. |default| False

__init__(url, method='GET', headers=None, auth=None, timeout=10, max_redirects=5, verify=True, data=None, max_response_length=700, output=None, dns_server=None, debug=False)[source]

Initialize the HttpInspect class for HTTP request inspection.

This class provides functionality to inspect HTTP requests and responses, including DNS resolution, timing analysis, and response handling.

Parameters:

url (str) – The target URL to inspect. Will be prefixed with ‘http://’ if no scheme is provided.
method (str, optional) – HTTP method to use. Defaults to ‘GET’. |default| 'GET'
headers (dict, optional) – Custom HTTP headers. Defaults to None. |default| None
auth (tuple, optional) – Authentication credentials. Defaults to None. |default| None
timeout (int, optional) – Request timeout in seconds. Defaults to 10. |default| 10
max_redirects (int, optional) – Maximum number of redirects to follow. Defaults to 5. |default| 5
verify (bool, optional) – Whether to verify SSL certificates. Defaults to True. |default| True
data (dict, optional) – Request body data. Defaults to None. |default| None
max_response_length (int, optional) – Maximum length of response to display. Defaults to 700. |default| 700
output (str, optional) – Output format specification. Defaults to None. |default| None
dns_server (str, optional) – DNS server to use. Defaults to None. |default| None
debug (bool, optional) – Whether to enable debug mode. Defaults to False. |default| False

get_ip_address(domain='', url='')[source]

Get the IP address of the specified domain.

Parameters:

domain (str, optional) – The domain name to lookup. Defaults to the domain extracted from the current URL. |default| ''
url (str, optional) – The URL to include in timing information. Defaults to the current URL. |default| ''

Returns:

The resolved IP address or None if lookup fails

Return type:

Optional[str]

get_dns_records(domain, record_types=None)[source]

Get the DNS records for the specified domain.

Parameters:

domain (str) – The domain name to lookup.
record_types (List[str], optional) – The list of record types to lookup. Defaults to A, AAAA, CNAME, MX, TXT, NS. |default| None

Return type:

None

make_http_request()[source]

Perform an HTTP request and analyze the response.

Returns:: Whether the request was successful
Return type:: bool

display_results()[source]: HTTP 요청 결과를 표시합니다.

display_dns_records()[source]: Display the DNS records.

export_results(format='json', filename=None)[source]

Export the results to a file.

Parameters:

format (str) – The format to export (‘json’, ‘text’) |default| 'json'
filename (str, optional) – The file name. Defaults to ‘{domain}_results.{format}’ |default| None

run()[source]: 전체 검사 프로세스를 실행합니다.

parse_auth(auth_str)[source]

parse_headers(headers_str)[source]

class CheckSSL(host, port=443, timeout=5.0, sni_hostname='')[source]

Bases: object

SSL Certificate Checker.

This class is responsible for checking the SSL certificate of a given host. It retrieves SSL certificate information, checks its expiry status, and displays the relevant details in a formatted table.

host

The hostname of the server to check.

Type:: str

port

The port number to connect to (default is 443).

Type:: int

timeout

The timeout for the connection in seconds (default is 5.0).

Type:: float

sni_hostname

The hostname to use for the SNI (Server Name Indication) handshake.

Type:: str

ssl_info

Stores the SSL certificate information.

Type:: dict

get_ssl()[source]

Retrieve the SSL certificate for the specified host.

Returns:: The SSL certificate information.
Return type:: dict
Raises:: SystemExit – If there is a connection error.

check_expiry()[source]

Check the expiry status of the SSL certificate.

Returns:: A status string and the number of days left until expiry.
Return type:: tuple

display()[source]: Display the SSL certificate information in a formatted table.

class CallWebsocket(url, verbose=0, timeout=10, on_send=None, on_receive=None, enable_status_console=False, ssl_options=None, headers=None, ping_interval=20, max_retries=3, max_fail_count=10, logger=None)[source]

Bases: object

Initialize the CallWebsocket.

Parameters:

url (str) – WebSocket URL for connection.
verbose (int, optional) – Level of verbosity for logging. |default| 0
timeout (int, optional) – Timeout duration for the WebSocket connection. |default| 10
on_send (Optional[Callable[..., Any]], optional) – Callable function for sending messages. |default| None
on_receive (Optional[Callable[..., Any]], optional) – Callable function for receiving messages. |default| None
enable_status_console (bool, optional) – If True, status console will be enabled. |default| False
ssl_options (optional) – SSL options for secure WebSocket connections. |default| None
headers (optional) – Optional headers for WebSocket connection. |default| None
ping_interval (int, optional) – Interval in seconds for sending ping messages. |default| 20
max_retries (int, optional) – Number of retries for a single reconnection attempt. |default| 3
max_fail_count (int, optional) – Maximum allowed total failure count before giving up. |default| 10

__init__(url, verbose=0, timeout=10, on_send=None, on_receive=None, enable_status_console=False, ssl_options=None, headers=None, ping_interval=20, max_retries=3, max_fail_count=10, logger=None)[source]

Initialize the CallWebsocket.

Parameters:

url (str) – WebSocket URL for connection.
verbose (int, optional) – Level of verbosity for logging. |default| 0
timeout (int, optional) – Timeout duration for the WebSocket connection. |default| 10
on_send (Optional[Callable[..., Any]], optional) – Callable function for sending messages. |default| None
on_receive (Optional[Callable[..., Any]], optional) – Callable function for receiving messages. |default| None
enable_status_console (bool, optional) – If True, status console will be enabled. |default| False
ssl_options (optional) – SSL options for secure WebSocket connections. |default| None
headers (optional) – Optional headers for WebSocket connection. |default| None
ping_interval (int, optional) – Interval in seconds for sending ping messages. |default| 20
max_retries (int, optional) – Number of retries for a single reconnection attempt. |default| 3
max_fail_count (int, optional) – Maximum allowed total failure count before giving up. |default| 10

connect(api_path='')[source]

Establish a WebSocket connection.

Parameters:: api_path (str, optional) – Additional API path to append to the WebSocket URL. |default| ''

reconnect(api_path='')[source]

Attempt to reconnect to the WebSocket server.

Parameters:: api_path (str, optional) – API path for WebSocket reconnection. |default| ''

run(api_path='/api/v3/icon_dex/block', use_status_console=False)[source]

Start the WebSocket communication.

Parameters:

api_path (str, optional) – API path for the WebSocket connection. |default| '/api/v3/icon_dex/block'
use_status_console (bool, optional) – Enable status console during the communication. |default| False

close()[source]: Close the WebSocket connection.

class GoloopWebsocket(url, verbose=0, timeout=10, blockheight=0, sec_thresholds=4, monitoring_target=None, ignore_ssl=True, network_info=None, on_send=None, on_receive=None, logger=None)[source]

Bases: CallWebsocket

Initialize the CallWebsocket.

Parameters:

url – WebSocket URL for connection.
verbose (optional) – Level of verbosity for logging. |default| 0
timeout (optional) – Timeout duration for the WebSocket connection. |default| 10
on_send (Optional[Callable[..., Any]], optional) – Callable function for sending messages. |default| None
on_receive (Optional[Callable[..., Any]], optional) – Callable function for receiving messages. |default| None
enable_status_console – If True, status console will be enabled.
ssl_options – SSL options for secure WebSocket connections.
headers – Optional headers for WebSocket connection.
ping_interval – Interval in seconds for sending ping messages.
max_retries – Number of retries for a single reconnection attempt.
max_fail_count – Maximum allowed total failure count before giving up.

request_blockheight_callback()[source]

parse_transfer_tx(confirmed_transaction_list=[])[source]

Parse a list of confirmed transactions, store transfer details, and log them.

Parameters:: confirmed_transaction_list (optional) – A list of confirmed transactions. |default| []
Returns:: A list of parsed transfer details.

parse_blockheight(response=None)[source]

output_format(key_string='', diff_time=0, is_string=False)[source]

get_response_content()[source]

get_last_blockheight()[source]

Return type:: int

get_preps()[source]

get_validator_info()[source]

fetch_and_store_preps_info()[source]

execute_rpc_call(method=None, params={}, return_key=None, governance_address=None)[source]

get_block_hash(tx_hash)[source]

get_tx_result(tx_hash, max_attempts=None)[source]

class AsyncCallWebsocket(url, verbose=0, timeout=10, on_send=None, on_receive=None, ssl_options=None, headers=None, ping_interval=20, max_retries=3, max_fail_count=10, enable_status_console=False, logger=None, session=None, additional_tasks=None, **kwargs)[source]

Bases: LoggerMixinVerbose

async connect(api_path='')[source]

async reconnect(api_path='')[source]

async run_tasks()[source]

async run(api_path='/api/v3/icon_dex/block')[source]

async close(exit_on_close=True, exit_code=0)[source]

async graceful_close(exit_code=0)[source]

Gracefully close the event loop and clean up resources before exit. :type exit_code: optional :param exit_code: Exit code to return when exiting the system (default is 0).

|default| 0

class AsyncGoloopWebsocket(url, verbose=0, timeout=10, blockheight=0, sec_thresholds=4, monitoring_target=None, ignore_ssl=True, network_info=None, on_send=None, on_receive=None, logger=None, process_transaction=None, address_filter=None, send_slack=True, slack_webhook_url='', max_retries=3, max_transaction_attempts=10, check_tx_result_enabled=True, ignore_data_types=None, session=None, preps_refresh_interval=600, use_shorten_tx_hash=True, bps_interval=0, skip_until=0, base_dir='./')[source]

Bases: AsyncCallWebsocket

BLOCKHEIGHT_FILE = 'last_blockheight.txt'

SLACK_BLOCKHEIGHT_FILE = 'last_slack_blockheight.txt'

read_last_processed_blockheight(filename, skip_log=False)[source]: Read the last processed block height from a file.

write_last_processed_blockheight(filename, blockheight)[source]: Write the last successfully processed block height to a file.

async run_from_blockheight(blockheight=None, api_path='/api/v3/icon_dex/block')[source]

Run starting from a specified block height, the latest block, or the last recorded block height.

Parameters:: blockheight (optional) – The block height to start from. - If None, tries to resume from the last saved block. - If 0, starts from the latest block. - Otherwise, starts from the provided block height. |default| None

async initialize(session=None)[source]

async periodic_preps_update()[source]

async request_blockheight_callback()[source]

async fetch_and_store_preps_info()[source]

async parse_blockheight(response=None, delay=2)[source]

Parse block height with retry logic.

Parameters:

response (optional) – The response containing block height data. |default| None
delay (optional) – Delay (in seconds) between retry attempts. |default| 2

async calculate_tps_bps(block_height, tx_count)[source]: 10초 단위로 TPS 및 BPS 계산

async handle_confirmed_transaction_list(response)[source]

validate_address_filter(address_filter)[source]

log_invalid_addresses(invalid_addresses)[source]

highlight_address(address)[source]: Highlights the address if it’s part of the address_filter.

get_prep_info(address=None, key='name', apply_format=True)[source]

Retrieves the requested information from preps_info for a given address.

Parameters:

address (optional) – The address to lookup in preps_info. |default| None
key (optional) – The key for the desired value (default is ‘name’). |default| 'name'
apply_format (optional) – If True, applies format_text to the returned value for Slack formatting (default is False). |default| True

Returns:

A formatted string with the value in parentheses, or an empty string if not found.

async default_process_transaction(tx, block_height)[source]: Process the individual transactions.

static formated_icx_value(value)[source]

get_stake_value(tx_data, is_tint=True)[source]

get_send_value(tx_data, is_tint=True)[source]

async log_message(message, slack_additional_message='', level='info', stack_level=4, block_height=None)[source]

Logs a message using the specified level and optionally sends to Slack.

Parameters:

message – The message to log.
slack_additional_message (optional) – Additional message to send along with Slack (optional). |default| ''
level (optional) – The logging level (e.g., ‘info’, ‘error’, ‘debug’). |default| 'info'
stack_level (optional) – stack level |default| 4
block_height (optional) – block height |default| None

format_message(message, for_console=True)[source]

Formats a message for either console or Slack.

Parameters:

message (str) – The original message.
for_console (bool, optional) – True if formatting for console, False for Slack. |default| True

Returns:

Formatted message.

static get_slack_color_by_level(level)[source]

Returns a color code for Slack messages based on log level.

Parameters:: level – The log level (e.g., ‘warn’, ‘error’, ‘info’, ‘critical’).
Returns:: Hex color code as a string.

async send_slack_notification(message, level='info')[source]

Send a message to Slack using attachments to add color and timestamp.

Parameters:

message – The message to send to Slack.
level (optional) – The log level (e.g., ‘warn’, ‘error’, ‘info’). |default| 'info'

class AsyncIconRpcHelper(url='', logger=None, session=None, verbose=True, timeout=10, retries=3, return_with_time=False, max_concurrency=20, **kwargs)[source]

Bases: LoggerMixinVerbose

A helper class for making asynchronous RPC calls to the ICON network. Provides methods to interact with the ICON blockchain, such as fetching blocks, transactions, and network information.

url

The base URL for the RPC endpoint.

Type:: str

logger

Logger instance for logging.

Type:: Optional[Union[logging.Logger, Console, ConsoleLoggerAdapter, Null]]

session

HTTP session for making requests.

Type:: aiohttp.ClientSession

verbose

Flag to enable verbose logging.

Type:: bool

timeout

Request timeout in seconds.

Type:: int

retries

Number of retry attempts for failed requests.

Type:: int

return_with_time

Whether to return elapsed time with responses.

Type:: bool

max_concurrency

Number of max concurrency

Type:: int

initialize()[source]: Initializes the aiohttp session if not already initialized.

close()[source]: Closes the aiohttp session.

execute_rpc_call()[source]: Executes an RPC call to the ICON network.

fetch()[source]: Makes a generic HTTP request (GET or POST).

get_block_hash()[source]: Fetches block information by hash.

get_last_blockheight()[source]: Retrieves the height of the last block on the blockchain.

get_network_info()[source]: Fetches network information from the ICON blockchain.

get_preps()[source]: Retrieves a list of P-Reps from the ICON network.

get_validator_info()[source]: Retrieves validator information from the ICON network.

get_tx_result()[source]: Fetches transaction results, with optional retries and waiting.

Example

import asyncio
from pawnlib.utils.http import AsyncIconRpcHelper

async def main():
    async with AsyncIconRpcHelper(url="https://icon-node-url.com") as rpc_helper:
        # Fetch last block height
        last_height = await rpc_helper.get_last_blockheight()
        print(f"Last Block Height: {last_height}")

        # Get block hash by transaction hash
        block_hash = await rpc_helper.get_block_hash(tx_hash="0x1234...")
        print(f"Block Hash: {block_hash}")

        # Fetch P-Reps
        preps = await rpc_helper.get_preps()
        print(f"P-Reps: {preps}")

asyncio.run(main())

async adjust_concurrency(new_max)[source]

Dynamically adjusts the maximum concurrency limit for asynchronous requests.

Updates both the semaphore and TCP connector limits while migrating existing tasks to the new concurrency configuration.

Parameters:: new_max (int) – New maximum number of concurrent connections (must be ≥1)
Raises:: ValueError – If new_max is less than 1

property concurrency_usage

Monitors current concurrency utilization.

Returns:

Dictionary containing concurrency metrics:

active (int): Number of currently used connections
available (int): Number of available connections
max (int): Maximum allowed concurrent connections

Return type:

dict

async close()[source]

async initialize()[source]

async execute_rpc_call(method=None, params={}, url=None, return_key=None, governance_address=None, return_on_error=True, keep_lists=True)[source]: Execute an RPC call to the ICON network.

async fetch(path='', data='', http_method='get', url='', headers=None, return_key=None, return_on_error=True, return_first=False, list_index=None, retries=None)[source]

check_aio_session(log_exception=True, return_on_error=False)[source]

static handle_response_with_key(response=None, return_key=None, keep_lists=True)[source]

async get_block_hash(tx_hash='', url=None)[source]

Return type:: dict

async get_last_blockheight(url=None)[source]

Return type:: int

async get_network_info(url=None)[source]

Return type:: int

async get_preps(url=None, return_dict_key='')[source]

async get_balance(address='', return_as_hex=True, return_key='result', use_hex_value=None, url=None)[source]

async get_stake(address='', return_as_hex=False, return_key='result', use_hex_value=None, url=None)[source]

async get_delegation(address='', return_as_hex=False, return_key='result', use_hex_value=None, url=None)[source]

async get_bond(address='', return_as_hex=False, return_key='result', use_hex_value=None, url=None)[source]

async get_iscore(address='', return_as_hex=False, return_key='result', use_hex_value=None, url=None)[source]

async get_node_name_by_address(url=None)[source]

Retrieves a dictionary mapping node addresses to their corresponding names.

This function calls get_preps to fetch the full dictionary of P-Rep information and then filters it to create a new dictionary where the keys are nodeAddress and the values are name.

Returns:: A dictionary with nodeAddress as keys and name as values.
Return type:: dict

Example

# Example output
{
    "hx12ffd8a005f9bc0a3164c2d133a0ed5ecfe70c21": "Clue",
    "hx34a8e8a005f9bc0b3164c2d133a0ed5ecfe70c22": "NodeX",
    ...
}

async get_validator_info(url=None, return_dict_key='')[source]

async get_tx_result(tx_hash, max_attempts=5, is_wait=True, return_key='result', url=None)[source]

gen_rpc_params(method=None, params=None)[source]

getBlockByHash(nodeHost, hash)[source]

getTransactionByHash(nodeHost, hash)[source]

getLastBlock(nodeHost)[source]

jequest(url, method='get', payload={}, elapsed=False, print_error=False, timeout=None, ipaddr=None, verify=True, **kwargs)[source]

This functions will be called the http requests.

Parameters:

url –
method (optional) – |default| 'get'
payload (optional) – |default| {}
elapsed (optional) – |default| False
print_error (optional) – |default| False
timeout (optional) – Timeout seconds |default| None
ipaddr (optional) – Change the request IP address in http request |default| None
verify (optional) – verify SSL |default| True
**kwargs – Optional arguments that request takes.

Return type:

dict

Returns:

async get_blockheight(response=None)[source]

Function to parse block height from a response.

Parameters:: response (optional) – The response containing block height data. |default| None
Returns:: Parsed block height.
Raises:: Exception if block height parsing fails.

async retry_operation(operation, max_attempts=3, delay=2, success_criteria=None, logger=None, verbose=0, *args, **kwargs)[source]

Retry the given operation multiple times in case of failure or based on the success criteria.

Parameters:

operation – The operation to retry (function).
max_attempts (optional) – Maximum number of retry attempts. |default| 3
delay (optional) – Delay between retries. |default| 2
success_criteria (optional) – A function that takes the result and returns True if the result is valid/successful. |default| None
logger (optional) – Logger for logging messages (optional). |default| None
verbose (optional) – Verbosity level for logging. |default| 0
args – Positional arguments to pass to the operation.
kwargs – Keyword arguments to pass to the operation.

Returns:

The result of the operation if successful.

Raises:

Exception if the operation fails after max_attempts or success criteria is not met.

remove_path_from_url(url)[source]

Removes the path, query, and fragment from a URL, leaving only the scheme and domain.

Parameters:: url (str) – The original URL.
Returns:: The URL without the path, query, or fragment.
Return type:: str

icon_rpc_call(url=None, method=None, timeout=5000, params={}, payload={}, print_error=False, reset_error=True, raise_on_failure=False, store_request_payload=True, return_key=None, http_method='post')

Return type:: dict

pawnlib.utils.log module

from pawnlib.utils import log

class CustomLog(name)[source]

Bases: object

Parameters:: name – logger name

Example

from pawnlib.utils.log import CustomLog

file_name = './time_log.txt'
logger = CustomLog("custom_log")
logger.set_level('DEBUG')
logger.stream_handler("INFO")
logger.time_rotate_handler(filename=file_name,
                           when="M",
                           interval=2,
                           backup_count=3,
                           level="INFO"
                           )
idx = 1
logger.log.debug(logger.log_formatter(f'debug {idx}'))
logger.log.info(logger.log_formatter(f'info {idx}'))
logger.log.warning(logger.log_formatter(f'warning {idx}'))
logger.log.error(logger.log_formatter(f'error {idx}'))
logger.log.critical(logger.log_formatter(f'critical {idx}'))

set_level(level)[source]

log_formatter(msg)[source]

Parameters:: msg –
Returns:

stream_handler(level)[source]

Parameters:: level –

Note

level

“DEBUG” : logging.DEBUG ,
“INFO” : logging.INFO ,
“WARNING” : logging.WARNING ,
“ERROR” : logging.ERROR ,
“CRITICAL” : logging.CRITICAL ,

Returns:

file_handler(file_name, mode)[source]

Parameters:

file_name – ~.txt / ~.log
mode – “w” / “a”

Returns:

file_rotating_handler(file_name, mode, level, backup_count, log_max_size)[source]

Parameters:

file_name – file의 이름 , ~.txt / ~.log
mode – “w” / “a”
backup_count – backup할 파일 개수
log_max_size – 한 파일당 용량 최대
level –

> “DEBUG” : logging.DEBUG , > “INFO” : logging.INFO , > “WARNING” : logging.WARNING , > “ERROR” : logging.ERROR , > “CRITICAL” : logging.CRITICAL , :return:

time_rotate_handler(filename='./log.txt', when='M', level='DEBUG', backup_count=4, atTime=datetime.time(0, 0), interval=1)[source]

Parameters:

level (optional) – |default| 'DEBUG'
filename (optional) – |default| './log.txt'
when (optional) – 저장 주기 |default| 'M'
interval (optional) – 저장 주기에서 어떤 간격으로 저장할지 |default| 1
backup_count (optional) – 5 |default| 4
atTime (optional) – datetime.time(0, 0, 0) |default| datetime.time(0, 0)

Returns:

class AppLogger(app_name='default', log_level='INFO', log_path='./logs', markup=True, stdout=False, stdout_level='INFO', stdout_log_formatter='%H:%M:%S,%f', log_format=None, std_log_format=None, debug=False, use_hook_exception=True, use_clean_text_filter=False, exception_handler='', **kwargs)[source]

Bases: object

A logger class for managing application logging.

This logger supports logging to both a file and stdout with customizable formats, levels, and handlers. It can also handle exceptions and apply filters to log messages.

Parameters:

app_name (str, optional) – The name of the application, which will be used as the log file name. |default| 'default'
log_level (Literal['INFO', 'WARN', 'DEBUG'], optional) – The logging level (default is “INFO”). Options include: “DEBUG”, “INFO”, “WARN”, “ERROR”. |default| 'INFO'
log_path (str, optional) – The directory path where log files will be stored (default is “./logs”). |default| './logs'
stdout (bool, optional) – If True, enables logging to stdout (default is False). |default| False
markup (bool, optional) – If True, enables markup formatting for stdout logging (default is False). |default| True
stdout_level (Literal['INFO', 'WARN', 'DEBUG', 'NOTSET'], optional) – The logging level for stdout (default is “INFO”). Options include “DEBUG”, “INFO”, “WARN”, “ERROR”, “NOTSET”. |default| 'INFO'
stdout_log_formatter (Callable, optional) – Custom formatter for stdout logging (default is None). |default| '%H:%M:%S,%f'
log_format (Optional[str], optional) – Custom format for log messages (default is a predefined format). Example: “[%(asctime)s] %(levelname)s - %(message)s”. |default| None
use_hook_exception (bool, optional) – If True, sets a hook to log uncaught exceptions (default is True). |default| True
exception_handler (Callable, optional) – A custom function to handle exceptions (default is None). |default| ''
debug (bool, optional) – If True, enables debug mode for additional logging information (default is False). |default| False

Example Usage:

from pawnlib.utils import log

app_logger, error_logger = log.AppLogger().get_logger() app_logger.info(“This is an info message.”) error_logger.error(“This is an error message.”)

# Advanced usage with configuration from pawnlib.config.globalconfig import pawnlib_config as pawn

app_logger, error_logger = log.AppLogger(
app_name=”app”, log_path=”./logs”, stdout=True, markup=True, log_level=”DEBUG”

).set_global()

pawn.app_logger.info(“This is an info message.”) pawn.error_logger.error(“This is an error message.”)

# Expected Output: # [2022-07-25 18:52:44,415] INFO::app_logging_test.py/main(38) This is an info message. # [2022-07-25 18:52:44,416] ERROR::app_logging_test.py/main(39) This is an error message.

logger: The main application logger instance.

error_logger: The error logger instance for capturing error messages.

get_logger()[source]: Returns the main and error logger instances.

set_global()[source]: Sets the logger instances globally in the pawnlib configuration.

Note

Ensure that the specified log path exists or can be created by the application.

get_realpath()[source]

set_logger(log_type='INFO')[source]

add_stream_handler(level)[source]

time_rotate_handler(filename='./log.txt', when='M', backup_count=4, atTime=datetime.time(0, 0), interval=1, encoding='utf-8')[source]

get_logger()[source]

Get the logger

Returns:

set_global()[source]

Add global config in pawnlib

Returns:

handle_exception(exc_type, exc_value, exc_traceback)[source]

class CleanTextFilter(name='')[source]

Bases: Filter

Initialize a filter.

Initialize with the name of the logger which, together with its children, will have its events allowed through the filter. If no name is specified, allow every event.

filter(record)[source]

Determine if the specified record is to be logged.

Returns True if the record should be logged, or False otherwise. If deemed appropriate, the record may be modified in-place.

class TightLevelRichHandler(level=0, console=None, *, show_time=True, omit_repeated_times=True, show_level=True, show_path=True, enable_link_path=True, highlighter=None, markup=False, rich_tracebacks=False, tracebacks_width=None, tracebacks_code_width=88, tracebacks_extra_lines=3, tracebacks_theme=None, tracebacks_word_wrap=True, tracebacks_show_locals=False, tracebacks_suppress=(), tracebacks_max_frames=100, locals_max_length=10, locals_max_string=80, log_time_format='[%x %X]', keywords=None)[source]

Bases: RichHandler

Initializes the instance - basically setting the formatter to None and the filter list to empty.

get_level_text(record)[source]

Get the level name from the record.

Parameters:: record (LogRecord) – LogRecord instance.
Returns:: A tuple of the style and level name.
Return type:: Text

print_logger_configurations(min_level='TRACE')[source]

list_all_loggers()[source]

pawnlib.utils.notify module

from pawnlib.utils import notify

class TelegramBot(bot_token=None, chat_id=None, verify_ssl=True, ignore_ssl_warning=False, async_mode=False, max_retries=5, retry_delay=5)[source]

Bases: object

A class to interact with the Telegram Bot API.

Parameters:

bot_token (optional) – Telegram bot token. If not provided, it will be fetched from the ‘TELEGRAM_BOT_TOKEN’ environment variable. |default| None
chat_id (optional) – Chat ID to send messages to. If not provided, it will be fetched from the ‘TELEGRAM_CHAT_ID’ environment variable or determined dynamically. |default| None
verify_ssl (optional) – Whether to verify the SSL certificate for HTTPS requests. (default: True). |default| True
ignore_ssl_warning (optional) – Whether to ignore SSL warnings. If True, SSL warnings will be suppressed. |default| False
async_mode (optional) – Whether to use asynchronous mode. If False, synchronous mode will be used. (default: True) |default| False
max_retries (optional) – Maximum number of retries for 429 Too Many Requests errors. (default: 5) |default| 5
retry_delay (optional) – Delay in seconds between retries for 429 Too Many Requests errors. (default: 5) |default| 5

Raises:

ValueError – If the bot token is not provided either as an argument or an environment variable.

Example

bot = TelegramBot(bot_token="your_bot_token", chat_id="your_chat_id", async_mode=False)
bot.send_message("Hello, world!")
bot.send_html_message("<b>Hello, world!</b>")
bot.send_plain_text_message("Just plain text.")
bot.send_dict_message({"key": "value"})

static escape_markdown(text)[source]

Escape Markdown special characters for MarkdownV2.

Parameters:: text – The text to escape.
Returns:: The escaped text.

Example

bot = TelegramBot(bot_token="your_bot_token", chat_id="your_chat_id")
escaped_text = bot.escape_markdown("Hello *world*!")
print(escaped_text)  # Output: Hello \*world\*!

async send_multiple_messages_async(messages)[source]

Send multiple messages asynchronously.

Parameters:: messages – A list of messages to send.
Returns:: A list of responses from the Telegram API.

send_multiple_messages(messages)[source]

Send multiple messages synchronously.

Parameters:: messages – A list of messages to send.
Returns:: A list of responses from the Telegram API.

send_message(message, parse_mode='Markdown', disable_web_page_preview=False)[source]

build_payload(message, parse_mode='Markdown', disable_web_page_preview=False, pass_escape=False)[source]: Build the payload for sending a message.

send_message_sync(message, parse_mode='Markdown', disable_web_page_preview=False, pass_escape=False)[source]

async send_message_async(message, parse_mode='MarkdownV2', disable_web_page_preview=False, pass_escape=False)[source]

send_html_message(message)[source]

async send_html_message_async(message)[source]

send_plain_text_message(message, parse_mode='Markdown')[source]

async send_plain_text_message_async(message, parse_mode='Markdown')[source]

send_dict_message(message_dict)[source]

async send_dict_message_async(message_dict)[source]

get_chat_id()[source]: Retrieve chat_id by getting updates from the Telegram bot API

display_all_chat_updates()[source]: Retrieve all unique chat_ids and their details by getting updates from the Telegram bot API

extract_chat_info(update)[source]: Extract chat information from the update.

save_chat_id(chat_id_file='chat_id.txt')[source]

load_chat_id(chat_id_file='chat_id.txt')[source]

send_auto_message(message)[source]

async send_auto_message_async(message)[source]

is_html(message)[source]

is_markdown(message)[source]: Check if the string contains MarkdownV2 special characters

class SlackNotifier(webhook_url=None, username='PawnBot', icon_emoji=':robot_face:', retries=3, retry_delay=2, verbose=1, logger=None)[source]

Bases: LoggerMixinVerbose

Class for sending Slack messages. Provides both synchronous and asynchronous methods.

Parameters:

webhook_url (Optional[str], optional) – Slack webhook URL. If not provided, it will be fetched from the environment variable ‘SLACK_WEBHOOK_URL’. |default| None
username (str, optional) – Username to display in Slack. |default| 'PawnBot'
icon_emoji (str, optional) – Emoji to use as the icon for the message. |default| ':robot_face:'
retries (int, optional) – Number of retry attempts in case of failure. |default| 3
retry_delay (int, optional) – Time to wait between retries (in seconds). |default| 2

create_message_payload(message, title='', text='', msg_level='info', status=None, simple_mode=False, footer='', timestamp_format=None, max_text_length=1000)[source]

Create message payload.

Return type:: Dict[str, Any]

send(message, title='', msg_level='info', status=None, simple_mode=False, footer='', timestamp_format=None, text='')[source]

Synchronously send a Slack message.

Parameters:

message (Union[str, Dict, List]) – The message to send (string, dictionary, list)
title (str, optional) – Message title |default| ''
msg_level (str, optional) – Message severity level (info, warn, error, critical, etc.) |default| 'info'
status (Union[str, StatusType, None], optional) – Status type or string for dynamic emoji and message formatting. |default| None
simple_mode (bool, optional) – If True, send a simplified message without additional info. |default| False
footer (str, optional) – Footer text to display at the bottom of the message. |default| ''
timestamp_format (Optional[str], optional) – Time display format (e.g. “%Y-%m-%d %H:%M:%S”) |default| None

Return type:

bool

Returns:

Boolean indicating success or failure.

async send_async(message, title='', msg_level='info', status=None, simple_mode=False, footer='', timestamp_format=None, text='')[source]

Asynchronously send a Slack message.

Parameters:

message (Union[str, Dict, List]) – The message to send (string, dictionary, list)
title (str, optional) – Message title |default| ''
msg_level (str, optional) – Message severity level (info, warn, error, critical, etc.) |default| 'info'
status (Union[str, StatusType, None], optional) – Status type or string for dynamic emoji and message formatting. |default| None
simple_mode (bool, optional) – If True, send a simplified message without additional info. |default| False
footer (str, optional) – Footer text to display at the bottom of the message. |default| ''
timestamp_format (Optional[str], optional) – Time display format (e.g. “%Y-%m-%d %H:%:M:%S”) |default| None

Return type:

bool

Returns:

Boolean indicating success or failure.

send_batch(messages)[source]

Send multiple messages in batch (synchronous)

Parameters:: messages (List[Dict]) – List of messages to send [{“message”: “content”, “title”: “title”, …}, …]
Return type:: List[bool]
Returns:: List of results for each message

async send_batch_async(messages)[source]

Send multiple messages in batch (asynchronous)

Parameters:: messages (List[Dict]) – List of messages to send [{“message”: “content”, “title”: “title”, …}, …]
Return type:: List[bool]
Returns:: List of results for each message

send_error(error, additional_info=None, title='Error occurred')[source]

Send error information

Parameters:

error (Exception) – The exception object that occurred
additional_info (Optional[Dict], optional) – Additional information dictionary |default| None
title (str, optional) – Message title |default| 'Error occurred'

Returns:

Boolean indicating success or failure

async send_error_async(error, additional_info=None, title='Error occurred')[source]

Send error information asynchronously

Parameters:

error (Exception) – The exception object that occurred
additional_info (Optional[Dict], optional) – Additional information dictionary |default| None
title (str, optional) – Message title |default| 'Error occurred'

Returns:

Boolean indicating success or failure

get_level_color(c_level='')[source]

Return type:: str

get_status_emoji(status)[source]

Return appropriate emoji based on the status.

Parameters:: status (Union[str, StatusType]) – The status (success, failed, warning, etc.)
Return type:: str
Returns:: Emoji string corresponding to the status

create_slack_payload(msg_text, title, send_user_name, msg_level, status, simple_mode, icon_emoji='', max_value_length=100, footer='', text='')[source]

Create the payload for sending a message to Slack.

Parameters:

msg_text (Union[str, dict, list]) – The main message text, can be a string, dict, or list.
title (str) – Title of the message.
send_user_name (str) – The username to display in Slack.
msg_level (str) – Severity level of the message (info, warning, error, critical).
status (Union[str, StatusType]) – Status type or string for dynamic emoji and message formatting.
simple_mode (bool) – If True, send a simplified message without additional info.
icon_emoji (str) – Optional emoji to display as the icon for the message. |default| ''
max_value_length (int) – Maximum length for any single value. |default| 100
footer (str) – Footer text to display at the bottom of the message. |default| ''
text (str) – Optional text to display in Slack |default| ''

Returns:

Dictionary containing the Slack message payload.

Return type:

dict

async send_slack_async(url, payload, retries)[source]

Asynchronous Slack message sender with retry logic.

Parameters:

url (str) – Slack webhook URL.
payload (dict) – The payload to send to Slack.
retries (int) – Number of retry attempts in case of failure.

Returns:

Boolean indicating success or failure.

Return type:

bool

send_slack_sync(url, payload, retries)[source]

Synchronous Slack message sender with retry logic.

Parameters:

url (str) – Slack webhook URL.
payload (dict) – The payload to send to Slack.
retries (int) – Number of retry attempts in case of failure.

Returns:

Boolean indicating success or failure.

Return type:

bool

send_slack(url='', msg_text='', title='', text='', send_user_name='CtxBot', msg_level='info', retries=1, status='ℹ️', simple_mode=False, async_mode=False, icon_emoji='', footer='')[source]

Send a message to Slack with optional retry logic and dynamic emoji based on status.

Parameters:

url (str) – Slack webhook URL (fetched from env SLACK_WEBHOOK_URL if not provided) |default| ''
msg_text (Union[str, dict, list]) – The main message to send |default| ''
title (str) – Optional title for the message |default| ''
text (str) – Optional text to display in Slack |default| ''
send_user_name (str) – Username to display in Slack |default| 'CtxBot'
msg_level (str) – Message severity level (info, warning, error, critical) |default| 'info'
retries (int) – Number of retries in case of failure |default| 1
status (Union[str, StatusType]) – Either a string or StatusType enum value to use a different format with emojis |default| 'ℹ️'
simple_mode (bool) – If True, send a simple message without extra info like host or date |default| False
async_mode (bool) – If True, sends the message asynchronously |default| False
icon_emoji (str) – Optional emoji to display as the icon for the message. |default| ''
footer (str) – Footer text to display at the bottom of the message. |default| ''

Returns:

Boolean indicating success or failure

Return type:

bool

Example

from pawnlib.utils.notify import send_slack

# Send a message with status using StatusType enum
send_slack(SLACK_WEBHOOK_URL, "The process completed successfully",
           title="Process Status", msg_level="info", status=StatusType.SUCCESS)

# Send a message with status using string
send_slack(SLACK_WEBHOOK_URL, "The process completed successfully",
           title="Process Status", msg_level="info", status="success")

# Send an error message
send_slack(SLACK_WEBHOOK_URL, "The process failed due to an unexpected error",
           title="Process Status", msg_level="error", status="failed")

# Send a warning message
send_slack(SLACK_WEBHOOK_URL, "The disk space is running low",
           title="System Warning", msg_level="warning", status="warning")

# Send a message for an in-progress task
send_slack(SLACK_WEBHOOK_URL, "The process is currently running",
           title="Process Status", msg_level="info", status="in_progress")

# Send a message for task completion
send_slack(SLACK_WEBHOOK_URL, "The task has been completed",
           title="Task Status", msg_level="info", status="complete")

send_slack_token(title=None, message=None, token=None, channel_name=None, send_user='python_app', msg_level='info')[source]

format_slack_message(title='', msg_text='', msg_level='info', status='', icon_emoji='', footer='', text='')[source]

Return type:: dict

async send_slack_notification(title, msg_text, level='info', icon_emoji='', footer='Pawnlib')[source]

pawnlib.utils.operate_handler module

from pawnlib.utils import operate_handler

class ThreadPoolRunner(func=None, tasks=[], max_workers=20, verbose=0, sleep=1)[source]

Bases: object

A class that runs a function with multiple arguments in parallel using a thread pool.

Parameters:

func (function) – The function to run in parallel. |default| None
tasks (list) – A list of arguments to pass to the function. |default| []
max_workers (int) – The maximum number of worker threads to use. |default| 20
verbose (int) – Whether to print the results of each task as they complete. |default| 0
sleep (int) – The number of seconds to sleep between runs when using forever_run. |default| 1

Example

runner = ThreadPoolRunner(func=my_function, tasks=my_args, max_workers=10, verbose=1, sleep=5)
results = runner.run()
runner.forever_run()

initializer_worker()[source]: A method that is run once by each worker thread when the thread pool is created.

run(tasks=None, timeout=None)[source]

Run the function with the given arguments in parallel using a thread pool.

Parameters:

tasks (list or generator) – A list or generator of tasks. |default| None
timeout (int or None) – Timeout for each task in seconds. If None, no timeout is applied. |default| None

Returns:

A list of results from each task, in the same order as the input.

Return type:

list

static log_results(results)[source]

Print the results of each task as they complete.

Parameters:: results (list) – A list of results from each task.

forever_run()[source]: Run the function with the given arguments in parallel using a thread pool indefinitely.

stop()[source]: Stop the forever_run loop.

class Daemon(pidfile, func=None, stdin=None, stdout=None, stderr=None, home_dir='.', umask=18, verbose=1, use_gevent=False, use_eventlet=False)[source]

Bases: object

A generic daemon class. Usage 1: subclass the Daemon class and override the run() method Usage 2: subclass the Daemon class and use func parameter

Parameters:

pidfile – pid file location
func (optional) – function to run as daemon |default| None
stdin (optional) – standard input , The default is sys.stdin, and providing a filename will output to a file. |default| None
stdout (optional) – standard output, The default is sys.stdout, and providing a filename will output to a file. |default| None
stderr (optional) – standard error, The default is sys.stderr, and providing a filename will output to a file. |default| None
home_dir (optional) – home directory |default| '.'
umask (optional) – umask |default| 18
verbose (optional) – verbosity level |default| 1
use_gevent (bool, optional) – use gevent |default| False

use_eventlet (bool, optional) –

use eventlet Example:

|default| False

from pawnlib.utils.operate_handler import Daemon

def main():
    while True:
        print(f"main loop")
        print("start daemon")
        time.sleep(5)


if __name__ == "__main__":
    if len(sys.argv) != 2:
        sys.exit()
    command = sys.argv[1]
    daemon = Daemon(
        pidfile="/tmp/jmon_agent.pid",
        func=main
    )
    if command == "start":
        daemon.start()
    elif command == "stop":
        daemon.stop()
    else:
        print("command not found [start/stop]")

log(*args)[source]

daemonize()[source]: Do the UNIX double-fork magic, see Stevens’ “Advanced Programming in the UNIX Environment” for details (ISBN 0201563177) http://www.erlenstar.demon.co.uk/unix/faq_2.html#SEC16

delpid()[source]

start(*args, **kwargs)[source]: Start the daemon

stop()[source]: Stop the daemon

restart()[source]: Restart the daemon

get_pid()[source]

is_running()[source]

run()[source]: You should override this method when you subclass Daemon. It will be called after the process has been daemonized by start() or restart().

timing(f)[source]

Get the time taken to complete the task.

Parameters:: f –
Returns:

get_inspect_module(full_module_name=None)[source]

job_start(full_module_name=None)[source]

job_done(error='', full_module_name='', elapsed=0)[source]

execute_function(module_func)[source]

run_execute(*args, **kwargs)[source]

execute_command(cmd, text=None, cwd=None, check_output=True, capture_output=True, hook_function=None, debug=False, use_spinner=False, spinner_type='dots', spinner_text=None, **kwargs)[source]

Executes a shell command and captures its output, with an optional spinner for visual feedback.

Parameters:

cmd (str) – Command to be executed.
text (str, optional) – Descriptive text or title for the command. |default| None
cwd (str, optional) – Working directory to execute the command in. |default| None
check_output (bool, optional) – If True, logs the command execution result. |default| True
capture_output (bool, optional) – If True, captures the command’s stdout. |default| True
hook_function (Callable[[str, int], None], optional) – Function to process each line of stdout. |default| None
debug (bool, optional) – If True, prints debug information. |default| False
use_spinner (bool, optional) – If True, shows a spinner while the command executes. |default| False
spinner_type (str, optional) – The type of spinner to use (from Rich library). |default| 'dots'
spinner_text (str, optional) – Custom text to display alongside the spinner. |default| None
**kwargs – Additional keyword arguments to pass to the hook_function.

Returns:

A dictionary containing the command’s execution results.

Return type:

Dict[str, Any]

Raises:

OSError – If an error occurs while executing the command.

execute_command_batch(tasks, stop_on_error=False, slack_url=None, default_kwargs=None, function_registry=None)[source]

Executes a batch of tasks, where each task can be a string (command) or a dictionary containing arguments for execute_command. Handles errors and sends optional Slack notifications.

Parameters:

tasks (Union[List[str], List[Dict[str, Any]]]) – List of commands (as strings or dictionaries).
stop_on_error (bool, optional) – If True, stops execution upon encountering an error. |default| False
slack_url (str, optional) – Slack webhook URL for sending notifications. |default| None
default_kwargs (Dict[str, Any], optional) – Default arguments to apply to each task. |default| None

Returns:

A list of results from execute_command for each task.

Return type:

List[Dict[str, Any]]

execute_registered_function(function_name, args=None, debug=False, function_registry=None)[source]

Executes the specified function by name and returns the result.

Parameters:

function_name (str) – The name of the function to execute.
args (Optional[Dict[str, Any]]) – A dictionary of arguments to pass to the function. |default| None
debug (bool, optional) – If True, enables debug logging. |default| False
function_registry (dict, optional) – A registry of available functions. |default| None

Returns:

A dictionary containing the execution results, including ‘stdout’, ‘stderr’, ‘return_code’, ‘line_no’, and ‘elapsed’.

Return type:

Dict[str, Any]

hook_print(*args, **kwargs)[source]

Print to output every 10th line

Parameters:

args –
kwargs –

Returns:

class Spinner(text='', delay=0.1)[source]

Bases: object

Create a spinning cursor

Parameters:

text (optional) – text |default| ''
delay (optional) – sleep time |default| 0.1

:Example

from pawnlib.utils.operate_handler import Spinner
with Spinner(text="Wait message"):
    time.sleep(10)

title(text=None)[source]

write_next()[source]

remove_spinner(cleanup=False)[source]

spinner_task()[source]

start()[source]

stop()[source]

class WaitStateLoop(loop_function, exit_function, timeout=30, delay=0.5, text='WaitStateLoop')[source]

Bases: object

loop_function is continuously executed and values are compared with exit_function.

Parameters:

loop_function (Callable) – function to run continuously
exit_function (Callable) – function to exit the loop
timeout (optional) – |default| 30
delay (optional) – sleep time |default| 0.5
text (optional) – text message |default| 'WaitStateLoop'

Example

from pawnlib.utils.operate_handler import WaitStateLoop
from functools import partial

def check_func(param=None):
    time.sleep(0.2)
    random_int = random.randint(1, 1000)
    # print(f"param= {param}, random_int = {random_int}")
    return random_int

def loop_exit_func(result):
    if result % 10 == 1.5:
        return True
    return False

WaitStateLoop(
    loop_function=partial(check_func, "param_one"),
    exit_function=loop_exit_func,
    timeout=10
).run()

run()[source]

Returns:

wait_state_loop(exec_function=None, func_args=[], check_key='status', wait_state='0x1', timeout_limit=30, increase_sec=0.5, health_status=None, description='', force_dict=True, logger=None)[source]

run_with_keyboard_interrupt(command, *args, **kwargs)[source]

run with KeyboardInterrupt :type command: :param command: :type args: :param args: :type kwargs: :param kwargs: :return:

Example

from pawnlib.utils.operate_handler import run_with_keyboard_interrupt

run_with_keyboard_interrupt(run_func, args, kwargs)

handle_keyboard_interrupt_signal()[source]

pawnlib.utils.icx_signer module

from pawnlib.utils import icx_signer

guess_wallet_type(data)[source]

Guesses the type of wallet based on the provided data.

Parameters:: data (str or object) – The data to analyze and determine the wallet type.
Returns:: The guessed wallet type. It can be “private_key” for a private key wallet, “json” for a JSON wallet, or None if the type cannot be determined.
Return type:: str or None

This function attempts to determine the type of a wallet based on the provided data. If the data parameter is a string and its length is either 66 or 64 characters, it is considered a PrivateKey wallet type. If the data parameter is a valid JSON object, it is considered a JSON wallet type. If the wallet type cannot be determined, None is returned.

Example

from pawnlib.utils.icx_signer import guess_wallet_type

#Example 1: PrivateKey wallet type
wallet_data = "0x1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef"
wallet_type = guess_wallet_type(wallet_data)
# wallet_type = "private_key"

#Example 2: JSON wallet type
wallet_data = '{"address": "0x1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef"}'
wallet_type = guess_wallet_type(wallet_data)
# wallet_type = "json"

#Example 3: Unknown wallet type
wallet_data = 12345
wallet_type = guess_wallet_type(wallet_data)
# wallet_type = None

class WalletCli(args=None)[source]

Bases: object

load()[source]

create(is_store_file=True)[source]

print_wallet()[source]

store_keystore_file_on_the_path(file_path, json_string, overwrite=False)[source]: Stores a created keystore string data which is JSON format on the file path. :type file_path: :param file_path: The path where the file will be saved. type(str) :type json_string: :param json_string: Contents of the keystore.

generate_wallet(file_path=None, password=None, overwrite=False, private_key=None, expected_address=None, is_store_file=True)[source]

is_private_key(private_key)[source]

exit_on_failure(raise_on_failure, exception)[source]

load_wallet_key(file_or_object=None, password=None, raise_on_failure=True, use_namespace=False)[source]

generate_keys()[source]

generate privkey and pubkey pair using coincurve.

Returns:: privkey(bytes, 32), pubkey(bytes, 65)
Return type:: tuple

get_address(pubkey_bytes)[source]

generate address from public key.

Parameters:: pubkey_bytes (bytes) – public key bytes
Returns:: icx address (20bytes)
Return type:: bytes

verify_recoverable_signature(msg_hash, signature_bytes, recovery_id)[source]

Parameters:

msg_hash (bytes) – 256bit hash value
signature_bytes (bytes) –
recovery_id (int) –

Returns:

recover_signature(msg_hash, signature_bytes, recovery_id)[source]

Parameters:

msg_hash (bytes) – sha3 256bit hash value
signature_bytes (bytes) –
recovery_id (int) –

Returns:

signature(bytes):

Return type:

pubkey(PublicKey)

class IcxSigner(data=None, raw=True)[source]

Bases: object

Digital Signing using coincurve

Constructor

Parameters:

data (object) – bytes or der |default| None
raw (bool) – True(bytes) False(der) |default| True

__init__(data=None, raw=True)[source]

Constructor

Parameters:

data (object) – bytes or der |default| None
raw (bool) – True(bytes) False(der) |default| True

set_privkey_bytes(data)[source]

Set private key using private key data in bytes.

Parameters:: data (bytes) – private key data

get_privkey_bytes()[source]

Get private key data in bytes.

Returns:: private key data (32 bytes)
Return type:: bytes

get_pubkey_bytes()[source]

get_address()[source]

Create an address with pubkey. address is made from pubkey.

Returns:: address represented in hexadecimal string starting with ‘0x’
Return type:: str

get_hx_address()[source]

Create an address with pubkey. address is made from pubkey.

Returns:: address represented in hexadecimal string starting with ‘0x’
Return type:: str

sign_tx(tx=None)[source]

sign(msg_hash)[source]

Make a signature using the hash value of msg.

Parameters:: msg_hash (bytes) – msg_hash = sha3_256(msg)
Returns:: signature bytes
Return type:: bytes

create_key_store_content(password, iterations=16384, kdf='scrypt')[source]

store(file_path, password, overwrite=False, expected_address=None)[source]

sign_recoverable(msg_hash)[source]

Make a recoverable signature using message hash data We can extract public key from recoverable signature.

Parameters:: msg_hash (bytes) – hash data of message
Returns:: bytes: 65 bytes data int: recovery id
Return type:: tuple

static from_bytes(data)[source]

static from_der(data)[source]

class IcxSignVerifier(data)[source]

Bases: object

Digital signature verification

Parameters:: data (bytes) – 65 bytes data which PublicKey.serialize() returns
Returns:: None

__init__(data)[source]

Parameters:: data (bytes) – 65 bytes data which PublicKey.serialize() returns
Returns:: None

get_address()[source]

Create an address with pubkey. address is made from pubkey.

Returns:: address represented in hexadecimal string starting with ‘0x’
Return type:: str

verify(msg_hash, signature_bytes)[source]

Check whether signature is valid or not.

Parameters:

pubkey_bytes (bytes) – byte data of pubkey
msg_hash (bytes) – hash value of msg
signature_bytes (bytes) – signature data

Returns:

the result of signature verification

Return type:

bool

static from_bytes(data)[source]

Parameters:: data (bytes) – bytes data which PublicKey.serialize() returns
Returns:: None

get_timestamp_us()[source]: Get epoch time in us.

icx_to_wei(icx)[source]

Convert amount in icx unitt to wei unit.

Parameters:: icx (float) – float value in icx unit
Returns:: int value in wei unit
Return type:: int

get_string_decimal(value, place)[source]

value를 10의 place 제곱으로 나눈 값을 string으로 변환하여 반환

Parameters:

value (int) –
place – 10의 몇 제곱을 나눌지 입력받음

sha3_256(data)[source]

Get hash value using sha3_256 hash function

Parameters:: data (bytes) – data to hash
Returns:: 256bit hash value (32 bytes)
Return type:: bytes

get_tx_hash(params=None)[source]

Create tx_hash from params object.

Parameters:: params (dict) – the value of ‘params’ key in jsonrpc |default| None
Returns:: sha3_256 hash value :param params:
Return type:: bytes

get_tx_phrase(method, params)[source]

Create tx phrase from method and params. tx_phrase means input text to create tx_hash.

Parameters:: params (dict) – the value of ‘params’ key in jsonrpc
Returns:: sha3_256 hash format without ‘0x’ prefix
Return type:: str

sign_recoverable(privkey_bytes, tx_hash_bytes)[source]

Parameters:: tx_hash (bytes) – 32byte tx_hash data
Returns:: signature_bytes + recovery_id(1)
Return type:: bytes

serialize(params)[source]: Serialized params of an original JSON request starting with icx_sendTransaction to generate a message hash for a signature. :type params: dict :param params: params in a original JSON request for transaction. :rtype: bytes :return: serialized params. For example, data like icx_sendTransaction.<key1>.<value1>.<key2>.<value2> is converted to bytes.

generate_message(params)[source]: Generates transaction’s message hash from params in request for transaction. :type params: dict :param params: params in request for transaction. :rtype: str :return: the 256 bit hash digest of a message. Hexadecimal encoded.

pawnlib.utils.genesis module

from pawnlib.utils import genesis

class GenesisGenerator(genesis_json_or_dict=None, base_dir=None, genesis_filename='icon_genesis.zip')[source]

Bases: object

A class to generate and manage genesis files, including creating a temporary directory, parsing JSON, creating CID, and writing the genesis zip file.

Parameters:

genesis_json_or_dict (Union[dict, str]) – A JSON dictionary or file path representing the genesis data. |default| None
base_dir (str) – The base directory where the temporary files and final zip will be stored. Defaults to the pawnlib path. |default| None
genesis_filename (str) – The name of the genesis zip file to be created. Defaults to ‘icon_genesis.zip’. |default| 'icon_genesis.zip'

Example

from pawnlib.utils import GenesisGenerator

# Define a genesis JSON object
genesis_data = {
    "accounts": [
        {
            "address": "hx112759c9e5718c48527f0242887b7f9f852da29d",
            "balance": "0x2961fff8ca4a62327800000",
            "name": "god"
        },
        {
            "address": "hx1000000000000000000000000000000000000000",
            "balance": "0x0",
            "name": "treasury"
        },
        {
            "address": "cx0000000000000000000000000000000000000001",
            "name": "governance",
            "score": {
                "contentId": "hash:{{hash:governance/governance-2.2.1-optimized.jar}}",
                "contentType": "application/java",
                "owner": "hx522759c9e5718c48527f0242887b7f9f852da29d"
            }
        }
    ],
    "chain": {
        "revision": "0x17",
        "blockInterval": "0x3e8",
        "roundLimitFactor": "0x10",
        "fee": {
            "stepPrice": "0x2e90edd00",
            "stepLimit": {
                "invoke": "0x9502f900",
                "query": "0x2faf080"
            },
            "stepCosts": {
                "apiCall": "0x2710",
                "contractCall": "0x61a8",
                "contractCreate": "0x3b9aca00",
                "contractSet": "0x3a98",
                "contractUpdate": "0x3b9aca00",
                "default": "0x186a0",
                "delete": "-0xf0",
                "deleteBase": "0xc8",
                "get": "0x19",
                "getBase": "0xbb8",
                "input": "0xc8",
                "log": "0x64",
                "logBase": "0x1388",
                "schema": "0x1",
                "set": "0x140",
                "setBase": "0x2710"
            }
        },
        "validatorList": [
            "hx522759c9e5718c48527f0242887b7f9f852da29d"
        ]
    },
    "message": "genesis for local node",
    "nid": "0x99"
}

# Create a GenesisGenerator object
generator = GenesisGenerator(genesis_json_or_dict=genesis_data)

# Run the generator to process and generate the genesis zip file
cid = generator.run()
print(f"Generated CID: {cid}")

Initialize the GenesisGenerator with the given genesis data, base directory, and zip filename.

Parameters:

genesis_json_or_dict (optional) – Genesis data as a dictionary or file path. |default| None
base_dir (optional) – Base directory to store the files. |default| None
genesis_filename (optional) – Name of the final genesis zip file. |default| 'icon_genesis.zip'

__init__(genesis_json_or_dict=None, base_dir=None, genesis_filename='icon_genesis.zip')[source]

Initialize the GenesisGenerator with the given genesis data, base directory, and zip filename.

Parameters:

genesis_json_or_dict (optional) – Genesis data as a dictionary or file path. |default| None
base_dir (optional) – Base directory to store the files. |default| None
genesis_filename (optional) – Name of the final genesis zip file. |default| 'icon_genesis.zip'

make_temp_dir()[source]: Create temporary directories for preparing and finalizing the genesis file.

initialize()[source]

Initialize the genesis generator by loading the genesis JSON data from either a dictionary or a file. If it’s a file path, the data will be loaded from the file.

Raises:: ValueError – If the genesis data is neither a valid dictionary nor a valid file path.

run(genesis_json_or_dict=None, base_dir=None, genesis_filename=None, cleanup_after_run=True)[source]

Main method to execute the genesis generation process. It initializes the data, logs relevant information, parses the genesis JSON, writes the zip file, and generates the CID.

Parameters:

genesis_json_or_dict (optional) – Optional genesis data as a dictionary or file path. |default| None
base_dir (optional) – Optional base directory to store the files. |default| None
genesis_filename (optional) – Optional name of the final genesis zip file. |default| None
cleanup_after_run (optional) – If True, cleanup temporary directories after run. |default| True

Returns:

CID (content identifier) of the generated genesis file.

Return type:

str

Return type:

bool

cleanup()[source]: Remove temporary directories to cleanup the environment.

validate_cid()[source]: Recalculate CID from the final genesis.json and compare with self.cid to ensure integrity. Raises ValueError if CID does not match.

write_metadata()[source]: Write metadata.json to final_temp_dir. Include CID, NID, timestamp, and genesis_filename.

log_initialization_info()[source]: Log the base directory and temporary directory information for debugging purposes.

create_cid(data=None)[source]

Create a CID (Content Identifier) by serializing the genesis data and calculating its hash.

Parameters:: data (optional) – Optional data to be serialized and used to create the CID. If not provided, it uses genesis_json_or_dict. |default| None
Returns:: The CID generated from the serialized data.
Return type:: str

process_content_id_and_write_genesis()[source]

Process accounts in genesis_data and write the resulting JSON to a temporary file.

Combines update_content_id_in_accounts() and write_genesis_json() functionality.

update_content_id_in_accounts()[source]

Process genesis_data accounts and replace contentId fields with processed results.

Updates self.genesis_data by transforming the contentId fields in accounts.

Raises:: ValueError – If genesis_data is not properly initialized.

is_already_hashed(content_id)[source]

Check if the given content_id is already in a hashed form like: hash:<64-hexdigits>.

Parameters:: content_id (str) – The content ID string to check.
Return type:: bool
Returns:: True if it’s already a processed hash form, False otherwise.

write_genesis_json()[source]

Write genesis_data to a JSON file in the final temporary directory.

This method assumes parse_genesis_json() has already processed genesis_data.

Raises:: ValueError – If genesis_data is not properly initialized.

write_genesis_zip()[source]

Write the genesis JSON to a zip file.

Returns:: Information about the generated zip file, including its size and attributes.
Return type:: dict

static extract_content_pattern(string)[source]

Extracts a template key, template type, and template directory from the provided string using a regex pattern.

Parameters:: string – The string containing the content pattern.
Returns:: A tuple of the extracted template key, template type, and template directory.
Return type:: tuple

make_score_zip(content_id)[source]

Create a zip file or copy the file based on the content ID and its type (e.g., ziphash or hash).

Parameters:: content_id – The content ID that describes how to process the score directory.
Returns:: The resulting hash with the file type.
Return type:: str
Raises:: ValueError – If the file or directory described by the content ID cannot be found.

make_zip_or_copy_file(template_type, score_dir)[source]

Either create a zip file or copy a file, depending on the template type, and return the hash of the result.

Parameters:

template_type – The type of the operation to perform (“ziphash” or “hash”).
score_dir – The directory or file to be processed.

Returns:

The hash of the resulting file.

Return type:

str

make_zip_without(src_dir, dst_file, exclude_dirs)[source]

Create a zip archive from the source directory, excluding specified subdirectories.

Parameters:

src_dir (str) – The source directory to zip.
dst_file (str) – The output file path for the zip archive.
exclude_dirs (list) – List of directory names to exclude from the zip archive.

calculate_hash(file_path)[source]

Return the SHA-256 hash of a file.

Parameters:: file_path (str) – Path to the file to be hashed.
Returns:: The SHA-256 hash of the file.
Return type:: str

create_cid(data)[source]

Create a CID (Content Identifier) from a given dictionary by serializing the data and hashing it.

Parameters:: data (dict) – The dictionary containing the genesis data.
Returns:: The CID created from the serialized data.
Return type:: str

create_cid_from_genesis_file(genesis_file)[source]

Create a CID from a genesis file by loading the file and generating the CID.

Parameters:

genesis_file (str) – Path to the genesis JSON file.

Returns:

The CID created from the genesis file.

Return type:

str

Raises:

FileNotFoundError – If the genesis file is not found.
PermissionError – If there are permission issues with the file.
Exception – If any other unexpected errors occur.

create_cid_from_genesis_zip(zip_file_name)[source]

Create a CID from a genesis zip file by extracting the JSON data from the zip and generating the CID.

Parameters:: zip_file_name (str) – Path to the genesis zip file.
Returns:: The CID created from the zip file.
Return type:: str

validate_genesis_json(genesis_json=None)[source]

Validate the structure of a genesis JSON object to ensure it contains all mandatory keys.

Parameters:: genesis_json (dict) – The genesis JSON object to validate. |default| None
Returns:: True if the genesis JSON is valid.
Return type:: bool
Raises:: SystemExit – If the genesis JSON is missing mandatory keys or is not a dictionary.

genesis_generator(genesis_json_or_dict=None, base_dir=None, genesis_filename=None, cleanup_after_run=True)

Main method to execute the genesis generation process. It initializes the data, logs relevant information, parses the genesis JSON, writes the zip file, and generates the CID.

Parameters:

genesis_json_or_dict (optional) – Optional genesis data as a dictionary or file path. |default| None
base_dir (optional) – Optional base directory to store the files. |default| None
genesis_filename (optional) – Optional name of the final genesis zip file. |default| None
cleanup_after_run (optional) – If True, cleanup temporary directories after run. |default| True

Returns:

CID (content identifier) of the generated genesis file.

Return type:

str

Return type:

bool

pawnlib.utils.in_memory_zip module

from pawnlib.utils import in_memory_zip

gen_deploy_data_content(_path)[source]

Generate bytes of zip data of SCORE.

Parameters:: _path (str) – Path of the directory to be zipped.
Return type:: bytes

read_file_from_zip(zip_file_name, target_file_name)[source]

read_genesis_dict_from_zip(zip_file_name='')[source]

Return type:: dict

class InMemoryZip[source]

Bases: object

Class for compressing data in memory using zip and BytesIO.

property data: bytes

Returns zip data

Returns:: zip data

zip_in_memory(_path)[source]

Compress zip data (bytes) in memory.

Parameters:: _path (str) – The path of the directory to be zipped.

pawnlib.utils.aws module

from pawnlib.utils import aws

pawnlib.utils.redis_helper module

from pawnlib.utils import redis_helper