-
Notifications
You must be signed in to change notification settings - Fork 0
/
oracle.py
128 lines (118 loc) · 5.32 KB
/
oracle.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
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
# -*- coding: utf-8 -*-
import cx_Oracle
from itertools import chain
import wx
import logging
logging.basicConfig(filename='journal_events.log',format='%(asctime)s %(levelname)s %(message)s',level=logging.DEBUG)
class WorkDB():
def __init__(self, connection):
self.connection = connection
def get_tables(self, schema):
try:
cursor = cx_Oracle.Cursor(self.connection)
# Получать все таблицы доступные пользователю.
sql = ("select TABLE_NAME from dba_tables WHERE OWNER = '%s'" % schema)
cursor.execute(sql)
tables = cursor.fetchall()
tables=[i[0] for i in tables]
cursor.close()
return tables
except Exception, info:
info = str(info)
info = info.decode('cp1251').encode('utf8')
wx.MessageBox(u'Внешняя ошибка базы данных:%s - code 25' % info)
logging.error(u'loading list of tables failed - code 26:', str(info))
def get_schemas(self):
try:
cursor = cx_Oracle.Cursor(self.connection)
sql = ("select USERNAME from dba_users")
cursor.execute(sql)
schemas = cursor.fetchall()
schemas=[i[0] for i in schemas]
cursor.close()
return schemas
except TypeError, info:
wx.MessageBox(u'Подключитесь к базе данных!')
def get_all_count(self, schema, table):
try:
cursor = cx_Oracle.Cursor(self.connection)
sql = ("select count(*) from %s.%s") % (schema, table)
cursor.execute(sql)
data=cursor.fetchall()
count = data[0][0]
cursor.close()
return count
except (cx_Oracle.DatabaseError, cx_Oracle.DataError), info:
info = str(info)
info = info.decode('cp1251').encode('utf8')
wx.MessageBox(u'Внешняя ошибка базы данных: %s - code 52' % info)
def get_regexp_count(self, schema, table, regexp):
try:
cursor = cx_Oracle.Cursor(self.connection)
sql = (u"select count(*) from %s.%s where %s") % (schema, table, regexp)
cursor.execute(sql)
data=cursor.fetchall()
count = data[0][0]
cursor.close()
return count
except (cx_Oracle.DatabaseError, cx_Oracle.DataError), info:
info = str(info)
info = info.decode('cp1251').encode('utf8')
wx.MessageBox(u'Внешняя ошибка базы данных: %s - code 66' % info)
def get_empty_values(self, schema, table):
try:
cursor = cx_Oracle.Cursor(self.connection)
sql = ("select * from %s.%s" % (schema, table))
cursor.execute(sql)
data = cursor.fetchall()
cursor.close()
data = list(chain(*data))
all_data = len(data)
DWEV = filter(bool, data)
EV = all_data - len(DWEV)
EV = float(EV)
return EV
except (cx_Oracle.DatabaseError, cx_Oracle.DataError), info:
info = str(info)
info = info.decode('cp1251').encode('utf8')
wx.MessageBox(u'Внешняя ошибка базы данных: %s - code 84' % info)
def get_uniq_values(self, column, schema, table):
try:
cursor = cx_Oracle.Cursor(self.connection)
sql = "select DISTINCT(count(%s)) from %s.%s" % (column, schema, table)
cursor.execute(sql)
count = cursor.fetchall()
cou = count[0][0]
cursor.close()
return cou
except (cx_Oracle.DatabaseError, cx_Oracle.DataError), info:
info = str(info)
info = info.decode('cp1251').encode('utf8')
wx.MessageBox(u'Внешняя ошибка базы данных - code 98')
def get_cols(self, table):
try:
cursor = cx_Oracle.Cursor(self.connection)
sql = ("select t.COLUMN_ID, t.COLUMN_NAME from all_tab_columns t where t.TABLE_NAME=\'%s\' order by t.COLUMN_ID") % table
cursor.execute(sql)
cuu=cursor.fetchall()
cursor.close()
col_names = [i[1] for i in cuu]
except (NameError, cx_Oracle.DatabaseError), info:
info = str(info)
info = info.decode('cp1251').encode('utf8')
error = ("Database Error: %s - code 111" % info)
wx.MessageBox(str(error))
return col_names
def get_date_table(self, table, schema):
try:
cursor = cx_Oracle.Cursor(self.connection)
sql = ("select created from dba_objects where owner=\'%s\' and object_name=\'%s\' and object_type=\'TABLE\'" % (schema, table))
cursor.execute(sql)
count = cursor.fetchall()
date = count[0][0]
cursor.close()
except (cx_Oracle.DatabaseError, cx_Oracle.DataError), info:
info = str(info)
info = info.decode('cp1251').encode('utf8')
wx.MessageBox(u'Внешняя ошибка базы данных - code 126')
return date