From 20c27eed6669b72e13a97f6551fb0eee045be8ae Mon Sep 17 00:00:00 2001 From: Bryan Strassner Date: Tue, 7 Aug 2018 19:52:52 -0500 Subject: [PATCH] [FIX] Single line sql split/join fix Because the type of the query was generated by sqlalchemy.sql.text(), the TextClause object did not have a split() method, and crashed out trying to reformat into one line. Using the str() wrapper provides a string that can be properly split and joined. Change-Id: I1ed9e39d7ebf3904d3d233330ee57082ad02c5f3 --- .../shipyard_airflow/db/common_db.py | 6 ++++-- .../tests/unit/db/test_common_db.py | 13 +++++++++++++ 2 files changed, 17 insertions(+), 2 deletions(-) diff --git a/src/bin/shipyard_airflow/shipyard_airflow/db/common_db.py b/src/bin/shipyard_airflow/shipyard_airflow/db/common_db.py index 3e77f4e6..ab514fb0 100644 --- a/src/bin/shipyard_airflow/shipyard_airflow/db/common_db.py +++ b/src/bin/shipyard_airflow/shipyard_airflow/db/common_db.py @@ -151,6 +151,8 @@ class DbAccess: def _query_single_line(query): """Reformats a query string to remove newlines and extra spaces - :param query: The query string to log + :param query: The query to log. This will work for anything that will + result in a string after str() is applied to it. Be aware of this + conversion. E.g. sqlalchemy's TextClause objects. """ - return " ".join(query.split()) + return " ".join(str(query).split()) diff --git a/src/bin/shipyard_airflow/tests/unit/db/test_common_db.py b/src/bin/shipyard_airflow/tests/unit/db/test_common_db.py index bae4f604..05d4ab64 100644 --- a/src/bin/shipyard_airflow/tests/unit/db/test_common_db.py +++ b/src/bin/shipyard_airflow/tests/unit/db/test_common_db.py @@ -11,6 +11,8 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. +import sqlalchemy + from shipyard_airflow.db import common_db @@ -45,9 +47,20 @@ M2_RES = 'SELECT things FROM tables' S_QUERY = "SELECT 1 FROM dual" +SQL_ALCHEMY_QUERY = sqlalchemy.sql.text(""" +SELECT + fields +FROM + tables +WHERE 1=1 +""") + +SQL_A_RES = 'SELECT fields FROM tables WHERE 1=1' + class TestCommonDb(): def test_single_line_query(self): assert M_RES == common_db._query_single_line(M_QUERY) assert M2_RES == common_db._query_single_line(M2_QUERY) assert S_QUERY == common_db._query_single_line(S_QUERY) + assert SQL_A_RES == common_db._query_single_line(SQL_ALCHEMY_QUERY)