Edit on GitHub

communex._common

View Source

  1import random
  2import re
  3from collections import defaultdict
  4from enum import Enum
  5from typing import Mapping, TypeVar
  6
  7from pydantic_settings import BaseSettings, SettingsConfigDict
  8
  9from communex.balance import from_nano
 10from communex.types import Ss58Address
 11
 12IPFS_REGEX = re.compile(r"^Qm[1-9A-HJ-NP-Za-km-z]{44}$")
 13
 14
 15class ComxSettings(BaseSettings):
 16    model_config = SettingsConfigDict(env_prefix="COMX_")
 17    # TODO: improve node lists
 18    NODE_URLS: list[str] = [
 19        "wss://commune-api-node-0.communeai.net",
 20        "wss://commune-api-node-1.communeai.net",
 21        "wss://commune-api-node-2.communeai.net",
 22        "wss://commune-api-node-3.communeai.net",
 23        "wss://commune-api-node-4.communeai.net",
 24        "wss://commune-api-node-5.communeai.net",
 25        "wss://commune-api-node-6.communeai.net",
 26        "wss://commune-api-node-7.communeai.net",
 27        "wss://commune-api-node-8.communeai.net",
 28        "wss://commune-api-node-9.communeai.net",
 29        "wss://commune-api-node-10.communeai.net",
 30        "wss://commune-api-node-11.communeai.net",
 31        "wss://commune-api-node-12.communeai.net",
 32        "wss://commune-api-node-13.communeai.net",
 33        "wss://commune-api-node-14.communeai.net",
 34        "wss://commune-api-node-15.communeai.net",
 35        "wss://commune-api-node-16.communeai.net",
 36        "wss://commune-api-node-17.communeai.net",
 37        "wss://commune-api-node-18.communeai.net",
 38        "wss://commune-api-node-19.communeai.net",
 39        "wss://commune-api-node-20.communeai.net",
 40        "wss://commune-api-node-21.communeai.net",
 41        "wss://commune-api-node-22.communeai.net",
 42        "wss://commune-api-node-23.communeai.net",
 43        "wss://commune-api-node-24.communeai.net",
 44        "wss://commune-api-node-25.communeai.net",
 45        "wss://commune-api-node-26.communeai.net",
 46        "wss://commune-api-node-27.communeai.net",
 47        "wss://commune-api-node-28.communeai.net",
 48        "wss://commune-api-node-29.communeai.net",
 49        "wss://commune-api-node-30.communeai.net",
 50        "wss://commune-api-node-31.communeai.net",
 51    ]
 52    TESTNET_NODE_URLS: list[str] = ["wss://testnet-commune-api-node-0.communeai.net"]
 53
 54
 55def get_node_url(
 56    comx_settings: ComxSettings | None = None, *, use_testnet: bool = False
 57) -> str:
 58    comx_settings = comx_settings or ComxSettings()
 59    match use_testnet:
 60        case True:
 61            node_url = random.choice(comx_settings.TESTNET_NODE_URLS)
 62        case False:
 63            node_url = random.choice(comx_settings.NODE_URLS)
 64    return node_url
 65
 66
 67def get_available_nodes(
 68    comx_settings: ComxSettings | None = None, *, use_testnet: bool = False
 69) -> list[str]:
 70    comx_settings = comx_settings or ComxSettings()
 71
 72    match use_testnet:
 73        case True:
 74            node_urls = comx_settings.TESTNET_NODE_URLS
 75        case False:
 76            node_urls = comx_settings.NODE_URLS
 77    return node_urls
 78
 79
 80class BalanceUnit(str, Enum):
 81    joule = "joule"
 82    j = "j"
 83    nano = "nano"
 84    n = "n"
 85
 86
 87def format_balance(balance: int, unit: BalanceUnit = BalanceUnit.nano) -> str:
 88    """
 89    Formats a balance.
 90    """
 91
 92    match unit:
 93        case BalanceUnit.nano | BalanceUnit.n:
 94            return f"{balance}"
 95        case BalanceUnit.joule | BalanceUnit.j:
 96            in_joules = from_nano(balance)
 97            round_joules = round(in_joules, 4)
 98            return f"{round_joules:,} COMAI"
 99
100
101K = TypeVar("K")
102V = TypeVar("V")
103Z = TypeVar("Z")
104
105
106def intersection_update(base: dict[K, V], update: dict[K, Z]) -> Mapping[K, V | Z]:
107    """
108    Update a dictionary with another dictionary, but only with keys that are already present.
109    """
110    updated = {k: update[k] for k in base if k in update}
111    return updated
112
113
114def transform_stake_dmap(stake_storage: dict[tuple[Ss58Address, Ss58Address], int]) -> dict[Ss58Address, list[tuple[Ss58Address, int]]]:
115    """
116    Transforms either the StakeTo or StakeFrom storage into the stake legacy data type.
117    """
118    transformed: dict[Ss58Address, list[tuple[Ss58Address, int]]] = defaultdict(list)
119    [transformed[k1].append((k2, v)) for (k1, k2), v in stake_storage.items()]
120
121    return dict(transformed)

IPFS_REGEX = re.compile('^Qm[1-9A-HJ-NP-Za-km-z]{44}$')

class ComxSettings(pydantic_settings.main.BaseSettings): View Source

16class ComxSettings(BaseSettings):
17    model_config = SettingsConfigDict(env_prefix="COMX_")
18    # TODO: improve node lists
19    NODE_URLS: list[str] = [
20        "wss://commune-api-node-0.communeai.net",
21        "wss://commune-api-node-1.communeai.net",
22        "wss://commune-api-node-2.communeai.net",
23        "wss://commune-api-node-3.communeai.net",
24        "wss://commune-api-node-4.communeai.net",
25        "wss://commune-api-node-5.communeai.net",
26        "wss://commune-api-node-6.communeai.net",
27        "wss://commune-api-node-7.communeai.net",
28        "wss://commune-api-node-8.communeai.net",
29        "wss://commune-api-node-9.communeai.net",
30        "wss://commune-api-node-10.communeai.net",
31        "wss://commune-api-node-11.communeai.net",
32        "wss://commune-api-node-12.communeai.net",
33        "wss://commune-api-node-13.communeai.net",
34        "wss://commune-api-node-14.communeai.net",
35        "wss://commune-api-node-15.communeai.net",
36        "wss://commune-api-node-16.communeai.net",
37        "wss://commune-api-node-17.communeai.net",
38        "wss://commune-api-node-18.communeai.net",
39        "wss://commune-api-node-19.communeai.net",
40        "wss://commune-api-node-20.communeai.net",
41        "wss://commune-api-node-21.communeai.net",
42        "wss://commune-api-node-22.communeai.net",
43        "wss://commune-api-node-23.communeai.net",
44        "wss://commune-api-node-24.communeai.net",
45        "wss://commune-api-node-25.communeai.net",
46        "wss://commune-api-node-26.communeai.net",
47        "wss://commune-api-node-27.communeai.net",
48        "wss://commune-api-node-28.communeai.net",
49        "wss://commune-api-node-29.communeai.net",
50        "wss://commune-api-node-30.communeai.net",
51        "wss://commune-api-node-31.communeai.net",
52    ]
53    TESTNET_NODE_URLS: list[str] = ["wss://testnet-commune-api-node-0.communeai.net"]

Base class for settings, allowing values to be overridden by environment variables.

This is useful in production for secrets you do not wish to save in code, it plays nicely with docker(-compose), Heroku and any 12 factor app design.

All the below attributes can be set via model_config.

Arguments:

_case_sensitive: Whether environment variables names should be read with case-sensitivity. Defaults to None.
_env_prefix: Prefix for all environment variables. Defaults to None.
_env_file: The env file(s) to load settings values from. Defaults to Path(''), which means that the value from model_config['env_file'] should be used. You can also pass None to indicate that environment variables should not be loaded from an env file.
_env_file_encoding: The env file encoding, e.g. 'latin-1'. Defaults to None.
_env_ignore_empty: Ignore environment variables where the value is an empty string. Default to False.
_env_nested_delimiter: The nested env values delimiter. Defaults to None.
_env_parse_none_str: The env string value that should be parsed (e.g. "null", "void", "None", etc.) into None type(None). Defaults to None type(None), which means no parsing should occur.
_env_parse_enums: Parse enum field names to values. Defaults to None., which means no parsing should occur.
_cli_prog_name: The CLI program name to display in help text. Defaults to None if _cli_parse_args is None. Otherwse, defaults to sys.argv[0].
_cli_parse_args: The list of CLI arguments to parse. Defaults to None. If set to True, defaults to sys.argv[1:].
_cli_settings_source: Override the default CLI settings source with a user defined instance. Defaults to None.
_cli_parse_none_str: The CLI string value that should be parsed (e.g. "null", "void", "None", etc.) into None type(None). Defaults to _env_parse_none_str value if set. Otherwise, defaults to "null" if _cli_avoid_json is False, and "None" if _cli_avoid_json is True.
_cli_hide_none_type: Hide None values in CLI help text. Defaults to False.
_cli_avoid_json: Avoid complex JSON objects in CLI help text. Defaults to False.
_cli_enforce_required: Enforce required fields at the CLI. Defaults to False.
_cli_use_class_docs_for_groups: Use class docstrings in CLI group help text instead of field descriptions. Defaults to False.
_cli_prefix: The root parser command line arguments prefix. Defaults to "".
_secrets_dir: The secret files directory. Defaults to None.

model_config = {'extra': 'forbid', 'arbitrary_types_allowed': True, 'validate_default': True, 'case_sensitive': False, 'env_prefix': 'COMX_', 'env_file': None, 'env_file_encoding': None, 'env_ignore_empty': False, 'env_nested_delimiter': None, 'env_parse_none_str': None, 'env_parse_enums': None, 'cli_prog_name': None, 'cli_parse_args': None, 'cli_settings_source': None, 'cli_parse_none_str': None, 'cli_hide_none_type': False, 'cli_avoid_json': False, 'cli_enforce_required': False, 'cli_use_class_docs_for_groups': False, 'cli_prefix': '', 'json_file': None, 'json_file_encoding': None, 'yaml_file': None, 'yaml_file_encoding': None, 'toml_file': None, 'secrets_dir': None, 'protected_namespaces': ('model_', 'settings_')}

NODE_URLS: list[str]

TESTNET_NODE_URLS: list[str]

model_fields = {'NODE_URLS': FieldInfo(annotation=list[str], required=False, default=['wss://commune-api-node-0.communeai.net', 'wss://commune-api-node-1.communeai.net', 'wss://commune-api-node-2.communeai.net', 'wss://commune-api-node-3.communeai.net', 'wss://commune-api-node-4.communeai.net', 'wss://commune-api-node-5.communeai.net', 'wss://commune-api-node-6.communeai.net', 'wss://commune-api-node-7.communeai.net', 'wss://commune-api-node-8.communeai.net', 'wss://commune-api-node-9.communeai.net', 'wss://commune-api-node-10.communeai.net', 'wss://commune-api-node-11.communeai.net', 'wss://commune-api-node-12.communeai.net', 'wss://commune-api-node-13.communeai.net', 'wss://commune-api-node-14.communeai.net', 'wss://commune-api-node-15.communeai.net', 'wss://commune-api-node-16.communeai.net', 'wss://commune-api-node-17.communeai.net', 'wss://commune-api-node-18.communeai.net', 'wss://commune-api-node-19.communeai.net', 'wss://commune-api-node-20.communeai.net', 'wss://commune-api-node-21.communeai.net', 'wss://commune-api-node-22.communeai.net', 'wss://commune-api-node-23.communeai.net', 'wss://commune-api-node-24.communeai.net', 'wss://commune-api-node-25.communeai.net', 'wss://commune-api-node-26.communeai.net', 'wss://commune-api-node-27.communeai.net', 'wss://commune-api-node-28.communeai.net', 'wss://commune-api-node-29.communeai.net', 'wss://commune-api-node-30.communeai.net', 'wss://commune-api-node-31.communeai.net']), 'TESTNET_NODE_URLS': FieldInfo(annotation=list[str], required=False, default=['wss://testnet-commune-api-node-0.communeai.net'])}

model_computed_fields = {}

Inherited Members

pydantic_settings.main.BaseSettings: BaseSettings; settings_customise_sources
pydantic.main.BaseModel: model_extra; model_fields_set; model_construct; model_copy; model_dump; model_dump_json; model_json_schema; model_parametrized_name; model_post_init; model_rebuild; model_validate; model_validate_json; model_validate_strings; dict; json; parse_obj; parse_raw; parse_file; from_orm; construct; copy; schema; schema_json; validate; update_forward_refs

def get_node_url( comx_settings: ComxSettings | None = None, *, use_testnet: bool = False) -> str: View Source

56def get_node_url(
57    comx_settings: ComxSettings | None = None, *, use_testnet: bool = False
58) -> str:
59    comx_settings = comx_settings or ComxSettings()
60    match use_testnet:
61        case True:
62            node_url = random.choice(comx_settings.TESTNET_NODE_URLS)
63        case False:
64            node_url = random.choice(comx_settings.NODE_URLS)
65    return node_url

def get_available_nodes( comx_settings: ComxSettings | None = None, *, use_testnet: bool = False) -> list[str]: View Source

68def get_available_nodes(
69    comx_settings: ComxSettings | None = None, *, use_testnet: bool = False
70) -> list[str]:
71    comx_settings = comx_settings or ComxSettings()
72
73    match use_testnet:
74        case True:
75            node_urls = comx_settings.TESTNET_NODE_URLS
76        case False:
77            node_urls = comx_settings.NODE_URLS
78    return node_urls

class BalanceUnit(builtins.str, enum.Enum): View Source

81class BalanceUnit(str, Enum):
82    joule = "joule"
83    j = "j"
84    nano = "nano"
85    n = "n"

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to 'strict'.

joule = <BalanceUnit.joule: 'joule'>

j = <BalanceUnit.j: 'j'>

nano = <BalanceUnit.nano: 'nano'>

n = <BalanceUnit.n: 'n'>

Inherited Members

enum.Enum: name; value
builtins.str: encode; replace; split; rsplit; join; capitalize; casefold; title; center; count; expandtabs; find; partition; index; ljust; lower; lstrip; rfind; rindex; rjust; rstrip; rpartition; splitlines; strip; swapcase; translate; upper; startswith; endswith; removeprefix; removesuffix; isascii; islower; isupper; istitle; isspace; isdecimal; isdigit; isnumeric; isalpha; isalnum; isidentifier; isprintable; zfill; format; format_map; maketrans

def format_balance( balance: int, unit: BalanceUnit = <BalanceUnit.nano: 'nano'>) -> str: View Source

88def format_balance(balance: int, unit: BalanceUnit = BalanceUnit.nano) -> str:
89    """
90    Formats a balance.
91    """
92
93    match unit:
94        case BalanceUnit.nano | BalanceUnit.n:
95            return f"{balance}"
96        case BalanceUnit.joule | BalanceUnit.j:
97            in_joules = from_nano(balance)
98            round_joules = round(in_joules, 4)
99            return f"{round_joules:,} COMAI"

Formats a balance.

def intersection_update(base: dict[~K, ~V], update: dict[~K, ~Z]) -> Mapping[~K, Union[~V, ~Z]]: View Source

107def intersection_update(base: dict[K, V], update: dict[K, Z]) -> Mapping[K, V | Z]:
108    """
109    Update a dictionary with another dictionary, but only with keys that are already present.
110    """
111    updated = {k: update[k] for k in base if k in update}
112    return updated

Update a dictionary with another dictionary, but only with keys that are already present.

def transform_stake_dmap( stake_storage: dict[tuple[communex.types.Ss58Address, communex.types.Ss58Address], int]) -> dict[communex.types.Ss58Address, list[tuple[communex.types.Ss58Address, int]]]: View Source

115def transform_stake_dmap(stake_storage: dict[tuple[Ss58Address, Ss58Address], int]) -> dict[Ss58Address, list[tuple[Ss58Address, int]]]:
116    """
117    Transforms either the StakeTo or StakeFrom storage into the stake legacy data type.
118    """
119    transformed: dict[Ss58Address, list[tuple[Ss58Address, int]]] = defaultdict(list)
120    [transformed[k1].append((k2, v)) for (k1, k2), v in stake_storage.items()]
121
122    return dict(transformed)

Transforms either the StakeTo or StakeFrom storage into the stake legacy data type.