Apply black

This is necessary to pass the CI lint tests on all PRs.

litecli/completion_refresher.py

@@ -7,7 +7,6 @@
 
 
 class CompletionRefresher(object):
-
     refreshers = OrderedDict()
 
     def __init__(self):

litecli/main.py

@@ -59,7 +59,6 @@
 
 
 class LiteCli(object):
-
     default_prompt = "\\d> "
     max_len_prompt = 45
 
@@ -113,7 +112,9 @@ def __init__(
         # Load startup commands.
         try:
             self.startup_commands = c["startup_commands"]
-        except KeyError: # Redundant given the load_config() function that merges in the standard config, but put here to avoid fail if user do not have updated config file.
+        except (
+            KeyError
+        ):  # Redundant given the load_config() function that merges in the standard config, but put here to avoid fail if user do not have updated config file.
             self.startup_commands = None
 
         self.completion_refresher = CompletionRefresher()
@@ -235,7 +236,6 @@ def change_prompt_format(self, arg, **_):
         return [(None, None, None, "Changed prompt format to %s" % arg)]
 
     def initialize_logging(self):
-
         log_file = self.config["main"]["log_file"]
         if log_file == "default":
             log_file = config_location() + "log"
@@ -303,7 +303,6 @@ def get(key):
         return {x: get(x) for x in keys}
 
     def connect(self, database=""):
-
         cnf = {"database": None}
 
         cnf = self.read_my_cnf_files(cnf.keys())
@@ -560,7 +559,6 @@ def one_iteration(text=None):
             complete_style = CompleteStyle.READLINE_LIKE
 
         with self._completer_lock:
-
             if self.key_bindings == "vi":
                 editing_mode = EditingMode.VI
             else:
@@ -594,10 +592,11 @@ def one_iteration(text=None):
                 editing_mode=editing_mode,
                 search_ignore_case=True,
             )
+
         def startup_commands():
             if self.startup_commands:
                 if "commands" in self.startup_commands:
-                    for command in self.startup_commands['commands']:
+                    for command in self.startup_commands["commands"]:
                         try:
                             res = sqlexecute.run(command)
                         except Exception as e:
@@ -606,19 +605,29 @@ def startup_commands():
                         else:
                             click.echo(command)
                             for title, cur, headers, status in res:
-                                if title == 'dot command not implemented':
-                                    self.echo("The SQLite dot command '" + command.split(' ', 1)[0]+"' is not yet implemented.", fg="yellow")
+                                if title == "dot command not implemented":
+                                    self.echo(
+                                        "The SQLite dot command '"
+                                        + command.split(" ", 1)[0]
+                                        + "' is not yet implemented.",
+                                        fg="yellow",
+                                    )
                                 else:
                                     output = self.format_output(title, cur, headers)
                                     for line in output:
                                         self.echo(line)
                 else:
-                    self.echo("Could not read commands. The startup commands needs to be formatted as: \n commands = 'command1', 'command2', ...", fg="yellow")
+                    self.echo(
+                        "Could not read commands. The startup commands needs to be formatted as: \n commands = 'command1', 'command2', ...",
+                        fg="yellow",
+                    )
 
         try:
             startup_commands()
         except Exception as e:
-            self.echo("Could not execute all startup commands: \n"+str(e), fg="yellow")
+            self.echo(
+                "Could not execute all startup commands: \n" + str(e), fg="yellow"
+            )
 
         try:
             while True:

litecli/packages/special/favoritequeries.py

@@ -3,7 +3,6 @@
 
 
 class FavoriteQueries(object):
-
     section_name = "favorite_queries"
 
     usage = """

litecli/packages/special/iocommands.py

@@ -461,7 +461,7 @@ def watch_query(arg, **kwargs):
             # Somewhere in the code the pager its activated after every yield,
             # so we disable it in every iteration
             set_pager_enabled(False)
-            for (sql, title) in sql_list:
+            for sql, title in sql_list:
                 cur.execute(sql)
                 if cur.description:
                     headers = [x[0] for x in cur.description]

litecli/packages/special/utils.py

@@ -55,10 +55,77 @@ def check_if_sqlitedotcommand(command):
     :returns: True/False
     """
 
-    sqlite3dotcommands = ['.archive','.auth','.backup','.bail','.binary','.cd','.changes','.check','.clone','.connection','.databases','.dbconfig','.dbinfo','.dump','.echo','.eqp','.excel','.exit','.expert','.explain','.filectrl','.fullschema','.headers','.help','.import','.imposter','.indexes','.limit','.lint','.load','.log','.mode','.nonce','.nullvalue','.once','.open','.output','.parameter','.print','.progress','.prompt','.quit','.read','.recover','.restore','.save','.scanstats','.schema','.selftest','.separator','.session','.sha3sum','.shell','.show','.stats','.system','.tables','.testcase','.testctrl','.timeout','.timer','.trace','.vfsinfo','.vfslist','.vfsname','.width']
+    sqlite3dotcommands = [
+        ".archive",
+        ".auth",
+        ".backup",
+        ".bail",
+        ".binary",
+        ".cd",
+        ".changes",
+        ".check",
+        ".clone",
+        ".connection",
+        ".databases",
+        ".dbconfig",
+        ".dbinfo",
+        ".dump",
+        ".echo",
+        ".eqp",
+        ".excel",
+        ".exit",
+        ".expert",
+        ".explain",
+        ".filectrl",
+        ".fullschema",
+        ".headers",
+        ".help",
+        ".import",
+        ".imposter",
+        ".indexes",
+        ".limit",
+        ".lint",
+        ".load",
+        ".log",
+        ".mode",
+        ".nonce",
+        ".nullvalue",
+        ".once",
+        ".open",
+        ".output",
+        ".parameter",
+        ".print",
+        ".progress",
+        ".prompt",
+        ".quit",
+        ".read",
+        ".recover",
+        ".restore",
+        ".save",
+        ".scanstats",
+        ".schema",
+        ".selftest",
+        ".separator",
+        ".session",
+        ".sha3sum",
+        ".shell",
+        ".show",
+        ".stats",
+        ".system",
+        ".tables",
+        ".testcase",
+        ".testctrl",
+        ".timeout",
+        ".timer",
+        ".trace",
+        ".vfsinfo",
+        ".vfslist",
+        ".vfsname",
+        ".width",
+    ]
 
     if isinstance(command, str):
-        command = command.split(' ', 1)[0].lower()
-        return (command in sqlite3dotcommands)
+        command = command.split(" ", 1)[0].lower()
+        return command in sqlite3dotcommands
     else:
-        return False
+        return False

litecli/sqlcompleter.py

@@ -448,7 +448,6 @@ def get_completions(self, document, complete_event):
         suggestions = suggest_type(document.text, document.text_before_cursor)
 
         for suggestion in suggestions:
-
             _logger.debug("Suggestion type: %r", suggestion["type"])
 
             if suggestion["type"] == "column":

litecli/sqlexecute.py

@@ -9,15 +9,16 @@
 import os.path
 
 from .packages import special
+
 _logger = logging.getLogger(__name__)
 
 # FIELD_TYPES = decoders.copy()
 # FIELD_TYPES.update({
 #     FIELD_TYPE.NULL: type(None)
 # })
 
-class SQLExecute(object):
 
+class SQLExecute(object):
     databases_query = """
         PRAGMA database_list
     """
@@ -78,7 +79,6 @@ def connect(self, database=None):
         # retrieve connection id
         self.reset_connection_id()
 
-
     def run(self, statement):
         """Execute the sql in the database and return the results. The results
         are a list of tuples. Each tuple has 4 values
@@ -130,7 +130,7 @@ def run(self, statement):
                     yield result
             except special.CommandNotFound:  # Regular SQL
                 if check_if_sqlitedotcommand(sql):
-                    yield ('dot command not implemented', None, None, None)
+                    yield ("dot command not implemented", None, None, None)
                 else:
                     _logger.debug("Regular sql statement. sql: %r", sql)
                     cur.execute(sql)

tasks.py

@@ -75,7 +75,6 @@ def run(self):
 
 
 class test(TestCommand):
-
     user_options = [("pytest-args=", "a", "Arguments to pass to pytest")]
 
     def initialize_options(self):

tests/test_dbspecial.py

@@ -84,6 +84,7 @@ def test_check_if_sqlitedotcommand():
         ["binary", False],
         [234, False],
         [".changes   test! test", True],
-        ["NotDotcommand", False]]
+        ["NotDotcommand", False],
+    ]
     for command, expected_result in test_cases:
-        assert check_if_sqlitedotcommand(command) == expected_result
+        assert check_if_sqlitedotcommand(command) == expected_result

tests/test_main.py

@@ -264,6 +264,9 @@ def test_import_command(executor):
 
 def test_startup_commands(executor):
     m = LiteCli(liteclirc=default_config_file)
-    assert m.startup_commands['commands'] == ['create table startupcommands(a text)', "insert into startupcommands values('abc')"]
+    assert m.startup_commands["commands"] == [
+        "create table startupcommands(a text)",
+        "insert into startupcommands values('abc')",
+    ]
 
-    # implement tests on executions of the startupcommands
+    # implement tests on executions of the startupcommands

tests/test_smart_completion_public_schema_only.py

@@ -15,7 +15,6 @@
 
 @pytest.fixture
 def completer():
-
     import litecli.sqlcompleter as sqlcompleter
 
     comp = sqlcompleter.SQLCompleter()
@@ -367,17 +366,15 @@ def test_auto_escaped_col_names(completer, complete_event):
             Document(text=text, cursor_position=position), complete_event
         )
     )
-    assert (
-        result
-        == [
-            Completion(text="*", start_position=0),
-            Completion(text="`ABC`", start_position=0),
-            Completion(text="`insert`", start_position=0),
-            Completion(text="id", start_position=0),
-        ]
-        + list(map(Completion, completer.functions))
-        + [Completion(text="select", start_position=0)]
-        + list(map(Completion, sorted(completer.keywords)))
+    assert result == [
+        Completion(text="*", start_position=0),
+        Completion(text="`ABC`", start_position=0),
+        Completion(text="`insert`", start_position=0),
+        Completion(text="id", start_position=0),
+    ] + list(map(Completion, completer.functions)) + [
+        Completion(text="select", start_position=0)
+    ] + list(
+        map(Completion, sorted(completer.keywords))
     )
 
 

tests/test_sqlexecute.py

@@ -94,21 +94,21 @@ def test_invalid_column_name(executor):
 @dbtest
 def test_unicode_support_in_output(executor):
     run(executor, "create table unicodechars(t text)")
-    run(executor, u"insert into unicodechars (t) values ('é')")
+    run(executor, "insert into unicodechars (t) values ('é')")
 
     # See issue #24, this raises an exception without proper handling
-    results = run(executor, u"select * from unicodechars")
-    assert_result_equal(results, headers=["t"], rows=[(u"é",)])
+    results = run(executor, "select * from unicodechars")
+    assert_result_equal(results, headers=["t"], rows=[("é",)])
 
 
 @dbtest
 def test_invalid_unicode_values_dont_choke(executor):
     run(executor, "create table unicodechars(t text)")
     # \xc3 is not a valid utf-8 char. But we can insert it into the database
     # which can break querying if not handled correctly.
-    run(executor, u"insert into unicodechars (t) values (cast(x'c3' as text))")
+    run(executor, "insert into unicodechars (t) values (cast(x'c3' as text))")
 
-    results = run(executor, u"select * from unicodechars")
+    results = run(executor, "select * from unicodechars")
     assert_result_equal(results, headers=["t"], rows=[("\\xc3",)])
 
 
@@ -120,13 +120,13 @@ def test_multiple_queries_same_line(executor):
         {
             "title": None,
             "headers": ["'foo'"],
-            "rows": [(u"foo",)],
+            "rows": [("foo",)],
             "status": "1 row in set",
         },
         {
             "title": None,
             "headers": ["'bar'"],
-            "rows": [(u"bar",)],
+            "rows": [("bar",)],
             "status": "1 row in set",
         },
     ]
@@ -369,8 +369,8 @@ def test_cd_command_current_dir(executor):
 
 @dbtest
 def test_unicode_support(executor):
-    results = run(executor, u"SELECT '日本語' AS japanese;")
-    assert_result_equal(results, headers=["japanese"], rows=[(u"日本語",)])
+    results = run(executor, "SELECT '日本語' AS japanese;")
+    assert_result_equal(results, headers=["japanese"], rows=[("日本語",)])
 
 
 @dbtest