-
Notifications
You must be signed in to change notification settings - Fork 1
/
get_last_modified.py
executable file
·45 lines (34 loc) · 1.35 KB
/
get_last_modified.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
#!/usr/bin/env python3
import argparse
from collections import defaultdict
import datetime
import sys
from typing import Tuple, List
from common import client, load_query, sql_quote_path, NotFoundError
def print_table_data(data_source_id: int, table_path: Tuple[str, ...]):
table_data = client.execute(
load_query('queries/get_modified_at_for_table.gql'),
{'path': sql_quote_path((str(data_source_id),) + table_path)}
)
if table_data['table'] is None:
raise NotFoundError('Table "%s" not found in data source #%d' % (
'.'.join(table_path),
data_source_id,
))
data = table_data['table']
print('.'.join(table_path), data['lastModifiedAt'])
for created_by in data['createdBySql']:
print(' ', created_by['startedAt'])
print()
def main():
parser = argparse.ArgumentParser(description='Analyze downstream dependencies')
parser.add_argument('data_source_id', type=int)
parser.add_argument('full_table_names', type=str, nargs='*',
help='db.schema.name format, case-sensitive')
args = parser.parse_args()
# fast and dirty, doesn't handle quotes
table_paths = [tuple(path.split('.')) for path in args.full_table_names]
for path in table_paths:
print_table_data(args.data_source_id, path)
if __name__ == '__main__':
main()