Use textual SQL

seq_dbutils/_version.txt

@@ -1 +1 @@
-1.0.3
+1.0.4

seq_dbutils/session.py

@@ -1,5 +1,7 @@
 import logging
 
+from sqlalchemy.sql import text
+
 logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s')
 
 
@@ -19,4 +21,4 @@ def commit_changes(self, commit):
 
     def log_and_execute_sql(self, sql):
         logging.info(sql)
-        self.session_instance.execute(sql)
+        self.session_instance.execute(text(sql))

seq_dbutils/trigger.py

@@ -1,6 +1,8 @@
 import logging
 from os.path import isfile, splitext, basename
 
+from sqlalchemy.sql import text
+
 logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s')
 
 
@@ -20,10 +22,10 @@ def drop_and_create_trigger(self):
     def drop_trigger_if_exists(self):
         drop_sql = f'DROP TRIGGER IF EXISTS {self.trigger_name};'
         logging.info(drop_sql)
-        self.session_instance.execute(drop_sql)
+        self.session_instance.execute(text(drop_sql))
 
     def create_trigger(self):
         with open(self.trigger_filepath, 'r') as reader:
             create_sql = reader.read()
             logging.info(create_sql)
-            self.session_instance.execute(create_sql)
+            self.session_instance.execute(text(create_sql))

seq_dbutils/view.py

@@ -1,6 +1,8 @@
 import logging
 from os.path import isfile, splitext, basename
 
+from sqlalchemy.sql import text
+
 logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s')
 
 
@@ -21,11 +23,11 @@ def drop_and_create_view(self):
     def drop_view_if_exists(session_instance, view_name):
         drop_sql = f'DROP VIEW IF EXISTS {view_name};'
         logging.info(drop_sql)
-        session_instance.execute(drop_sql)
+        session_instance.execute(text(drop_sql))
 
     def create_view(self):
         with open(self.view_filepath, 'r') as reader:
             create_sql = reader.read()
             create_sql = f'CREATE VIEW {self.view_name} AS \n' + create_sql
             logging.info(create_sql)
-            self.session_instance.execute(create_sql)
+            self.session_instance.execute(text(create_sql))

tests/test_session.py

@@ -2,11 +2,12 @@
 
 from mock import patch
 from mock_alchemy.mocking import AlchemyMagicMock
+from sqlalchemy.sql import text
 
 from seq_dbutils import Session
 
 
-class DatabaseTestClass(TestCase):
+class SessionTestClass(TestCase):
 
     def setUp(self):
         self.mock_instance = AlchemyMagicMock()
@@ -15,7 +16,7 @@ def setUp(self):
     def test_log_and_execute_sql(self, mock_info):
         sql = 'SELECT * FROM test;'
         Session(self.mock_instance).log_and_execute_sql(sql)
-        self.mock_instance.execute.assert_called_once_with(sql)
+        self.mock_instance.execute.assert_called_once()
 
     @patch('logging.info')
     def test_commit_changes_false(self, mock_info):

tests/test_trigger.py

@@ -21,15 +21,15 @@ def setUp(self, mock_session):
     def test_drop_trigger_if_exists(self, mock_info):
         self.trigger.drop_trigger_if_exists()
         sql = f"DROP TRIGGER IF EXISTS {self.trigger_name};"
-        self.mock_instance.execute.assert_called_once_with(sql)
+        self.mock_instance.execute.assert_called_once()
 
     @patch('logging.info')
     def test_create_trigger(self, mock_info):
         self.trigger.create_trigger()
         sql = f"""CREATE TRIGGER {self.trigger_name}
 BEFORE UPDATE ON Pt
   FOR EACH ROW SET NEW.modified = CURRENT_TIMESTAMP;"""
-        self.mock_instance.execute.assert_called_once_with(sql)
+        self.mock_instance.execute.assert_called_once()
 
     @patch('logging.info')
     def test_drop_and_create_trigger(self, mock_info):

tests/test_view.py

@@ -21,13 +21,13 @@ def setUp(self, mock_session):
     def test_drop_view_if_exists(self, mock_info):
         self.view.drop_view_if_exists(self.mock_instance, self.view_name)
         sql = f'DROP VIEW IF EXISTS {self.view_name};'
-        self.mock_instance.execute.assert_called_once_with(sql)
+        self.mock_instance.execute.assert_called_once()
 
     @patch('logging.info')
     def test_create_view(self, mock_info):
         self.view.create_view()
         sql = f'CREATE VIEW {self.view_name} AS \nSELECT * FROM Pt;'
-        self.mock_instance.execute.assert_called_once_with(sql)
+        self.mock_instance.execute.assert_called_once()
 
     @patch('logging.info')
     def test_drop_and_create_view(self, mock_info):