Function csv_export has a Cognitive Complexity of 51 (exceeds 5 allowed). Consider refactoring.

Open

def csv_export(table: Any, file: str, parameters: Dict = {}) -> Tuple:
    """Export DynamoDB table to csv

    Args:
        table (Table): boto3 DynamoDB table object

• Read up
  Read up

Function csv_import has a Cognitive Complexity of 33 (exceeds 5 allowed). Consider refactoring.

Open

def csv_import(table: Any, file: str, ignore: bool = False) -> Tuple:
    """csv import into DynamoDB table

    Args:
        table (Any): boto3 DynamoDB table object

• Read up
  Read up

Function execute has a Cognitive Complexity of 18 (exceeds 5 allowed). Consider refactoring.

Open

def execute() -> Tuple:
    """Command execute

    Raises:
        ValueError: invalid config

• Read up
  Read up

Function convert_item has a Cognitive Complexity of 11 (exceeds 5 allowed). Consider refactoring.

Open

def convert_item(spec: str, item: Dict, key: str) -> Any:
    """convert item

    Args:
        spec (str): type of item

• Read up
  Read up

Function write_to_dynamo has a Cognitive Complexity of 11 (exceeds 5 allowed). Consider refactoring.

Open

def write_to_dynamo(table: Any, rows: Dict, ignore: bool = False) -> None:
    """csv rows into DynamoDB

    Args:
        table (Any): boto3 DynamoDB table object

• Read up
  Read up

Function convert_column has a Cognitive Complexity of 10 (exceeds 5 allowed). Consider refactoring.

Open

def convert_column(spec: str, row: Dict, key: str) -> Any:
    """convert column

    Args:
        spec (str): type of column

• Read up
  Read up

Function move has a Cognitive Complexity of 9 (exceeds 5 allowed). Consider refactoring.

Open

def move(tables: List) -> Tuple:
    """DynamoDB table items move

    Args:
        tables (List): boto3 DynamoDB table objects

• Read up
  Read up

Avoid deeply nested control flow statements.

Open

                        if "ConvertBlankToDropAttrs" in csv_spec["IMPORT_OPTION"] and not row[key]:
                            to_drop_attrs = csv_spec.get("IMPORT_OPTION", "ConvertBlankToDropAttrs").split(",")
                            if key in to_drop_attrs:
                                del row[key]
                                continue

Avoid deeply nested control flow statements.

Open

                        if "ConvertBlankToNullAttrs" in csv_spec["IMPORT_OPTION"] and not row[key]:
                            to_null_attrs = csv_spec.get("IMPORT_OPTION", "ConvertBlankToNullAttrs").split(",")
                            if key in to_null_attrs:
                                row[key] = None
                                continue

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

Open

def truncate(table: Any) -> Tuple:
    """DynamoDB table truncate

    Args:
        table (Any): boto3 DynamoDB table object

• Read up
  Read up

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

Open

def config_read_and_get_table(args: Any) -> List:
    """Config read and Create DynamoDB table instance

    Args:
        args (Any): arguments

• Read up
  Read up

Avoid too many return statements within this function.

Open

        return json.loads(row[key], parse_float=Decimal)

Avoid too many return statements within this function.

Open

        return " ".join(list(map(str, item[key])))

Avoid too many return statements within this function.

Open

        return json.dumps(item[key], default=decimal_encode)

Avoid too many return statements within this function.

Open

        return row[key].split()

Avoid too many return statements within this function.

Open

        return item[key]

Avoid too many return statements within this function.

Open

        return row[key]

Avoid too many return statements within this function.

Open

        return (str(e), 1)

Avoid too many return statements within this function.

Open

        return set(row[key].split())

Avoid too many return statements within this function.

Open

        return list(map(Decimal, row[key].split()))