# Databricks notebook source


# COMMAND ----------

# MAGIC %run ./pg_Connect

# COMMAND ----------

import json
context = json.loads(dbutils.notebook.entry_point.getDbutils().notebook().getContext().toJson())
#context
print(f'currentRunId: {context["currentRunId"]}')
run_id = context["currentRunId"] if context["currentRunId"] is not None else {'id': 0}
run_id = run_id['id']
print(run_id)

dbutils.widgets.text("Extract_id","","")
extract_id=dbutils.widgets.get("Extract_id")
print("Extract_id is : ",extract_id)
print(extract_id)

print(f"{extract_id} LOGGING: registering run_id: {run_id}")
query = f"""
    with s as (
    SELECT extract_instance_id, extract_reg_id, composer_run_id 
    FROM lcef_rule.extract_instance 
    WHERE composer_run_id = '{run_id}' and extract_reg_id = {extract_id}
    ), i as (
    INSERT INTO lcef_rule.extract_instance
        (extract_reg_id, composer_run_id, status,audit_insert_ts)
    SELECT {extract_id}, '{run_id}' , 'Started', NOW()
        where not exists (select 1 from s)
        returning extract_instance_id, {extract_id}, '{run_id}'
    )
    select extract_instance_id, {extract_id}, '{run_id}'
    from i
    union all
    select extract_instance_id, {extract_id}, '{run_id}'
    from s    

"""


print(query)
result = execute_sql(query)
extract_instance_id = result[0]['extract_instance_id']
print(extract_instance_id)

# COMMAND ----------

dbutils.jobs.taskValues.set(key="run_id",value = extract_instance_id)
dbutils.jobs.taskValues.get(taskKey="Register_Run",key="run_id",debugValue="abc123")
