Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@special_command('\\!', '\\! [command]', 'Pass commands to shell.')
def shell_command(cur, pattern, verbose):
cur, headers = [], []
params = shlex.split(pattern)
return [(None, cur, headers, subprocess.call(params))]
@special_command('\\dD', '\\dD[+] [pattern]', 'List or describe domains.')
def list_domains(cur, pattern, verbose):
if verbose:
extra_cols = r''',
pg_catalog.array_to_string(t.typacl, E'\n') AS "Access privileges",
d.description as "Description"'''
extra_joins = '''
LEFT JOIN pg_catalog.pg_description d ON d.classoid = t.tableoid
AND d.objoid = t.oid AND d.objsubid = 0'''
else:
extra_cols = extra_joins = ''
sql = '''\
SELECT n.nspname AS "Schema",
t.typname AS "Name",
pg_catalog.format_type(t.typbasetype, t.typtypmod) AS "Type",
pg_catalog.ltrim((COALESCE((SELECT (' collate ' || c.collname)
@special_command('\\dE', '\\dE[+] [pattern]', 'List foreign tables.', aliases=() )
def list_foreign_tables(cur, pattern, verbose):
if verbose:
verbose_cols = '''
, pg_catalog.pg_size_pretty(pg_catalog.pg_table_size(c.oid)) as "Size",
pg_catalog.obj_description(c.oid, 'pg_class') as "Description" '''
else:
verbose_cols = ''
if pattern:
_, tbl_name = sql_name_pattern(pattern)
filter = " AND c.relname OPERATOR(pg_catalog.~) '^(%s)$' " % tbl_name
else:
filter = ''
query = '''
@special_command('\\dF', '\\dF[+] [pattern]', 'List text search configurations.')
def list_text_search_configurations(cur, pattern, verbose):
def _find_text_search_configs(cur, pattern):
sql = '''
SELECT c.oid,
c.cfgname,
n.nspname,
p.prsname,
np.nspname AS pnspname
FROM pg_catalog.pg_ts_config c
LEFT JOIN pg_catalog.pg_namespace n ON n.oid = c.cfgnamespace,
pg_catalog.pg_ts_parser p
LEFT JOIN pg_catalog.pg_namespace np ON np.oid = p.prsnamespace
WHERE p.oid = c.cfgparser
'''
@special_command('\\dm', '\\dm[+] [pattern]', 'List materialized views.')
def list_materialized_views(cur, pattern, verbose):
return list_objects(cur, pattern, verbose, ['m', 's', ''])
@special_command('\\ds', '\\ds[+] [pattern]', 'List sequences.')
def list_sequences(cur, pattern, verbose):
return list_objects(cur, pattern, verbose, ['S', 's', ''])
@special_command('\\dv', '\\dv[+] [pattern]', 'List views.')
def list_views(cur, pattern, verbose):
return list_objects(cur, pattern, verbose, ['v', 's', ''])
@special_command('\\l', '\\l[+] [pattern]', 'List databases.', aliases=('\\list',))
def list_databases(cur, pattern, verbose):
query = '''SELECT d.datname as "Name",
pg_catalog.pg_get_userbyid(d.datdba) as "Owner",
pg_catalog.pg_encoding_to_char(d.encoding) as "Encoding",
d.datcollate as "Collate",
d.datctype as "Ctype",
pg_catalog.array_to_string(d.datacl, E'\n') AS "Access privileges"'''
if verbose:
query += ''',
CASE WHEN pg_catalog.has_database_privilege(d.datname, 'CONNECT')
THEN pg_catalog.pg_size_pretty(pg_catalog.pg_database_size(d.datname))
ELSE 'No Access'
END as "Size",
t.spcname as "Tablespace",
pg_catalog.shobj_description(d.oid, 'pg_database') as "Description"'''
query += '''
@special_command('\\nd', '\\nd [name]', 'Delete a named query.')
def delete_named_query(pattern, **_):
"""Delete an existing named query.
"""
usage = 'Syntax: \\nd name.\n\n' + NamedQueries.instance.usage
if not pattern:
return [(None, None, None, usage)]
status = NamedQueries.instance.delete(pattern)
return [(None, None, None, status)]
@special_command('\\ns', '\\ns name query', 'Save a named query.')
def save_named_query(pattern, **_):
"""Save a new named query.
Returns (title, rows, headers, status)"""
usage = 'Syntax: \\ns name query.\n\n' + NamedQueries.instance.usage
if not pattern:
return [(None, None, None, usage)]
name, _, query = pattern.partition(' ')
# If either name or query is missing then print the usage and complain.
if (not name) or (not query):
return [(None, None, None,
usage + 'Err: Both name and query are required.')]
NamedQueries.instance.save(name, query)