-
Notifications
You must be signed in to change notification settings - Fork 5
/
functions.py
188 lines (157 loc) · 5.91 KB
/
functions.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
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
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
""" functions.py
REPOSITORY: https://github.com/DavidJLambert/Python-Universal-DB-Client
AUTHOR: David J. Lambert
VERSION: 0.7.6
DATE: Jul 9, 2020
"""
import sys
from traceback import print_exception
from MyQueries import NOT_IMPLEMENTED, NOT_POSSIBLE_SQL
from platform import uname, python_implementation
from struct import calcsize
from os import pathsep, environ, scandir
from os.path import exists
from subprocess import Popen, PIPE
def print_stacktrace() -> None:
""" Print a stack trace, then resume.
Used everywhere.
Parameters:
Returns:
"""
print()
print_exception(*sys.exc_info(), limit=None, file=sys.stdout)
return
# End of function print_stacktrace.
def is_skip_operation(sql: str) -> bool:
""" Detect whether to skip this operation and operations dependent on it
Only used in DBClient.py.
Parameters:
sql (str): SQL from MyQueries.py.
Returns:
skip (bool): skip this operation and operations dependent on it.
"""
skip = (sql in {NOT_IMPLEMENTED, NOT_POSSIBLE_SQL})
return skip
# End of function is_skip_operation.
def os_python_version_info() -> (str, int, int):
""" Method to print OS and Python version info.
Only used in UniversalClient*.py.
Parameters:
Returns:
os (str): 'Windows', 'Linux', or 'Darwin'
Python major version (int): 2 or 3
Python minor version (int)
"""
q = ('OS: {}\nHost Name: {}\nOS Major Version: {}\nOS Full Version: {}'
'\nProcessor Type: {}\nProcessor: {}')
u = uname()
q = q.format(u.system, u.node, u.release, u.version, u.machine, u.processor)
print(q)
sys_version_info = sys.version_info
py_bits = 8 * calcsize("P")
py_version = '.'.join(str(x) for x in sys_version_info)
py_type = python_implementation()
print('\n{} Version {}, {} bits.'.format(py_type, py_version, py_bits))
return u.system, sys_version_info[0], sys_version_info[1]
# End of function os_python_version_info.
def is_file_in_path(os: str, filename: str) -> bool:
""" Method to find if file in PATH.
Only used in DBInstance.py.
Parameters:
os (str): 'Windows', 'Linux', or 'Darwin'
filename (str): the name of the file to find in PATH.
Returns:
found (bool): whether or not the file was found in PATH.
"""
filename = filename.lower()
if os == 'Windows':
filename += '.exe'
found = False
for folder in environ['PATH'].split(pathsep):
if folder not in {'', '.'}:
# Ignore invalid folders in PATH.
if exists(folder):
for f in scandir(folder):
if f.is_file():
if f.name.lower() == filename:
found = True
break
return found
# End of function is_file_in_path.
def pick_one(choices: list, choice_type: str) -> int:
""" Function to print message if skipping operation in
db_table_schema or db_view_schema.
Only used in DBClient.py.
Parameters:
choices (list): list of choices (string) to pick one from.
choice_type (str): the name of the type being chosen.
Returns:
choice (int): the choice made, or -1 if no choices are available.
"""
if len(choices) == 0:
choice = -1
print('You own no {}s! Nothing to see!'.format(choice_type))
elif len(choices) == 1:
# Only one table. Choose it.
choice = 0
choice_name = choices[choice]
print('\nYou have one {}: {}.'.format(choice_type, choice_name))
else:
# Ask end-user to choose a table.
prompt = '\nHere are the {} available to you:\n'.format(choice_type)
for item_num, choice_name in enumerate(choices):
prompt += str(item_num + 1) + ': ' + choice_name + '\n'
prompt += ('Enter the number for the {} you want info about,\n'
'Or enter "Q" to quit:\n'.format(choice_type))
# Keep looping until valid choice made.
while True:
choice = input(prompt).strip().upper()
# Interpret the choice.
if choice == "Q":
print('Quitting as requested.')
# Not cleaning up connections and cursors.
exit(0)
choice = int(choice) - 1
if choice in range(len(choices)):
choice_name = choices[choice]
break
else:
print('Invalid choice, please try again.')
print('\nYou chose {} "{}":'.format(choice_type, choice_name))
return choice
# End of function pick_one.
def sql_cmdline(cmdline_list: list, sql: str) -> list:
""" Run SQL against database using command line client.
Parameters:
cmdline_list (list): the list of commands for Popen.
sql (str): text of the SQL to run.
Returns:
sql_output (list): rows of output.
"""
if cmdline_list[0] == 'Error':
print(cmdline_list[1])
return list()
p = Popen(cmdline_list, stdin=PIPE, stdout=PIPE, stderr=PIPE)
(stdout, stderr) = p.communicate(sql.encode('utf-8'))
stderr = stderr.decode('utf-8').split("\n")
# MySQL warning to ignore.
z = [('WARNING: Using a password on the command line interface can be '
'insecure.'), '']
if len(stderr) > 0 and stderr != [''] and stderr != z:
print("PROBLEM IN SQL_CMDLINE:")
print('cmd: ', cmdline_list)
print('sql: ', sql)
print('stderr: ', stderr)
return stdout.decode('utf-8').split("\n")
# End of function sql_cmdline.
def quote_a_string(quote_me: str) -> str:
""" Enclose a string in single quotes, after escaping slashes and single
quotes.
Parameters:
quote_me (str): the string to quote.
Returns:
output (list): the string enclosed in single quotes.
"""
quote_me2 = quote_me.replace("'", "''")
return "'" + quote_me2 + "'"
# End of function quote_a_string.