def main()

in common_components/monitoring/dbt_perf.py [0:0]

• 29 lines of code
• 2 McCabe index (conditional complexity)

def main():
    """Extract DBT JSON performance graph from historical (dbt_log) or live data"""

    parser = argparse.ArgumentParser(description='Extract DBT performance results')
    parser.add_argument('--project_id', type=str, default=os.getenv('PROJECT_ID'),
                        help='Project for DBT log')
    parser.add_argument('--dataset_id', type=str, required=True,
                        help='Dataset for DBT log')
    parser.add_argument('--table_id', type=str, default='dbt_log',
                        help='Table ID for DBT Log')
    parser.add_argument('--num', type=int, default=10,
                        help='Number of recent invocations to list (if none fetched)')
    parser.add_argument('--live', action='store_true',
                        help='Query INFORMATION_SCHEMA instead of dbt log (region required)')
    parser.add_argument('dbt_invocation_id', type=str, nargs='?',
                        help='DBT invocation id to extract')

    args = parser.parse_args()

    client = bigquery.Client()

    if not args.dbt_invocation_id:
        dbt_recent_invocations(client,
                               args.project_id,
                               args.dataset_id,
                               args.table_id,
                               args.num)

    else:
        dbt_dump_invocation(client,
                            args.project_id,
                            args.dataset_id,
                            args.table_id,
                            args.dbt_invocation_id,
                            args.live)