Task 3: Store the new events data in Postgres.

airflow_usgs_task_three.py

def save_events_to_db(**context):
    """ Inserts the events to database. """
    insert_query = """
        INSERT INTO public.earthquake_events (event_id, event_name, magnitude, longitude, latitude, date)
        VALUES (%s, %s, %s, %s, %s, %s); 
        """
    # pull event data from previous task instance using XCom
    events = context["ti"].xcom_pull(task_ids='get_new_events', key='events')
    for event in events:
        params = tuple(event.values())
        PostgresHook(
            postgres_conn_id="postgres_default", 
            schema="usgs_eq"
        ).run(insert_query, parameters=params)
        
        
# Task 3: Store the new events data in Postgres.
# The PythonOperator calls the Python Function defined above.
task_three = PythonOperator(
    task_id = 'save_events_to_db',
    python_callable = save_events_to_db,
    provide_context = True
)