initial_data.py
3.24 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
115
116
117
118
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.sektor import SektorData
from data.rekening import RekeningData
from data.pajak import PajakData
from data.produk import ProdukData
from data.group import GroupData
from data.peruntukan import PeruntukanData
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),
('sektors', SektorData),
('rekenings', RekeningData),
('pajaks', PajakData),
('peruntukans', PeruntukanData),
('produks', ProdukData),
('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