[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
This commit is contained in:
Bryan Strassner 2018-08-07 19:52:52 -05:00
parent 0341954f00
commit 20c27eed66
2 changed files with 17 additions and 2 deletions

View File

@ -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())

View File

@ -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)