xcc.es
[READ-ONLY] Mirror of https://github.com/mrgnw/scripts.
1#!/usr/bin/env python3
2
3import argparse
4from pathlib import Path
5import sqlglot
6from sqlglot.optimizer import optimize
7from watchdog.observers import Observer
8from watchdog.events import FileSystemEventHandler
9
10
11class SqlFileChangeHandler(FileSystemEventHandler):
12 def __init__(self, src, tgt):
13 self.src_path = Path(f'{src}.sql').resolve()
14 self.tgt_path = Path(f'{tgt}.sql').resolve()
15 self.src = src
16 self.tgt = tgt
17
18 def on_modified(self, event):
19 if Path(event.src_path).resolve() == self.src_path:
20 self.transpile_and_format_sql()
21
22 def transpile_and_format_sql(self):
23 with open(self.src_path, 'r') as file:
24 sql = file.read()
25
26 optimized = optimize(sqlglot.parse_one(sql)).sql(pretty=True)
27 transpiled = '\n'.join(sqlglot.transpile(optimized, read=self.src, write=self.tgt, pretty=True))
28
29 with open(self.tgt_path, 'w') as file:
30 file.write(transpiled)
31
32 print(f'\t{self.src_path.name} ▸ {self.tgt_path.name}')
33
34
35def main():
36 parser = argparse.ArgumentParser(description='Monitor and transpile SQL files.')
37 parser.add_argument('src_dialect', default='duckdb', nargs='?', help='Source SQL dialect')
38 parser.add_argument('tgt_dialect', default='mysql', nargs='?', help='Target SQL dialect')
39
40 args = parser.parse_args()
41 src = args.src_dialect
42 tgt = args.tgt_dialect
43 Path(f'{src}.sql').touch()
44
45 event_handler = SqlFileChangeHandler(src, tgt)
46 observer = Observer()
47 observer.schedule(event_handler, path=Path(src).parent)
48 observer.start()
49
50 print(f'Monitoring {src}.sql for changes...')
51 try:
52 while True:
53 pass
54 except KeyboardInterrupt:
55 observer.stop()
56 observer.join()
57
58
59if __name__ == '__main__':
60 main()