IntenseWebs/pgadmin4
3
0
Fork 0
mirror of https://github.com/pgadmin-org/pgadmin4.git synced 2025-02-25 18:55:31 -06:00
Code Issues Packages Projects Releases Wiki Activity

Fixed cognitive complexity issues reported by SonarQube.

Browse Source
This commit is contained in:
Nikhil Mohite  2020-08-06 12:38:35 +05:30 committed by Akshay Joshi
parent e3dfe03a2a
commit 1122e06f48
3 changed files with 306 additions and 155 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

123
web/pgadmin/browser/server_groups/servers/databases/foreign_data_wrappers/__init__.py
View File

@ -587,6 +587,78 @@ class ForeignDataWrapperView(PGChildNodeView):
except Exception as e:
return internal_server_error(errormsg=str(e))
def _get_create_sql(self, data):
"""
Get sql for create foreign data wrapper.
:param data: Data.
:return: Create sql statement for foreign data wrapper.
"""
for key in ['fdwacl']:
if key in data and data[key] is not None:
data[key] = parse_priv_to_db(data[key], ['U'])
# Allow user to set the blank value in
# fdwvalue field in option model
is_valid_options = False
if 'fdwoptions' in data:
is_valid_options, data['fdwoptions'] = validate_options(
data['fdwoptions'], 'fdwoption', 'fdwvalue'
)
sql = render_template("/".join([self.template_path,
self._CREATE_SQL]),
data=data, conn=self.conn,
is_valid_options=is_valid_options
)
return sql
def _check_and_parse_priv_to_db(self, data):
"""
Check foreign data wrapper privilege and parse privileges before
sending to database.
:param data: Data.
"""
for key in ['fdwacl']:
if key in data and data[key] is not None:
if 'added' in data[key]:
data[key]['added'] = parse_priv_to_db(
data[key]['added'],
['U']
)
if 'changed' in data[key]:
data[key]['changed'] = parse_priv_to_db(
data[key]['changed'],
['U']
)
if 'deleted' in data[key]:
data[key]['deleted'] = parse_priv_to_db(
data[key]['deleted'],
['U']
)
def _check_validate_options(self, data):
"""
Check option in foreign data wrapper Data and This will call
validated options function to set flag for option valid or not in sql.
:param data: Data.
"""
is_valid_added_options = is_valid_changed_options = False
if 'fdwoptions' in data and 'added' in data['fdwoptions']:
is_valid_added_options, data['fdwoptions']['added'] = \
validate_options(
data['fdwoptions']['added'],
'fdwoption',
'fdwvalue')
if 'fdwoptions' in data and 'changed' in data['fdwoptions']:
is_valid_changed_options, data['fdwoptions']['changed'] = \
validate_options(
data['fdwoptions']['changed'],
'fdwoption',
'fdwvalue')
return is_valid_added_options, is_valid_changed_options
def get_sql(self, gid, sid, data, did, fid=None):
"""
This function will generate sql from model data.
@ -625,23 +697,7 @@ class ForeignDataWrapperView(PGChildNodeView):
'fdwoption', 'fdwvalue'
)
for key in ['fdwacl']:
if key in data and data[key] is not None:
if 'added' in data[key]:
data[key]['added'] = parse_priv_to_db(
data[key]['added'],
['U']
)
if 'changed' in data[key]:
data[key]['changed'] = parse_priv_to_db(
data[key]['changed'],
['U']
)
if 'deleted' in data[key]:
data[key]['deleted'] = parse_priv_to_db(
data[key]['deleted'],
['U']
)
self._check_and_parse_priv_to_db(data)
old_data = res['rows'][0]
for arg in required_args:
@ -650,19 +706,8 @@ class ForeignDataWrapperView(PGChildNodeView):
# Allow user to set the blank value in fdwvalue
# field in option model
is_valid_added_options = is_valid_changed_options = False
if 'fdwoptions' in data and 'added' in data['fdwoptions']:
is_valid_added_options, data['fdwoptions']['added'] = \
validate_options(
data['fdwoptions']['added'],
'fdwoption',
'fdwvalue')
if 'fdwoptions' in data and 'changed' in data['fdwoptions']:
is_valid_changed_options, data['fdwoptions']['changed'] = \
validate_options(
data['fdwoptions']['changed'],
'fdwoption',
'fdwvalue')
is_valid_added_options, \
is_valid_changed_options = self._check_validate_options(data)
sql = render_template(
"/".join([self.template_path, self._UPDATE_SQL]),
@ -674,23 +719,7 @@ class ForeignDataWrapperView(PGChildNodeView):
)
return sql, data['name'] if 'name' in data else old_data['name']
else:
for key in ['fdwacl']:
if key in data and data[key] is not None:
data[key] = parse_priv_to_db(data[key], ['U'])
# Allow user to set the blank value in
# fdwvalue field in option model
is_valid_options = False
if 'fdwoptions' in data:
is_valid_options, data['fdwoptions'] = validate_options(
data['fdwoptions'], 'fdwoption', 'fdwvalue'
)
sql = render_template("/".join([self.template_path,
self._CREATE_SQL]),
data=data, conn=self.conn,
is_valid_options=is_valid_options
)
sql = self._get_create_sql(data)
return sql, data['name']

158
web/pgadmin/browser/server_groups/servers/databases/schemas/foreign_tables/__init__.py
View File

@ -1101,10 +1101,10 @@ class ForeignTableView(PGChildNodeView, DataTypeReader,
Returns:
"""
SQL = render_template("/".join([self.template_path,
sql = render_template("/".join([self.template_path,
self._PROPERTIES_SQL]),
scid=scid, foid=foid)
status, res = self.conn.execute_dict(SQL)
status, res = self.conn.execute_dict(sql)
if not status:
return False, internal_server_error(errormsg=res)
@ -1117,10 +1117,10 @@ class ForeignTableView(PGChildNodeView, DataTypeReader,
if self.manager.version >= 90200:
# Fetch privileges
SQL = render_template("/".join([self.template_path,
sql = render_template("/".join([self.template_path,
self._ACL_SQL]),
foid=foid)
status, aclres = self.conn.execute_dict(SQL)
status, aclres = self.conn.execute_dict(sql)
if not status:
return False, internal_server_error(errormsg=aclres)
@ -1136,24 +1136,40 @@ class ForeignTableView(PGChildNodeView, DataTypeReader,
data.update({'strftoptions': data['ftoptions']})
data.update(self._parse_variables_from_db(data['ftoptions']))
SQL = render_template("/".join([self.template_path,
sql = render_template("/".join([self.template_path,
self._GET_CONSTRAINTS_SQL]), foid=foid)
status, cons = self.conn.execute_dict(SQL)
status, cons = self.conn.execute_dict(sql)
if not status:
return False, internal_server_error(errormsg=cons)
if cons and 'rows' in cons:
data['constraints'] = cons['rows']
SQL = render_template("/".join([self.template_path,
sql = render_template("/".join([self.template_path,
self._GET_COLUMNS_SQL]), foid=foid)
status, cols = self.conn.execute_dict(SQL)
status, cols = self.conn.execute_dict(sql)
if not status:
return False, internal_server_error(errormsg=cols)
# The Length and the precision of the Datatype should be separated.
# The Format we getting from database is: numeric(1,1)
# So, we need to separate it as Length: 1, Precision: 1
self._get_datatype_precision(cols)
if cols and 'rows' in cols:
data['columns'] = cols['rows']
# Get Inherited table names from their OID
is_error, errmsg = self._get_inherited_table_name(data, inherits)
if is_error:
return False, internal_server_error(errormsg=errmsg)
return True, data
def _get_datatype_precision(self, cols):
"""
The Length and the precision of the Datatype should be separated.
The Format we getting from database is: numeric(1,1)
So, we need to separate it as Length: 1, Precision: 1
:param cols: list of columns.
"""
for c in cols['rows']:
if c['fulltype'] != '' and c['fulltype'].find("(") > 0:
substr = self.extract_type_length_precision(c)
@ -1170,27 +1186,30 @@ class ForeignTableView(PGChildNodeView, DataTypeReader,
att_opt = self._parse_variables_from_db(c['attfdwoptions'])
c['coloptions'] = att_opt['ftoptions']
if cols and 'rows' in cols:
data['columns'] = cols['rows']
# Get Inherited table names from their OID
def _get_inherited_table_name(self, data, inherits):
"""
Get inherited table name.
:param data: Data.
:param inherits: flag which is used If True then inherited table
will be fetched from database.
"""
if inherits and 'inherits' in data and data['inherits']:
inherits = tuple([int(x) for x in data['inherits']])
if len(inherits) == 1:
inherits = "(" + str(inherits[0]) + ")"
SQL = render_template("/".join([self.template_path,
sql = render_template("/".join([self.template_path,
self._GET_TABLES_SQL]),
attrelid=inherits)
status, res = self.conn.execute_dict(SQL)
status, res = self.conn.execute_dict(sql)
if not status:
return False, internal_server_error(errormsg=res)
return True, res
if 'inherits' in res['rows'][0]:
data['inherits'] = res['rows'][0]['inherits']
return True, data
return False, ''
@staticmethod
def convert_precision_to_int(typlen):
@ -1424,6 +1443,18 @@ class ForeignTableView(PGChildNodeView, DataTypeReader,
return ajax_response(response=sql)
@staticmethod
def _check_const_for_obj_compare(data):
"""
Check for constraint in fetched objects for compare.
:param data: Data.
"""
if 'constraints' in data and data['constraints'] is not None \
and len(data['constraints']) > 0:
for item in data['constraints']:
if 'conoid' in item:
item.pop('conoid')
@check_precondition
def fetch_objects_to_compare(self, sid, did, scid):
"""
@ -1446,12 +1477,7 @@ class ForeignTableView(PGChildNodeView, DataTypeReader,
status, data = self._fetch_properties(0, sid, did, scid,
row['oid'])
if status:
if 'constraints' in data and data['constraints'] is not None \
and len(data['constraints']) > 0:
for item in data['constraints']:
if 'conoid' in item:
item.pop('conoid')
ForeignTableView._check_const_for_obj_compare(data)
res[row['name']] = data
return res
@ -1489,6 +1515,57 @@ class ForeignTableView(PGChildNodeView, DataTypeReader,
json_resp=False)
return sql
@staticmethod
def _modify_column_data(data, tmp_columns):
"""
Modifies data for column.
:param data: Data for columns.
:param tmp_columns: tmp_columns list.
"""
if 'added' in data['columns']:
for item in data['columns']['added']:
tmp_columns.append(item)
if 'changed' in data['columns']:
for item in data['columns']['changed']:
tmp_columns.append(item)
if 'deleted' in data['columns']:
for item in data['columns']['deleted']:
tmp_columns.remove(item)
@staticmethod
def _modify_constraints_data(data):
"""
Modifies data for constraints.
:param data: Data for constraints.
:return: tmp_constraints list.
"""
tmp_constraints = []
if 'added' in data['constraints']:
for item in data['constraints']['added']:
tmp_constraints.append(item)
if 'changed' in data['constraints']:
for item in data['constraints']['changed']:
tmp_constraints.append(item)
return tmp_constraints
@staticmethod
def _modify_options_data(data, tmp_ftoptions):
"""
Modifies data for options.
:param data: Data for options.
:param tmp_ftoptions: tmp_ftoptions list.
"""
if 'added' in data['ftoptions']:
for item in data['ftoptions']['added']:
tmp_ftoptions.append(item)
if 'changed' in data['ftoptions']:
for item in data['ftoptions']['changed']:
tmp_ftoptions.append(item)
if 'deleted' in data['ftoptions']:
for item in data['ftoptions']['deleted']:
tmp_ftoptions.remove(item)
def modify_data_for_schema_diff(self, data, old_data):
"""
This function modifies the data for columns, constraints, options
@ -1499,41 +1576,22 @@ class ForeignTableView(PGChildNodeView, DataTypeReader,
tmp_columns = []
if 'columns_for_schema_diff' in old_data:
tmp_columns = old_data['columns_for_schema_diff']
if 'columns' in data:
if 'added' in data['columns']:
for item in data['columns']['added']:
tmp_columns.append(item)
if 'changed' in data['columns']:
for item in data['columns']['changed']:
tmp_columns.append(item)
if 'deleted' in data['columns']:
for item in data['columns']['deleted']:
tmp_columns.remove(item)
ForeignTableView._modify_column_data(data, tmp_columns)
data['columns'] = tmp_columns
tmp_constraints = []
if 'constraints' in data:
if 'added' in data['constraints']:
for item in data['constraints']['added']:
tmp_constraints.append(item)
if 'changed' in data['constraints']:
for item in data['constraints']['changed']:
tmp_constraints.append(item)
tmp_constraints = ForeignTableView._modify_constraints_data(data)
data['constraints'] = tmp_constraints
tmp_ftoptions = []
if 'ftoptions' in old_data:
tmp_ftoptions = old_data['ftoptions']
if 'ftoptions' in data:
if 'added' in data['ftoptions']:
for item in data['ftoptions']['added']:
tmp_ftoptions.append(item)
if 'changed' in data['ftoptions']:
for item in data['ftoptions']['changed']:
tmp_ftoptions.append(item)
if 'deleted' in data['ftoptions']:
for item in data['ftoptions']['deleted']:
tmp_ftoptions.remove(item)
ForeignTableView._modify_options_data(data, tmp_ftoptions)
data['ftoptions'] = tmp_ftoptions

180
web/pgadmin/browser/server_groups/servers/databases/schemas/tables/constraints/index_constraint/utils.py
View File

@ -108,6 +108,82 @@ def get_index_constraints(conn, did, tid, ctype, cid=None, template_path=None):
return True, result['rows']
def _get_sql_to_delete_constraints(data, constraint, sql, template_path, conn):
"""
Check for delete constraints.
:param data: data.
:param constraint: constraint according to it's type from data.
:param sql: sql list for all sql statements.
:param template_path: Template path.
:param conn: connection.
:return:
"""
if 'deleted' in constraint:
for c in constraint['deleted']:
del_cols = []
if 'columns_to_be_dropped' in data:
del_cols = list(map(lambda x, y: x['column'] in y,
c['columns'],
data['columns_to_be_dropped'])
)
if len(del_cols) == 0:
c['schema'] = data['schema']
c['table'] = data['name']
# Sql for drop
sql.append(render_template("/".join([template_path,
'delete.sql']),
data=c,
conn=conn).strip('\n'))
def _get_sql_to_change_constraints(did, tid, ctype, data, constraint,
sql, conn):
"""
Check for chnage constraints.
:param did: data base id.
:param tid: table id.
:param ctype: constraint type.
:param data: data.
:param constraint: constraint according to it's type from data.
:param sql: sql list for all sql statements.
:param conn: connection.
:return:
"""
if 'changed' in constraint:
for c in constraint['changed']:
c['schema'] = data['schema']
c['table'] = data['name']
modified_sql, name = get_sql(conn, c, did, tid, ctype,
c['oid'])
if modified_sql:
sql.append(modified_sql.strip('\n'))
def _get_sql_to_add_constraints(did, tid, ctype, data, constraint,
sql, conn):
"""
Check for add constraints.
:param did: data base id.
:param tid: table id.
:param ctype: constraint type.
:param data: data.
:param constraint: constraint according to it's type from data.
:param sql: sql list for all sql statements.
:param conn: connection.
:return:
"""
if 'added' in constraint:
for c in constraint['added']:
c['schema'] = data['schema']
c['table'] = data['name']
add_sql, name = get_sql(conn, c, did, tid, ctype)
sql.append(add_sql.strip("\n"))
@get_template_path
def get_index_constraint_sql(conn, did, tid, data, template_path=None):
"""
@ -131,41 +207,14 @@ def get_index_constraint_sql(conn, did, tid, data, template_path=None):
if index_constraints[ctype] in data:
constraint = data[index_constraints[ctype]]
# If constraint(s) is/are deleted
if 'deleted' in constraint:
for c in constraint['deleted']:
del_cols = []
if 'columns_to_be_dropped' in data:
del_cols = list(map(lambda x, y: x['column'] in y,
c['columns'],
data['columns_to_be_dropped'])
)
if len(del_cols) == 0:
c['schema'] = data['schema']
c['table'] = data['name']
# Sql for drop
sql.append(render_template("/".join([template_path,
'delete.sql']),
data=c,
conn=conn).strip('\n'))
if 'changed' in constraint:
for c in constraint['changed']:
c['schema'] = data['schema']
c['table'] = data['name']
modified_sql, name = get_sql(conn, c, did, tid, ctype,
c['oid'])
if modified_sql:
sql.append(modified_sql.strip('\n'))
if 'added' in constraint:
for c in constraint['added']:
c['schema'] = data['schema']
c['table'] = data['name']
add_sql, name = get_sql(conn, c, did, tid, ctype)
sql.append(add_sql.strip("\n"))
_get_sql_to_delete_constraints(data, constraint, sql,
template_path, conn)
# Get SQL for change constraints.
_get_sql_to_change_constraints(did, tid, ctype, data, constraint,
sql, conn)
# Get SQL for add constraints.
_get_sql_to_add_constraints(did, tid, ctype, data, constraint,
sql, conn)
if len(sql) > 0:
# Join all the sql(s) as single string
@ -174,6 +223,40 @@ def get_index_constraint_sql(conn, did, tid, data, template_path=None):
return None
def is_key_str(key, data):
return isinstance(data[key], str) and data[key] != ""
def _check_required_args(data, name):
"""
Check required arguments are present.
:param data: Data for check.
:param name: constraint name.
:return: If any error return error.
"""
required_args = [
[u'columns', u'index'] # Either of one should be there.
]
def is_key_list(key, data):
return isinstance(data[key], list) and len(data[param]) > 0
for arg in required_args:
if isinstance(arg, list):
for param in arg:
if param in data and \
(is_key_str(param, data) or
is_key_list(param, data)):
break
else:
return True, '-- definition incomplete', name
elif arg not in data:
return True, '-- definition incomplete', name
return False, '', name
@get_template_path
def get_sql(conn, data, did, tid, ctype, cid=None, template_path=None):
"""
@ -197,7 +280,7 @@ def get_sql(conn, data, did, tid, ctype, cid=None, template_path=None):
if not status:
raise ExecuteError(res)
if len(res['rows']) == 0:
elif len(res['rows']) == 0:
raise ObjectGone(
_('Could not find the constraint in the table.'))
@ -209,28 +292,9 @@ def get_sql(conn, data, did, tid, ctype, cid=None, template_path=None):
data=data,
o_data=old_data)
else:
required_args = [
[u'columns', u'index'] # Either of one should be there.
]
def is_key_str(key, data):
return isinstance(data[key], str) and data[key] != ""
def is_key_list(key, data):
return isinstance(data[key], list) and len(data[param]) > 0
for arg in required_args:
if isinstance(arg, list):
for param in arg:
if param in data and \
(is_key_str(param, data) or
is_key_list(param, data)):
break
else:
return _('-- definition incomplete'), name
elif arg not in data:
return _('-- definition incomplete'), name
is_error, errmsg, name = _check_required_args(data, name)
if is_error:
return _(errmsg), name
sql = render_template("/".join([template_path, 'create.sql']),
data=data,