DataWriterJsonToDB.py

from typing import List #, Dict, Optional
import json
import sqlite3

def setup_database():
    """Initialize SQLite database"""
    try:
        conn = sqlite3.connect('plants.db')
        c = conn.cursor()
        c.execute('''CREATE TABLE IF NOT EXISTS plants (
            name TEXT, scientific_name TEXT, alternate_names TEXT, 
            description TEXT, plant_family TEXT, origin TEXT, growth_habitat TEXT,
            active_components TEXT, treatable_conditions TEXT, preparation_methods TEXT,
            dosage TEXT, duration TEXT, contraindications TEXT, side_effects TEXT,
            interactions TEXT, part_used TEXT, harvesting_time TEXT, storage_tips TEXT,
            images TEXT, related_videos TEXT, sources TEXT
        )''')
        conn.commit()
        print("Database created successfully!")
        conn.close()
    except sqlite3.Error as e:
        print(f"An error occurred: {e}")


def write_to_database():
    """Save extracted data to SQLite"""
    try:
        conn = sqlite3.connect('plants.db')
        c = conn.cursor()
        with open("plants_data.json", 'r') as f:
            mapped_data = json.load(f)
            # Prepare the insert statement
            insert_sql = '''INSERT INTO plants (name, scientific_name, alternate_names, description, plant_family, origin, growth_habitat, active_components, treatable_conditions, preparation_methods, dosage, duration, contraindications, side_effects, interactions, part_used, harvesting_time, storage_tips, images, related_videos, sources ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)'''
            for each in mapped_data:
                c.execute(insert_sql, (
                each.get('Name', ''),
                each.get('Scientific Name', ''),
                each.get('Alternate Names', ''),
                each.get('Description', ''),
                each.get('Plant Family', ''),
                each.get('Origin', ''),
                each.get('Growth Habitat', ''),
                each.get('Active Components', ''),
                each.get('Treatable Conditions', ''),
                each.get('Preparation Methods', ''),
                each.get('Dosage', ''),
                each.get('Duration', ''),
                each.get('Contraindications', ''),
                each.get('Side Effects', ''),
                each.get('Interactions', ''),
                each.get('Part Used', ''),
                each.get('Harvesting Time', ''),
                each.get('Storage Tips', ''),
                each.get('Images', ''),
                each.get('Related Videos', ''),
                each.get('Sources', '')
            ))        
            conn.commit()
            conn.close()
    except sqlite3.Error as e:
        print(f"An error occurred: {e}")
        
def view_database()->List:
    conn = sqlite3.connect('plants.db')
    # Create a cursor object
    cursor = conn.cursor()
    # Execute a query to retrieve data
    cursor.execute("SELECT * FROM plants")
    # Fetch all results
    rows = cursor.fetchall()
    # Print the results
    plants=[]
    for row in rows:
        plants.append(row)
    # Close the connection
    conn.close()
    return plants

setup_database()
write_to_database()