communex._common
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)
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 frommodel_config['env_file']
should be used. You can also passNone
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 toNone
. - _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 toNone
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 isNone
. 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 isFalse
, and "None" if _cli_avoid_json isTrue
. - _cli_hide_none_type: Hide
None
values in CLI help text. Defaults toFalse
. - _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
.
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
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
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
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'.
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
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.
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.
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.