validation_db.py
5.33 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
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
from sqlalchemy.orm.attributes import InstrumentedAttribute
from .base_validation import BaseValidation
class ValidationDB(object):
__invalid_key = 'DB VALIDATION ERROR'
def __init__(self, inputname, value, params):
if not isinstance(params, dict):
self.__invalid_rule_value()
self.__inputname = inputname
self.__value = value
self.__params = params
# get value berdasarkan key dari params
def __get_key(self, key):
return key in self.__params and self.__params[key] or None
# raise error dgn argument msg
def __raise_invalid(self, msg):
raise Exception(self.__invalid_key, msg)
# raise error karena rule value salah format
def __invalid_rule_value(self):
self.__raise_invalid('Rule harus berupa "dictionary object".')
# raise error karena tidak ada key connection (database session) dalam params
def __invalid_db_session(self):
self.__raise_invalid('DB Session dibutuhkan dalam validasi.')
# raise error karena tidak ada key table dalam params
def __invalid_table(self):
self.__raise_invalid('Nama Table dibutuhkan dalam validasi.')
# raise error karena field tidak ada dalam table
def __invalid_field(self):
self.__raise_invalid('Field tidak ditemukan.')
# raise error karena format condition harus dict object
def __invalid_condition(self):
self.__raise_invalid('Condition value harus berupa "list". Contoh: [table.id == id, table.kode == kode]')
def __invalid_method(self):
self.__raise_invalid('Query method tidak dapat diproses.')
# get dbSession / connection
def __db_session(self):
val = self.__get_key('session')
if not val:
self.__invalid_db_session()
return val
# get tablename
def __db_table(self):
val = self.__get_key('table')
if not val:
self.__invalid_table()
return val
def __fix_field(self, field):
table = self.__db_table()
if not isinstance(field, InstrumentedAttribute):
if isinstance(field, str):
field = getattr(table, field, None)
else:
field = None
if not field:
self.__invalid_field()
# check existensi field di table
# ambil nama field stringnya
name = field.name
if (name == '(no name)') or (not name) or (not getattr(table, name, None)):
self.__invalid_field()
return field
# get field yang akan dicek uniknya
def __field(self):
field = self.__get_key('field') or self.__inputname
return self.__fix_field(field)
# get condition / filter dari table
def __conditions(self):
return self.__get_key('conditions')
# query berdasarkan field yang akan dicek uniknya
def query(self):
session = self.__db_session()
table = self.__db_table()
# field = self.__field()
# return session.query(table).filter(field == self.__value)
return session.query(table)
# query setelah diambil query dari field yang harus unik berdasarkan kondisi keunikanya
def query_condition(self, query=None):
method = self.__get_key('method')
if method != 'query':
if not query:
query = self.query()
cond = self.__conditions()
if cond:
if not isinstance(cond, list):
self.__invalid_condition()
# query = apply_filters(query, cond)
query = query.filter(*cond)
else:
field = self.__field()
query = query.filter(field == self.__value)
else:
query = self.__get_key('query')
if not query:
self.__invalid_method()
return query
# validasi untuk fields yg unique
class ValidationUniqueDB(BaseValidation, ValidationDB):
__invalid_key = 'VALIDATION "UNIQUE_DB" ERROR'
def __init__(self, cldr, params=None):
BaseValidation.__init__(self, cldr=cldr, params=params)
ValidationDB.__init__(self, inputname=self.inputname, value=self.value, params=self.rulevalue)
self.__message = self.rulemessage or ':attribute sudah digunakan.'
# return True jika data tidak ada. raise form (False) jika sudah ada
def validate(self):
query = self.query_condition()
data = query.first()
ok = (data is None)
if not ok:
self.colander_invalid[self.inputname] = self.__message.replace(':attribute', self.title)
return ok
# validasi untuk existensi data
class ValidationExistsDB(BaseValidation, ValidationDB):
__invalid_key = 'VALIDATION "EXIST_DB" ERROR'
def __init__(self, cldr, params=None):
BaseValidation.__init__(self, cldr=cldr, params=params)
ValidationDB.__init__(self, inputname=self.inputname, value=self.value, params=self.rulevalue)
self.__message = self.rulemessage or ':attribute tidak tersedia.'
# return True jika data ada. raise form (False) jika data tidak ada
def validate(self):
query = self.query_condition()
data = query.first()
ok = (data is not None)
if not ok:
self.colander_invalid[self.inputname] = self.__message.replace(':attribute', self.title)
return ok