diff --git a/py_dbcn/connectors/core/records.py b/py_dbcn/connectors/core/records.py
index 3305fab4fbf7ea92177378b2c109b7afc549e329..1bc484e2b6226ded7b518262180b73bdde6ba2d1 100644
--- a/py_dbcn/connectors/core/records.py
+++ b/py_dbcn/connectors/core/records.py
@@ -32,11 +32,18 @@ class BaseRecords:
# Define provided direct parent object.
self._parent = parent
- def select(self, table_name, select_clause=None, where_clause=None, display_query=True, display_results=True):
+ def select(
+ self,
+ table_name,
+ select_clause=None, where_clause=None, order_by_clause=None,
+ display_query=True, display_results=True,
+ ):
"""Selects records from provided table.
:param table_name: Name of table to select from.
:param select_clause: Clause to choose selected columns.
+ :param where_clause: Clause to limit selected records.
+ :param order_by_clause: Clause to adjust sort order of records.
:param display_query: Bool indicating if query should output to console. Defaults to True.
:param display_results: Bool indicating if results should output to console. Defaults to True.
"""
@@ -57,8 +64,11 @@ class BaseRecords:
if not self._base.validate.where_clause(where_clause):
raise ValueError('Invalid WHERE clause of "{0}".'.format(where_clause))
+ # Check that provided ORDER BY clause is valid format.
+ order_by_clause = self._base.validate.sanitize_order_by_clause(order_by_clause)
+
# Select record.
- query = 'SELECT {0} FROM {1}{2};'.format(select_clause, table_name, where_clause)
+ query = 'SELECT {0} FROM {1}{2}{3};'.format(select_clause, table_name, where_clause, order_by_clause)
results = self._base.query.execute(query, display_query=display_query)
if display_results:
self._base.display.records.select(results, logger, table_name, select_clause)
diff --git a/py_dbcn/connectors/core/validate.py b/py_dbcn/connectors/core/validate.py
index 059aee7430e8e74e5cf64c8fba15069a6ffbf61a..8178c51f5f0b53bf4816ead55bbafcf1281f1c12 100644
--- a/py_dbcn/connectors/core/validate.py
+++ b/py_dbcn/connectors/core/validate.py
@@ -371,14 +371,38 @@ class BaseValidate:
# For now, always return as valid.
return True
- def order_by_clause(self, clause):
+ def sanitize_order_by_clause(self, clause):
"""
Validates that provided clause follows acceptable format.
:param clause: ORDER_BY clause to validate.
:return: True if valid | False otherwise.
"""
+ # TODO: Implement proper sanitization checks.
+
+ if clause is None:
+ clause = ''
+
+ # Strip any potential whitespace.
+ clause = str(clause).strip()
+
+ # Handle if clause is not empty.
+ if clause != '':
+ # Remove prefix, if present.
+ if clause.lower().startswith('order by'):
+ clause = clause[8:]
+
+ # Strip again, with prefix removed.
+ clause = clause.strip()
+
+ # Prevent wildcard use.
+ if clause == '*':
+ raise ValueError('ORDER BY clause cannot use wildcard.')
+
+ # Put in expected format.
+ clause = ' ORDER BY {0}'.format(clause)
+
# For now, always return as valid.
- return True
+ return clause
# endregion Clause Validation
diff --git a/tests/connectors/core/test_records.py b/tests/connectors/core/test_records.py
index f7dd47da3bf2adc4c1055d6c76205c61190920fd..5a6984197b3c70aa74b525d6bdca7073f3a3aae0 100644
--- a/tests/connectors/core/test_records.py
+++ b/tests/connectors/core/test_records.py
@@ -206,6 +206,459 @@ class CoreRecordsTestMixin:
self.assertNotIn(row_1, results)
self.assertNotIn(row_3, results)
+ def test__select__with_order_by__success__by_one_col(self):
+ """
+ Test `SELECT` query when using order_by clauses on a single column.
+ """
+ table_name = 'test_queries__select__order_by__single__success'
+ # Verify table exists.
+ try:
+ self.connector.query.execute('CREATE TABLE {0}{1};'.format(table_name, self._columns_clause__basic))
+ except self.connector.errors.table_already_exists:
+ # Table already exists, as we want.
+ pass
+
+ with self.subTest('SELECT with ORDER BY when table has no records'):
+ # Mostly just making sure there are no errors in this case.
+
+ # Run test query.
+ results = self.connector.records.select(table_name, order_by_clause='description')
+
+ # Verify records returned in expected order.
+ self.assertEqual(len(results), 0)
+
+ # Insert record.
+ row_1 = (1, 'test_name_1', 'test_desc_1')
+ self.connector.query.execute('INSERT INTO {0} VALUES {1};'.format(table_name, row_1))
+
+ with self.subTest('SELECT with ORDER BY when table has one record'):
+ # Mostly just making sure there are no errors in this case.
+
+ # Run test query.
+ results = self.connector.records.select(table_name, order_by_clause='description')
+
+ # Verify records returned in expected order.
+ self.assertEqual(len(results), 1)
+ self.assertEqual(row_1, results[0])
+
+ # Insert record.
+ row_2 = (2, 'z name', 'z desc')
+ self.connector.query.execute('INSERT INTO {0} VALUES {1};'.format(table_name, row_2))
+
+ with self.subTest('SELECT with ORDER BY when table has two records - By Unspecified'):
+ # Run test query.
+ results = self.connector.records.select(table_name, order_by_clause='description')
+
+ # Verify records returned in expected order.
+ self.assertEqual(len(results), 2)
+ self.assertEqual(row_1, results[0])
+ self.assertEqual(row_2, results[1])
+
+ with self.subTest('SELECT with ORDER BY when table has two records - By ASC'):
+ # Run test query.
+ results = self.connector.records.select(table_name, order_by_clause='description ASC')
+
+ # Verify records returned in expected order.
+ self.assertEqual(len(results), 2)
+ self.assertEqual(row_1, results[0])
+ self.assertEqual(row_2, results[1])
+
+ with self.subTest('SELECT with ORDER BY when table has two records - By DESC'):
+ # Run test query.
+ results = self.connector.records.select(table_name, order_by_clause='description DESC')
+
+ # Verify records returned in expected order.
+ self.assertEqual(len(results), 2)
+ self.assertEqual(row_2, results[0])
+ self.assertEqual(row_1, results[1])
+
+ # Insert record.
+ row_3 = (3, 'a name', 'a desc')
+ self.connector.query.execute('INSERT INTO {0} VALUES {1};'.format(table_name, row_3))
+
+ with self.subTest('SELECT with ORDER BY when table has three records - By Unspecified'):
+ # Run test query.
+ results = self.connector.records.select(table_name, order_by_clause='description')
+
+ # Verify records returned in expected order.
+ self.assertEqual(len(results), 3)
+ self.assertEqual(row_3, results[0])
+ self.assertEqual(row_1, results[1])
+ self.assertEqual(row_2, results[2])
+
+ with self.subTest('SELECT with ORDER BY when table has three records - By ASC'):
+ # Run test query.
+ results = self.connector.records.select(table_name, order_by_clause='description ASC')
+
+ # Verify records returned in expected order.
+ self.assertEqual(len(results), 3)
+ self.assertEqual(row_3, results[0])
+ self.assertEqual(row_1, results[1])
+ self.assertEqual(row_2, results[2])
+
+ with self.subTest('SELECT with ORDER BY when table has three records - By DESC'):
+ # Run test query.
+ results = self.connector.records.select(table_name, order_by_clause='description DESC')
+
+ # Verify records returned in expected order.
+ self.assertEqual(len(results), 3)
+ self.assertEqual(row_2, results[0])
+ self.assertEqual(row_1, results[1])
+ self.assertEqual(row_3, results[2])
+
+ # Insert record.
+ row_4 = (4, 'the name', 'the desc')
+ self.connector.query.execute('INSERT INTO {0} VALUES {1};'.format(table_name, row_4))
+
+ with self.subTest('SELECT with ORDER BY when table has four records - By Unspecified'):
+ # Run test query.
+ results = self.connector.records.select(table_name, order_by_clause='description')
+
+ # Verify records returned in expected order.
+ self.assertEqual(len(results), 4)
+ self.assertEqual(row_3, results[0])
+ self.assertEqual(row_1, results[1])
+ self.assertEqual(row_4, results[2])
+ self.assertEqual(row_2, results[3])
+
+ with self.subTest('SELECT with ORDER BY when table has four records - By ASC'):
+ # Run test query.
+ results = self.connector.records.select(table_name, order_by_clause='description ASC')
+
+ # Verify records returned in expected order.
+ self.assertEqual(len(results), 4)
+ self.assertEqual(row_3, results[0])
+ self.assertEqual(row_1, results[1])
+ self.assertEqual(row_4, results[2])
+ self.assertEqual(row_2, results[3])
+
+ with self.subTest('SELECT with ORDER BY when table has four records - By DESC'):
+ # Run test query.
+ results = self.connector.records.select(table_name, order_by_clause='description DESC')
+
+ # Verify records returned in expected order.
+ self.assertEqual(len(results), 4)
+ self.assertEqual(row_2, results[0])
+ self.assertEqual(row_4, results[1])
+ self.assertEqual(row_1, results[2])
+ self.assertEqual(row_3, results[3])
+
+ # Insert record.
+ row_5 = (5, 'b name', 'b desc')
+ self.connector.query.execute('INSERT INTO {0} VALUES {1};'.format(table_name, row_5))
+
+ with self.subTest('SELECT with ORDER BY when table has five records - By Unspecified'):
+ # Run test query.
+ results = self.connector.records.select(table_name, order_by_clause='description')
+
+ # Verify records returned in expected order.
+ self.assertEqual(len(results), 5)
+ self.assertEqual(row_3, results[0])
+ self.assertEqual(row_5, results[1])
+ self.assertEqual(row_1, results[2])
+ self.assertEqual(row_4, results[3])
+ self.assertEqual(row_2, results[4])
+
+ with self.subTest('SELECT with ORDER BY when table has five records - By ASC'):
+ # Run test query.
+ results = self.connector.records.select(table_name, order_by_clause='description ASC')
+
+ # Verify records returned in expected order.
+ self.assertEqual(len(results), 5)
+ self.assertEqual(row_3, results[0])
+ self.assertEqual(row_5, results[1])
+ self.assertEqual(row_1, results[2])
+ self.assertEqual(row_4, results[3])
+ self.assertEqual(row_2, results[4])
+
+ with self.subTest('SELECT with ORDER BY when table has five records - By DESC'):
+ # Run test query.
+ results = self.connector.records.select(table_name, order_by_clause='description DESC')
+
+ # Verify records returned in expected order.
+ self.assertEqual(len(results), 5)
+ self.assertEqual(row_2, results[0])
+ self.assertEqual(row_4, results[1])
+ self.assertEqual(row_1, results[2])
+ self.assertEqual(row_5, results[3])
+ self.assertEqual(row_3, results[4])
+
+ def test__select__with_order_by__success__by_multiple_cols(self):
+ """
+ Test `SELECT` query when using order_by clauses on multiple column.
+ """
+ table_name = 'test_queries__select__order_by__multiple__success'
+ # Verify table exists.
+ try:
+ self.connector.query.execute('CREATE TABLE {0}{1};'.format(table_name, self._columns_clause__basic))
+ except self.connector.errors.table_already_exists:
+ # Table already exists, as we want.
+ pass
+
+ with self.subTest('SELECT with ORDER BY when table has no records'):
+ # Mostly just making sure there are no errors in this case.
+
+ # Run test query.
+ results = self.connector.records.select(table_name, order_by_clause='description, name')
+
+ # Verify records returned in expected order.
+ self.assertEqual(len(results), 0)
+
+ # Insert records.
+ row_1 = (1, 'test_name_1', 'test_desc_1')
+ self.connector.query.execute('INSERT INTO {0} VALUES {1};'.format(table_name, row_1))
+ row_2 = (2, 'This is a test name', 'Some desc')
+ self.connector.query.execute('INSERT INTO {0} VALUES {1};'.format(table_name, row_2))
+ row_3 = (3, 'aaa name', 'zzz desc')
+ self.connector.query.execute('INSERT INTO {0} VALUES {1};'.format(table_name, row_3))
+ row_4 = (4, 'zzz name', 'This is a test description')
+ self.connector.query.execute('INSERT INTO {0} VALUES {1};'.format(table_name, row_4))
+ row_5 = (5, 'some name', 'aaa desc')
+ self.connector.query.execute('INSERT INTO {0} VALUES {1};'.format(table_name, row_5))
+ row_6 = (6, 'test_name_1', 'test_desc_1')
+ self.connector.query.execute('INSERT INTO {0} VALUES {1};'.format(table_name, row_6))
+ # Duplicate records in all but pk. Guarantees there is multiple overlap with ordering fields.
+ row_7 = (7, 'test_name_1', 'test_desc_1')
+ self.connector.query.execute('INSERT INTO {0} VALUES {1};'.format(table_name, row_7))
+ row_8 = (8, 'This is a test name', 'Some desc')
+ self.connector.query.execute('INSERT INTO {0} VALUES {1};'.format(table_name, row_8))
+ row_9 = (9, 'aaa name', 'zzz desc')
+ self.connector.query.execute('INSERT INTO {0} VALUES {1};'.format(table_name, row_9))
+ row_10 = (10, 'zzz name', 'This is a test description')
+ self.connector.query.execute('INSERT INTO {0} VALUES {1};'.format(table_name, row_10))
+ row_11 = (11, 'some name', 'aaa desc')
+ self.connector.query.execute('INSERT INTO {0} VALUES {1};'.format(table_name, row_11))
+ row_12 = (12, 'test_name_1', 'test_desc_1')
+ self.connector.query.execute('INSERT INTO {0} VALUES {1};'.format(table_name, row_12))
+
+ with self.subTest('SELECT with ORDER BY - By name, id - ASC'):
+ # Run test query.
+ results = self.connector.records.select(table_name, order_by_clause='name, id')
+
+ # Verify records returned in expected order.
+ self.assertEqual(len(results), 12)
+ self.assertEqual(row_3, results[0])
+ self.assertEqual(row_9, results[1])
+ self.assertEqual(row_5, results[2])
+ self.assertEqual(row_11, results[3])
+ self.assertEqual(row_1, results[4])
+ self.assertEqual(row_6, results[5])
+ self.assertEqual(row_7, results[6])
+ self.assertEqual(row_12, results[7])
+ self.assertEqual(row_2, results[8])
+ self.assertEqual(row_8, results[9])
+ self.assertEqual(row_4, results[10])
+ self.assertEqual(row_10, results[11])
+
+ with self.subTest('SELECT with ORDER BY - By name, id - DESC'):
+ # Run test query.
+ results = self.connector.records.select(table_name, order_by_clause='name DESC, id DESC')
+
+ # Verify records returned in expected order.
+ self.assertEqual(len(results), 12)
+ self.assertEqual(row_10, results[0])
+ self.assertEqual(row_4, results[1])
+ self.assertEqual(row_8, results[2])
+ self.assertEqual(row_2, results[3])
+ self.assertEqual(row_12, results[4])
+ self.assertEqual(row_7, results[5])
+ self.assertEqual(row_6, results[6])
+ self.assertEqual(row_1, results[7])
+ self.assertEqual(row_11, results[8])
+ self.assertEqual(row_5, results[9])
+ self.assertEqual(row_9, results[10])
+ self.assertEqual(row_3, results[11])
+
+ with self.subTest('SELECT with ORDER BY - By name, id - MIXED'):
+ # Run test query.
+ results = self.connector.records.select(table_name, order_by_clause='name DESC, id ASC')
+
+ # Verify records returned in expected order.
+ self.assertEqual(len(results), 12)
+ self.assertEqual(row_4, results[0])
+ self.assertEqual(row_10, results[1])
+ self.assertEqual(row_2, results[2])
+ self.assertEqual(row_8, results[3])
+ self.assertEqual(row_1, results[4])
+ self.assertEqual(row_6, results[5])
+ self.assertEqual(row_7, results[6])
+ self.assertEqual(row_12, results[7])
+ self.assertEqual(row_5, results[8])
+ self.assertEqual(row_11, results[9])
+ self.assertEqual(row_3, results[10])
+ self.assertEqual(row_9, results[11])
+
+ with self.subTest('SELECT with ORDER BY - By description, id - ASC'):
+ # Run test query.
+ results = self.connector.records.select(table_name, order_by_clause='description, id')
+
+ # Verify records returned in expected order.
+ self.assertEqual(len(results), 12)
+ self.assertEqual(row_5, results[0])
+ self.assertEqual(row_11, results[1])
+ self.assertEqual(row_2, results[2])
+ self.assertEqual(row_8, results[3])
+ self.assertEqual(row_1, results[4])
+ self.assertEqual(row_6, results[5])
+ self.assertEqual(row_7, results[6])
+ self.assertEqual(row_12, results[7])
+ self.assertEqual(row_4, results[8])
+ self.assertEqual(row_10, results[9])
+ self.assertEqual(row_3, results[10])
+ self.assertEqual(row_9, results[11])
+
+ with self.subTest('SELECT with ORDER BY - By description, id - DESC'):
+ # Run test query.
+ results = self.connector.records.select(table_name, order_by_clause='description DESC, id DESC')
+
+ # Verify records returned in expected order.
+ self.assertEqual(len(results), 12)
+ self.assertEqual(row_9, results[0])
+ self.assertEqual(row_3, results[1])
+ self.assertEqual(row_10, results[2])
+ self.assertEqual(row_4, results[3])
+ self.assertEqual(row_12, results[4])
+ self.assertEqual(row_7, results[5])
+ self.assertEqual(row_6, results[6])
+ self.assertEqual(row_1, results[7])
+ self.assertEqual(row_8, results[8])
+ self.assertEqual(row_2, results[9])
+ self.assertEqual(row_11, results[10])
+ self.assertEqual(row_5, results[11])
+
+ with self.subTest('SELECT with ORDER BY - By description, id - MIXED'):
+ # Run test query.
+ results = self.connector.records.select(table_name, order_by_clause='description DESC, id ASC')
+
+ # Verify records returned in expected order.
+ self.assertEqual(len(results), 12)
+ self.assertEqual(row_3, results[0])
+ self.assertEqual(row_9, results[1])
+ self.assertEqual(row_4, results[2])
+ self.assertEqual(row_10, results[3])
+ self.assertEqual(row_1, results[4])
+ self.assertEqual(row_6, results[5])
+ self.assertEqual(row_7, results[6])
+ self.assertEqual(row_12, results[7])
+ self.assertEqual(row_2, results[8])
+ self.assertEqual(row_8, results[9])
+ self.assertEqual(row_5, results[10])
+ self.assertEqual(row_11, results[11])
+
+ with self.subTest('SELECT with ORDER BY - By id, name, description - ASC'):
+ # Run test query.
+ results = self.connector.records.select(table_name, order_by_clause='id, name, description')
+
+ # Verify records returned in expected order.
+ self.assertEqual(len(results), 12)
+ self.assertEqual(row_1, results[0])
+ self.assertEqual(row_2, results[1])
+ self.assertEqual(row_3, results[2])
+ self.assertEqual(row_4, results[3])
+ self.assertEqual(row_5, results[4])
+ self.assertEqual(row_6, results[5])
+ self.assertEqual(row_7, results[6])
+ self.assertEqual(row_8, results[7])
+ self.assertEqual(row_9, results[8])
+ self.assertEqual(row_10, results[9])
+ self.assertEqual(row_11, results[10])
+ self.assertEqual(row_12, results[11])
+
+ with self.subTest('SELECT with ORDER BY - By id, name, description - DESC'):
+ # Run test query.
+ results = self.connector.records.select(table_name, order_by_clause='id DESC, name DESC, description DESC')
+
+ # Verify records returned in expected order.
+ self.assertEqual(len(results), 12)
+ self.assertEqual(row_12, results[0])
+ self.assertEqual(row_11, results[1])
+ self.assertEqual(row_10, results[2])
+ self.assertEqual(row_9, results[3])
+ self.assertEqual(row_8, results[4])
+ self.assertEqual(row_7, results[5])
+ self.assertEqual(row_6, results[6])
+ self.assertEqual(row_5, results[7])
+ self.assertEqual(row_4, results[8])
+ self.assertEqual(row_3, results[9])
+ self.assertEqual(row_2, results[10])
+ self.assertEqual(row_1, results[11])
+
+ with self.subTest('SELECT with ORDER BY - By name, description, id - ASC'):
+ # Run test query.
+ results = self.connector.records.select(table_name, order_by_clause='name ASC, description ASC, id ASC')
+
+ # Verify records returned in expected order.
+ self.assertEqual(len(results), 12)
+ self.assertEqual(row_3, results[0])
+ self.assertEqual(row_9, results[1])
+ self.assertEqual(row_5, results[2])
+ self.assertEqual(row_11, results[3])
+ self.assertEqual(row_1, results[4])
+ self.assertEqual(row_6, results[5])
+ self.assertEqual(row_7, results[6])
+ self.assertEqual(row_12, results[7])
+ self.assertEqual(row_2, results[8])
+ self.assertEqual(row_8, results[9])
+ self.assertEqual(row_4, results[10])
+ self.assertEqual(row_10, results[11])
+
+ with self.subTest('SELECT with ORDER BY - By name, description, id - DESC'):
+ # Run test query.
+ results = self.connector.records.select(table_name, order_by_clause='name DESC, description DESC, id DESC')
+
+ # Verify records returned in expected order.
+ self.assertEqual(len(results), 12)
+ self.assertEqual(row_10, results[0])
+ self.assertEqual(row_4, results[1])
+ self.assertEqual(row_8, results[2])
+ self.assertEqual(row_2, results[3])
+ self.assertEqual(row_12, results[4])
+ self.assertEqual(row_7, results[5])
+ self.assertEqual(row_6, results[6])
+ self.assertEqual(row_1, results[7])
+ self.assertEqual(row_11, results[8])
+ self.assertEqual(row_5, results[9])
+ self.assertEqual(row_9, results[10])
+ self.assertEqual(row_3, results[11])
+
+ with self.subTest('SELECT with ORDER BY - By description, name, id - ASC'):
+ # Run test query.
+ results = self.connector.records.select(table_name, order_by_clause='description ASC, name ASC, id ASC')
+
+ # Verify records returned in expected order.
+ self.assertEqual(len(results), 12)
+ self.assertEqual(row_5, results[0])
+ self.assertEqual(row_11, results[1])
+ self.assertEqual(row_2, results[2])
+ self.assertEqual(row_8, results[3])
+ self.assertEqual(row_1, results[4])
+ self.assertEqual(row_6, results[5])
+ self.assertEqual(row_7, results[6])
+ self.assertEqual(row_12, results[7])
+ self.assertEqual(row_4, results[8])
+ self.assertEqual(row_10, results[9])
+ self.assertEqual(row_3, results[10])
+ self.assertEqual(row_9, results[11])
+
+ with self.subTest('SELECT with ORDER BY - By description, name, id - DESC'):
+ # Run test query.
+ results = self.connector.records.select(table_name, order_by_clause='description DESC, name DESC, id DESC')
+
+ # Verify records returned in expected order.
+ self.assertEqual(len(results), 12)
+ self.assertEqual(row_9, results[0])
+ self.assertEqual(row_3, results[1])
+ self.assertEqual(row_10, results[2])
+ self.assertEqual(row_4, results[3])
+ self.assertEqual(row_12, results[4])
+ self.assertEqual(row_7, results[5])
+ self.assertEqual(row_6, results[6])
+ self.assertEqual(row_1, results[7])
+ self.assertEqual(row_8, results[8])
+ self.assertEqual(row_2, results[9])
+ self.assertEqual(row_11, results[10])
+ self.assertEqual(row_5, results[11])
+
def test__insert__basic__success(self):
"""
Test `INSERT` query with basic values.