def get_schema_dict()

in route.py [0:0]

• 41 lines of code
• 13 McCabe index (conditional complexity)

def get_schema_dict(db, kk=3):
    """
    Get database's schema, which is a dict with table name as key
    and list of column names as value
    :param db: database path
    :return: schema dict
    """ 
    data = dump_db_json_schema(db,db.split('/')[-1])
    tables = data['table_names_original']
    column_types = data['column_types']
    primary_keys = data['primary_keys']
    foreign_keys = data['foreign_keys']
    column_names = data['column_names_original']
    
    schema_dict = {
        'tables': {},
        'foreign_keys':[]
    }

    for i, table in enumerate(tables): 
        t = {}
        for j, c in enumerate(column_names):
            if c[0] == i:
                if j in primary_keys:
                    t[c[1]] = [column_types[j].upper(), True]
                else:
                    t[c[1]] = [column_types[j].upper(), True]
        schema_dict['tables'][table] = t

    for foreign_key in foreign_keys:  
        t1 = tables[column_names[foreign_key[0]][0]]
        c1 = column_names[foreign_key[0]][1] 
        t2 = tables[column_names[foreign_key[1]][0]] 
        c2 = column_names[foreign_key[1]][1] 
        schema_dict['foreign_keys'].append([t1,c1,t2,c2]) 

    conn = sqlite3.connect(db)
    cursor = conn.cursor()
    # get exapmles
    for table in schema_dict['tables'].keys():
        try:
            select_query = f'SELECT * FROM `{table}` LIMIT {kk}' 
            cursor.execute(select_query)
            rows = cursor.fetchall() 
            cursor.execute(f"PRAGMA table_info(`{table}`);")
            columns = [column[1] for column in cursor.fetchall() ]  
            for i, c in enumerate(columns): 
                cls_valuse = [f"{row[i][0:100]}..." if type(row[i]) is str and len(row[i]) > 100 else row[i]  for row in rows] 
                schema_dict['tables'][table][c].append(cls_valuse)
        except Exception as e:
            print(e)
    return schema_dict