import uuid
import airflow
from datetime import datetime
from airflow import DAG
from airflow.utils.trigger_rule import TriggerRule
from airflow.operators.postgres_operator import PostgresOperator
from airflow.hooks.postgres_hook import PostgresHook
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import PythonOperator

dag_params = {
    'owner': 'airflow',
    'dag_id': 'insert_db',
    'start_date': airflow.utils.dates.days_ago(2)
    
}

dag = DAG(
    'insert_db',
    schedule_interval="@once",
    default_args=dag_params
    )

create_table = PostgresOperator(
    task_id='insert_table',
    postgres_conn_id='postgres_default',
    sql='''
        INSERT INTO product
SELECT id, type, name, ppu, batters, topping
FROM json_populate_recordset (NULL::product,
    '[{
      	"id": "0001",
        "type": "donut",
        "name": "Cake",
        "ppu": 0.55,
      	"batters": {"batter":
                    [
                        { "id": "1001", "type": "Regular" },
                        { "id": "1002", "type": "Chocolate" },
                        { "id": "1003", "type": "Blueberry" },
                        { "id": "1004", "type": "Devil''s Food" }
                    ]},
        
        "topping":[
                { "id": "5001", "type": "None" },
                { "id": "5002", "type": "Glazed" },
                { "id": "5005", "type": "Sugar" },
                { "id": "5007", "type": "Powdered Sugar" },
                { "id": "5006", "type": "Chocolate with Sprinkles" },
                { "id": "5003", "type": "Chocolate" },
                { "id": "5004", "type": "Maple" }
            ]
                
    },
{
      	"id": "0002",
        "type": "donut",
        "name": "Raised",
        "ppu": 0.55,
        "batters":
            {
                "batter":
                    [
                        { "id": "1001", "type": "Regular" }
                    ]
            },
        "topping":
            [
                { "id": "5001", "type": "None" },
                { "id": "5002", "type": "Glazed" },
                { "id": "5005", "type": "Sugar" },
                { "id": "5003", "type": "Chocolate" },
                { "id": "5004", "type": "Maple" }
            ]   
    },
	{
		"id": "0003",
        "type": "donut",
        "name": "Old Fashioned",
        "ppu": 0.55,
        "batters":
            {
                "batter":
                    [
                        { "id": "1001", "type": "Regular" },
                        { "id": "1002", "type": "Chocolate" }
                    ]
            },
        "topping":
            [
                { "id": "5001", "type": "None" },
                { "id": "5002", "type": "Glazed" },
                { "id": "5003", "type": "Chocolate" },
                { "id": "5004", "type": "Maple" }
            ]						  
	}
	]'
);''',
        dag=dag
)


start = DummyOperator(
    task_id="start",
    dag=dag
)

start >> insert_table