import logging
from dr_data.static_strings import *
__author__ = AUTHOR
__copyright__ = COPYRIGHT
__license__ = LICENSE
_logger = logging.getLogger(__name__)
[docs]class Sql:
"""
Class that contains all of the SQL to execute
"""
[docs] @staticmethod
def build_columns_query(table_schema_name, table_name):
"""
Query that gets all columns by schema and table names
:return: query string
:rtype: str
"""
return """select
table_schema,
table_name,
column_name,
column_default,
is_nullable::boolean,
data_type,
udt_name,
col_description((table_schema || '."' || table_name || '"')::regclass, ordinal_position)
from information_schema.columns
where table_schema = '{schema_name}' and table_name = '{table_name}'
order by table_schema, table_name, ordinal_position
""".format(schema_name=table_schema_name, table_name=table_name)
[docs] @staticmethod
def build_tables_query(table_schema_name):
"""
Query that gets all table schema based off table schem.
:return: query string
:rtype: str
"""
return """select
table_schema,
table_name,
obj_description((table_schema || '."' || table_name || '"')::regclass, 'pg_class')
from information_schema.tables
where table_schema = '{name}'
order by table_schema, table_name
""".format(name=table_schema_name)
[docs] @staticmethod
def build_insertion_table_order():
"""
Query that gets the insertion order of the database
:return: query string
:rtype: str
"""
return """WITH RECURSIVE fkeys AS (
SELECT conrelid AS source,
confrelid AS target
FROM pg_constraint
WHERE contype = 'f'
),
tables AS (
(
SELECT oid AS table_name,
1 AS level,
ARRAY[oid] AS trail,
FALSE AS circular
FROM pg_class
WHERE relkind = 'r'
AND NOT relnamespace::regnamespace::text LIKE ANY
(ARRAY['pg_catalog', 'information_schema', 'pg_temp_%'])
EXCEPT
SELECT source,
1,
ARRAY[ source ],
FALSE
FROM fkeys
)
UNION ALL
SELECT fkeys.source,
tables.level + 1,
tables.trail || fkeys.source,
tables.trail @> ARRAY[fkeys.source]
FROM fkeys
JOIN tables ON tables.table_name = fkeys.target
WHERE cardinality(array_positions(tables.trail, fkeys.source)) < 2
),
ordered_tables AS (
SELECT DISTINCT ON (table_name)
table_name,
level,
circular
FROM tables
ORDER BY table_name, level DESC
)
SELECT table_name::regclass,
level
FROM ordered_tables
ORDER BY level, table_name
"""
[docs] @staticmethod
def build_column_constraints(table_schema_name, table_name, column_name):
"""
Query that gets the column constraints
:return: query string
:rtype: str
"""
return """select
coalesce(table_schema, referenced_schema) as table_schema,
coalesce(table_name, referenced_table) as table_name,
coalesce(column_name, referenced_column) as column_name,
constraint_schema,
constraint_name,
constraint_type,
check_clause,
referenced_schema,
referenced_table,
referenced_column
from information_schema.table_constraints
natural full join information_schema.key_column_usage
natural full join information_schema.check_constraints
inner join (
select
table_schema as referenced_schema,
table_name as referenced_table,
column_name as referenced_column,
constraint_name
from information_schema.constraint_column_usage
) as referenced_columns using (constraint_name)
where constraint_schema = '{schema_name}' and table_name = '{table_name}' and column_name = '{column_name}'
order by table_schema, table_name, ordinal_position
""".format(schema_name=table_schema_name, table_name=table_name, column_name=column_name)
[docs] @staticmethod
def build_values_from_type(type):
"""
Query that gets values from type
:return: query string
:rtype: str
"""
return """SELECT pg_type.typname AS enumtype,
pg_enum.enumlabel AS enumlabel
FROM pg_type
JOIN pg_enum
ON pg_enum.enumtypid = pg_type.oid
WHERE pg_type.typname = '{type}'""".format(type=type)
[docs] @staticmethod
def build_populate_insert(table_name, columns):
"""
Query that inserts table data
:return: query string
:rtype: str
"""
return """INSERT INTO "{table_name}"({columns} ) VALUES %s ON CONFLICT DO NOTHING""".format(table_name=table_name, columns=columns)
[docs] @staticmethod
def build_random_row(columns, table):
"""
Query that gets a random row from a table
:return: query string
:rtype: str
"""
return "SELECT {columns} FROM {table} ORDER BY random() LIMIT 1".format(columns=columns, table=table)
[docs] @staticmethod
def build_random_row_where(columns, table, query):
"""
Query that gets random row from a table with a where statement
:return: query string
:rtype: str
"""
return "SELECT {columns} FROM {table} {query} ORDER BY random() LIMIT 1".format(columns=columns, table=table, query=query)