def dbt_recent_invocations()

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

• 11 lines of code
• 3 McCabe index (conditional complexity)

def dbt_recent_invocations(client, project_id, dataset_id, table_id, num):
    """Display recent DBT invocations"""

    full_table_id = f'{project_id}.{dataset_id}.{table_id}'
    results = client.query(dbt_recent_invocations_query(full_table_id, num))
    print('dbt_invocation_id                    start_date          end_date')
    for row in results:
        end_time = ''
        if row['end_time']:
            end_time = row['end_time'].strftime(TIME_FORMAT)
        print(row['dbt_invocation_id'],
              row['start_time'].strftime(TIME_FORMAT),
              end_time)