nnPXUskr

1import sqlite3
2from typing import Dict, List
3import json
4
5table1 = [
6    "name1",
7    "name2",
8    "name3",
9]
10
11table2 = [
12    "name4",
13    "name5",
14    "name6",
15]
16
17def extract_table(table : List[str], data) -> List[Dict]:
18    data_rows = []
19    for item in data:
20        try:
21            item_dict = json.loads(item)
22        except json.JSONDecodeError:
23            item_dict = {}
24        row = {column: item_dict.get(column, None) for column in table}
25        data_rows.append(row)
26    return data_rows
27
28def get_all_data(json_file):
29    with open(json_file, 'r', encoding='utf8') as handle:
30        data = json.load(handle)
31    data_table1 = extract_table(table1, data)
32    data_table2 = extract_table(table2, data)
33
34
35def create_database():
36    connection = sqlite3.connect("name.db")
37    cursor = connection.cursor()
38    cursor.execute(
39        """CREATE TABLE IF NOT EXISTS t_table1 (
40            name1 TEXT,
41            name2 TEXT,
42            name3 TEXT)"""
43    )
44
45    cursor.execute(
46        """ CREATE TABLE IF NOT EXISTS t_table2 (
47            name4 TEXT,
48            name5 TEXT,
49            name6 TEXT)"""
50    )
51    connection.commit()
52
53def load_data(table_name, table : List[Dict]):
54    try:
55        connection = sqlite3.connect("database.db")
56        cursor = connection.cursor()
57        cursor.execute(f"PRAGMA table_info{table_name}")
58        columns = [column[1] for column in cursor.fetchall()]
59        for item in table:
60            placeholders = ', '.join(['?'] * len(columns))
61            query = f"INSERT INTO {table_name} ({', '.join(columns)}) VALUES (){placeholders}"
62            data_values = [item.get(column) for column in columns]
63            cursor.execute(query, data_values)
64        connection.commit()
65        connection.close()
66    except sqlite3.Error as e:
67        print(f"Error occured: {e}")