/
utils.py
34 lines (30 loc) · 1021 Bytes
/
utils.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
try:
import pysqlite3 as sqlite3
import pysqlite3.dbapi2
OperationalError = pysqlite3.dbapi2.OperationalError
except ImportError:
import sqlite3
OperationalError = sqlite3.OperationalError
def suggest_column_types(records):
all_column_types = {}
for record in records:
for key, value in record.items():
all_column_types.setdefault(key, set()).add(type(value))
column_types = {}
for key, types in all_column_types.items():
if len(types) == 1:
t = list(types)[0]
# But if it's list / tuple / dict, use str instead as we
# will be storing it as JSON in the table
if t in (list, tuple, dict):
t = str
elif {int, bool}.issuperset(types):
t = int
elif {int, float, bool}.issuperset(types):
t = float
elif {bytes, str}.issuperset(types):
t = bytes
else:
t = str
column_types[key] = t
return column_types