Divvy JSONB accessor generator

jsonbSQLAccessorGenerator.py

import json 
import typing
import sys

'''
What this does:
Take in a representative jsonb blob and generate hierarchical sql accessors for 
all its fields. You can use this with a querry like: 
  select twi.workflow_state, twi.workflow_type  from temporal_workflow_indicies twi where twi.workflow_type = 'siteVisitWorkflow' limit 5;
to grab some representative JSON and then put it into Sigma with something like: 
  select
    CONCAT(p.street_address, ', ', p.city, ', ', p.state, ' ', p.zip_code) AS full_address, p.metro, p.asset_id,   
    -- insert script output here -- 
    twi.property_id, 
    CONCAT('https://admin.divvyhomes.com/business-process/asset-planning?selectedId=', twi.business_process_id) AS mission_control_url
  FROM temporal_workflow_indicies twi 
  JOIN properties p ON twi.property_id = p.id
  where twi.workflow_type = 'assetPlanningWorkflow' 
  --	and twi.updated_at >= CURRENT_DATE - INTERVAL '7 days'
  ORDER BY COALESCE(twi.completed_at AT TIME ZONE 'UTC', twi.updated_at AT TIME ZONE 'UTC') NULLS last
  ;
To generate a MVP reporting dashboard in 5 mins or less.

Usage: 
  python3 jsonbSQLAccessorGenerator.py 'representative json blob'

  specifically: 
  python3 jsonbSQLAccessorGenerator.py '{"type": "secureAndScope", "status": "awaiting_site_visit", "assetId": "dcfafa86-f943-4f78-822f-8caf850b66e4", "endTime": "8:30am", "gcalDate": "2023-10-02", "startTime": "8:00am", "propertyId": "dcfafa86-f943-4f78-822f-8caf850b66e4", "gcalEndTime": "08:30AM", "gcalEventUrl": "https://www.google.com/calendar/event?eid=c25vdGExa2wza2w1cmkxYnY0NTBodml2ZjQgY19hamJzdmc2NmNsNGVtYjFqMGRoNmZlOTYyOEBn", "eventMetadata": {"asset_plan": "Hold (Rent or ReDivvy)", "task_request": "Please take-back possession, scope for repairs, change locks & install the lockbox using the last four digits of the zip code.", "fotonotes_link": "https://api.sitecapture.com/app/#/project/show/6549541", "occupancy_check": "No"}, "gcalAttendees": ["galen.grimes@divvyhomes.com"], "gcalStartTime": "08:00AM", "localTimeZone": "EST", "requestedDate": "2023-10-01", "siteVisitDate": "2023-10-01", "siteVisitType": "secureAndScope", "businessProcessId": "ac1365f6-7b90-44e5-9511-eb01552db88b", "assignedFieldOperator": "dc72ba93-0b41-423a-9ded-dce58317f33a", "startedByBusinessProcessId": "5c913f5e-9807-4c89-8658-a56f77f9f8e2"}'
'''


def key_path_as_sql(path: list[str]):
  '''
  Format as a hierarchially indented sql getter from jsonb
  '''
  p = ", ".join([item for item in path])
  sqlStmt = """{whitespace}workflow_state#>>'{{{path}}}'""".format(whitespace = "  "*(len(path)-1), path=p)
  return sqlStmt


def path_builder(obj: dict, current_path: list = [], accum_paths = []):
  '''
  Extract the JSON structure into paths for Max-approved SQL jsonb extraction. 
  '''
  for key in obj.keys():
    if isinstance(obj[key], dict): 
      path_builder(obj[key], current_path + [key], accum_paths = accum_paths)
    else: 
      accum_paths.append(current_path+[key])
  return accum_paths

json_string = json_string = sys.argv[1]
json_data = json.loads(json_string)


paths = path_builder(json_data)

sql_stmts = [key_path_as_sql(p) for p in paths]
print(",\n".join(stmt for stmt in sql_stmts))