[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:
parent
0341954f00
commit
20c27eed66
|
@ -151,6 +151,8 @@ class DbAccess:
|
||||||
def _query_single_line(query):
|
def _query_single_line(query):
|
||||||
"""Reformats a query string to remove newlines and extra spaces
|
"""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())
|
||||||
|
|
|
@ -11,6 +11,8 @@
|
||||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||||
# See the License for the specific language governing permissions and
|
# See the License for the specific language governing permissions and
|
||||||
# limitations under the License.
|
# limitations under the License.
|
||||||
|
import sqlalchemy
|
||||||
|
|
||||||
from shipyard_airflow.db import common_db
|
from shipyard_airflow.db import common_db
|
||||||
|
|
||||||
|
|
||||||
|
@ -45,9 +47,20 @@ M2_RES = 'SELECT things FROM tables'
|
||||||
|
|
||||||
S_QUERY = "SELECT 1 FROM dual"
|
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():
|
class TestCommonDb():
|
||||||
def test_single_line_query(self):
|
def test_single_line_query(self):
|
||||||
assert M_RES == common_db._query_single_line(M_QUERY)
|
assert M_RES == common_db._query_single_line(M_QUERY)
|
||||||
assert M2_RES == common_db._query_single_line(M2_QUERY)
|
assert M2_RES == common_db._query_single_line(M2_QUERY)
|
||||||
assert S_QUERY == common_db._query_single_line(S_QUERY)
|
assert S_QUERY == common_db._query_single_line(S_QUERY)
|
||||||
|
assert SQL_A_RES == common_db._query_single_line(SQL_ALCHEMY_QUERY)
|
||||||
|
|
Loading…
Reference in New Issue