Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
# Split the sql into separate queries and run each one.
# Unless it's saving a favorite query, in which case we
# want to save them all together.
if statement.startswith("\\fs"):
components = [statement]
else:
components = sqlparse.split(statement)
for sql in components:
# Remove spaces, eol and semi-colons.
sql = sql.rstrip(";")
# \G is treated specially since we have to set the expanded output.
if sql.endswith("\\G"):
special.set_expanded_output(True)
sql = sql[:-2].strip()
if not self.conn and not (
sql.startswith(".open")
or sql.lower().startswith("use")
or sql.startswith("\\u")
or sql.startswith("\\?")
or sql.startswith("\\q")
or sql.startswith("help")
or sql.startswith("exit")
or sql.startswith("quit")
):
_logger.debug(
"Not connected to database. Will not run statement: %s.", sql
)
raise OperationalError("Not connected to database.")
def one_iteration(text=None):
if text is None:
try:
text = self.prompt_app.prompt()
except KeyboardInterrupt:
return
special.set_expanded_output(False)
try:
text = self.handle_editor_command(text)
except RuntimeError as e:
logger.error("sql: %r, error: %r", text, e)
logger.error("traceback: %r", traceback.format_exc())
self.echo(str(e), err=True, fg="red")
return
if not text.strip():
return
if self.destructive_warning:
destroy = confirm_destructive_query(text)
if destroy is None:
pass # Query was not destructive. Nothing to do here.