initial_data.py
3.04 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
import os
import csv
from types import DictType
from sqlalchemy import (
Table,
MetaData,
)
from data.user import UserData
from data.routes import RouteData
from data.apps import AppsData
from data.unit import UnitData
from data.rekening import RekeningData
from data.pajak import PajakData
from data.group import GroupData
from data.group_route import GroupRouteData
from data.jns_pajak import JnsPajakData
from data.wilayah import WilayahData
from DbTools import (
get_pkeys,
execute,
set_sequence,
split_tablename,
)
from ..models import (
Base,
BaseModel,
CommonModel,
DBSession,
User,
)
fixtures = [
('users', UserData),
('routes', RouteData),
('units', UnitData),
('rekenings', RekeningData),
('pajaks', PajakData),
('groups', GroupData),
('groups_routes_permissions', GroupRouteData),
('jns_pajaks', JnsPajakData),
('wilayahs', WilayahData),
]
def insert():
insert_(fixtures)
def insert_(fixtures):
engine = DBSession.bind
metadata = MetaData(engine)
tablenames = []
for tablename, data in fixtures:
if 'csv' in data:
data['data'] = csv2fixture(data['csv'])
if tablename == 'users':
T = User
table = T.__table__
else:
schema, tablename_ = split_tablename(tablename)
table = Table(tablename_, metadata, autoload=True, schema=schema)
class T(Base, BaseModel, CommonModel):
__table__ = table
if type(data) == DictType:
options = data['options']
data = data['data']
else:
options = []
if 'insert if not exists' not in options:
q = DBSession.query(T).limit(1)
if q.first():
continue
tablenames.append(tablename)
keys = get_pkeys(table)
for d in data:
filter_ = {}
for key in keys:
val = d[key]
filter_[key] = val
q = DBSession.query(T).filter_by(**filter_)
if q.first():
continue
tbl = T()
tbl.from_dict(d)
if tablename == 'users' and 'password' in d:
tbl.password = d['password']
DBSession.add(tbl)
DBSession.flush()
update_sequence(tablenames)
# Perbaharui nilai sequence
def update_sequence(tablenames):
engine = DBSession.bind
metadata = MetaData(engine)
for item in fixtures:
tablename = item[0]
if tablename not in tablenames:
continue
schema, tablename = split_tablename(tablename)
class T(Base):
__table__ = Table(tablename, metadata, autoload=True, schema=schema)
set_sequence(T)
# Fixture from CSV file
def csv2fixture(filename):
base_dir = os.path.split(__file__)[0]
filename = os.path.join(base_dir, 'data', filename)
csvfile = open(filename)
reader = csv.DictReader(csvfile)
data = list(reader)
csvfile.close()
return data