Function __init__ has 5 arguments (exceeds 4 allowed). Consider refactoring.

Open

    def __init__(self, name: str, rpc_address: str, testnet: bool, b64_private_key: str, authorization: Optional[str] = None):

Function make_broadcast_futures has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

def make_broadcast_futures(session: aiohttp.ClientSession, block_id: str, level: int, chain_ids: set) -> Optional[Set[asyncio.Task]]:
    """Initiate broadcasts for a block id to certain higher level nodes
    Args:
        session: aiohttp session to use for making http requests
        block_id: the block id to broadcast

• Read up
  Read up

Function sign_block has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

    def sign_block(self, signable_block: "model.BlockModel") -> str:
        """Sign a block with this class' keys
        Args:
            block: BlockModel to sign
        Returns:

• Read up
  Read up

Function hash_l3_block has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

def hash_l3_block(hash_type: SupportedHashes, block: "l3_block_model.L3BlockModel", nonce: int = 0) -> bytes:
    """Hash an l3 block
    Args:
        hash_type: SupportedHashes enum type
        block: L3BlockModel with appropriate data to hash

• Read up
  Read up

Function get has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

def get(key: str, cache_expire: Optional[int] = None, should_cache: bool = True) -> bytes:
    """Returns an object from storage
    Args:
        key: The key to get from storage
        cache_expire: The amount of time (in seconds) until the key expires if cache miss

• Read up
  Read up

Function start_task has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

def start_task() -> None:
    task_definition = get_next_task()
    if task_definition:
        job = get_existing_job_status(task_definition)
        if job and job.status.active:

• Read up
  Read up

Function verify_custom_indexes_options has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

def verify_custom_indexes_options(custom_indexes: Iterable["custom_index"]) -> None:
    """Validate an array of custom index DTOs from user input (Raises on error)"""
    for index in custom_indexes:
        index_type = index.get("type")
        if index_type == "text":

• Read up
  Read up

Function get_next_item has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

def get_next_item() -> Optional[Any]:
    """Get and json.loads the next item from the queue"""
    item = cast(bytes, redis.rpoplpush_sync(INCOMING_TX_KEY, PROCESSING_TX_KEY, decode=False))
    if item is not None:
        if LEVEL != "1":

• Read up
  Read up

Function get_registered_transaction_types_or_default has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

def get_registered_transaction_types_or_default(transaction_types: Iterable[str]) -> Dict[str, transaction_type_model.TransactionTypeModel]:
    """Bulk get of registered transaction types
       Note: If a transaction type is not found, it is sent back as a TransactionTypeModel with default values
    Args:
        transaction_types: a list of transaction types to fetch models for

• Read up
  Read up

Function bech32_decode has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

def bech32_decode(bech):
    """Validate a Bech32 string, and determine HRP and data."""
    if (any(ord(x) < 33 or ord(x) > 126 for x in bech)) or (bech.lower() != bech and bech.upper() != bech):
        return None, None
    bech = bech.lower()

• Read up
  Read up

Avoid too many return statements within this function.

Open

                return crypto.verify_l5_block_trust(self.hash, self.encryption, self.pub, cast("l5_block_model.L5BlockModel", block))

Avoid too many return statements within this function.

Open

    return hrp, data[:-6]

Avoid too many return statements within this function.

Open

                return crypto.verify_l4_block_trust(self.hash, self.encryption, self.pub, cast("l4_block_model.L4BlockModel", block))

Avoid too many return statements within this function.

Open

                return crypto.verify_l4_block_pow(self.hash, cast("l4_block_model.L4BlockModel", block))

Avoid too many return statements within this function.

Open

    return allowed

Avoid too many return statements within this function.

Open

            return crypto.sign_l5_block(self.hash, self.encryption, self.priv, signable_block)

Avoid too many return statements within this function.

Open

                return crypto.verify_l3_block_trust(self.hash, self.encryption, self.pub, cast("l3_block_model.L3BlockModel", block))

Avoid too many return statements within this function.

Open

                return crypto.verify_l3_block_pow(self.hash, cast("l3_block_model.L3BlockModel", block))

Function get_pending_l4_blocks has a Cognitive Complexity of 6 (exceeds 5 allowed). Consider refactoring.

Open

def get_pending_l4_blocks(block_id: str) -> List[str]:
    all_waiting_verification_keys = storage.list_objects(f"BROADCAST/TO_BROADCAST/{block_id}")

    l4_blocks = []
    for key in all_waiting_verification_keys:

• Read up
  Read up

Function bech32_polymod has a Cognitive Complexity of 6 (exceeds 5 allowed). Consider refactoring.

Open

def bech32_polymod(values):
    """Internal function that computes the Bech32 checksum."""
    generator = [0x3B6A57B2, 0x26508E6D, 0x1EA119FA, 0x3D4233DD, 0x2A1462B3]
    chk = 1
    for value in values:

• Read up
  Read up