-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathsql2dart.py
More file actions
114 lines (89 loc) · 2.97 KB
/
Copy pathsql2dart.py
File metadata and controls
114 lines (89 loc) · 2.97 KB
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
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
import typing as T
from contextlib import redirect_stdout
from pathlib import Path
import click
from .core import SQLFunctionsLexer, SQLFunctionsParser, SQLFunction, QueryVar
HEADER = """\
//
// Generated by 'sql2dart'
//
import 'package:sqflite/sqflite.dart';
"""
@click.command()
@click.argument('SQL_FILES', nargs=-1, type=click.Path(exists=True))
@click.argument('DEST', type=click.Path(dir_okay=True))
def cli(sql_files: T.Sequence[str], dest: str):
lexer = SQLFunctionsLexer()
parser = SQLFunctionsParser()
if len(sql_files) < 1:
exit("No SQL files provided!")
dest = Path(dest)
if not dest.exists() and len(sql_files) > 1:
dest.mkdir(parents=True)
if not dest.is_dir():
dest.write_text(HEADER)
for sql_file in sql_files:
in_path = Path(sql_file)
if dest.is_dir():
out_path = dest / (in_path.stem + '.dart')
out_path.write_text(HEADER)
else:
out_path = dest
print(f"'{in_path}' -> '{out_path}'")
with open(out_path, 'a') as f, redirect_stdout(f):
for fn in parser.parse(lexer.tokenize(in_path.read_text())):
print_fn(fn)
def print_fn(fn: SQLFunction):
if len(fn.body) > 1:
ret_type = 'Future<List<dynamic>>'
else:
_, ret_type = choose_fn(fn.body[0])
print(
'%s %s(Database db, %s) async {'
% (ret_type, fn.name, ''.join(f'{i},' for i in fn.params))
)
print('return await db.transaction((txn) async {')
if len(fn.body) > 1:
print('final batch = txn.batch();')
call_prefix = 'batch'
else:
print('return ', end='')
call_prefix = 'await txn'
for sql_stmt in fn.body:
print_fn_call(call_prefix, sql_stmt, fn)
if len(fn.body) > 1:
print('return await batch.commit();')
print('});}')
def print_fn_call(
call_prefix: str, sql_stmt: T.List[T.Union[str, QueryVar]], fn: SQLFunction
):
dart_fn, _ = choose_fn(sql_stmt)
print(f'{call_prefix}.{dart_fn}("""', end='')
args = []
for fragment in sql_stmt:
if isinstance(fragment, QueryVar):
var = fragment.name
if var not in fn.params:
raise ValueError(
f"The parameter '{var}' was not declared in the signature '{fn.name}({', '.join(fn.params)})'!"
)
args.append(var)
if fragment.is_quoted:
print(" '?' ", end='')
else:
print(' ? ', end='')
else:
print(f'{fragment}', end='')
print(f'""", [{", ".join(args)}],);')
def choose_fn(sql_stmt):
first = sql_stmt[0].strip().upper()[:6]
try:
return SQL_2_DART_FN[first]
except KeyError:
return 'execute', 'Future<void>'
SQL_2_DART_FN = {
'SELECT': ('rawQuery', 'Future<List<Map<String, dynamic>>>'),
'INSERT': ('rawInsert', 'Future<int>'),
'UPDATE': ('rawUpdate', 'Future<int>'),
'DELETE': ('rawDelete', 'Future<int>'),
}