def get_tables_list()

in data_validation/cli_tools.py [0:0]

• 39 lines of code
• 7 McCabe index (conditional complexity)

def get_tables_list(arg_tables, default_value=None, is_filesystem=False):
    """Returns dictionary of tables. Backwards compatible for JSON input.

    arg_table (str): tables_list argument specified
    default_value (Any): A default value to supply when arg_value is empty.
    is_filesystem (boolean): Boolean indicating whether source connection is a FileSystem. In this case, a schema is not required.
    """
    if not arg_tables:
        return default_value

    json_tables_list = _read_json_value(arg_tables)
    if json_tables_list:
        return json_tables_list

    tables_list = []
    tables_mapping = list(csv.reader([arg_tables]))[0]
    source_schema_required = bool(not is_filesystem)

    for mapping in tables_mapping:
        tables_map = mapping.split("=")
        if len(tables_map) == 1:
            schema, table = split_table(
                tables_map, schema_required=source_schema_required
            )
            table_dict = {
                "schema_name": schema,
                "table_name": table,
            }
        elif len(tables_map) == 2:
            src_schema, src_table = split_table(
                [tables_map[0]], schema_required=source_schema_required
            )

            table_dict = {
                "schema_name": src_schema,
                "table_name": src_table,
            }

            targ_schema, targ_table = split_table(
                [tables_map[1]], schema_required=False
            )

            if targ_schema:
                table_dict["target_schema_name"] = targ_schema
            table_dict["target_table_name"] = targ_table

        else:
            raise ValueError(
                "Unable to parse tables list. Please provide valid mapping."
            )

        tables_list.append(table_dict)

    return tables_list