/source filenames with spaces must be quoted

changelog.md

@@ -12,6 +12,7 @@ Features
 Bug Fixes
 ---------
 * Show CLI error on invalid `--execute` containing `/source`.
+* Require `/source` filenames containing spaces to be quoted.
 
 
 2.15.0 (2026/08/20)

mycli/client_commands.py

@@ -4,11 +4,13 @@
 import logging
 import os
 import re
+import shlex
 from typing import TYPE_CHECKING, Any, cast
 
 import click
 import sqlparse
 
+from mycli.compat import WIN
 from mycli.config import write_default_config
 from mycli.main_modes.repl import set_all_external_titles
 from mycli.packages import special
@@ -33,6 +35,7 @@
 DSN_CONFIG_VALUE = object()
 FAVORITES_CONFIG_VALUE = object()
 HIDDEN_CONFIG_SECTIONS = frozenset({'alias_dsn', 'favorite_queries'})
+INVALID_SOURCE_FILENAME = 'Source accepts exactly one filename; filenames containing spaces must be quoted.'
 SOURCE_SAFE_SPECIAL_COMMANDS = frozenset({
     'connect',
     'fd',
@@ -81,6 +84,45 @@ def _parse_source_arguments(arg: str) -> tuple[str, bool, bool, bool]:
     return filename, allow_special, show_queries, page_output
 
 
+def _has_unquoted_whitespace(value: str) -> bool:
+    quote: str | None = None
+    escaped = False
+    for character in value:
+        if escaped:
+            if quote is None and character.isspace():
+                return True
+            escaped = False
+            continue
+        if not WIN and character == '\\' and quote != "'":
+            escaped = True
+            continue
+        if character in ("'", '"'):
+            if quote is None:
+                quote = character
+            elif quote == character:
+                quote = None
+        elif quote is None and character.isspace():
+            return True
+    return False
+
+
+def _parse_source_filename(filename: str) -> str:
+    if not filename:
+        return ''
+    if _has_unquoted_whitespace(filename):
+        raise ValueError(INVALID_SOURCE_FILENAME)
+    try:
+        arguments = shlex.split(filename, posix=not WIN)
+    except ValueError as error:
+        raise ValueError(f'Invalid source filename: {error}.') from None
+    if len(arguments) != 1:
+        raise ValueError(INVALID_SOURCE_FILENAME)
+    parsed_filename = arguments[0]
+    if WIN and len(parsed_filename) >= 2 and parsed_filename[0] == parsed_filename[-1] and parsed_filename[0] in ("'", '"'):
+        parsed_filename = parsed_filename[1:-1]
+    return parsed_filename
+
+
 def _registered_special_command(query: str) -> tuple[str, str] | None:
     command, _verbosity, arg = special.parse_special_command(query)
     registered = special_main.COMMANDS.get(command)
@@ -355,8 +397,13 @@ def execute_from_file(self, arg: str, **_) -> Generator[SQLResult, None, None]:
         filename, allow_special, show_queries, page_output = _parse_source_arguments(arg)
         if page_output:
             yield SQLResult(command={'name': 'source_page'})
+        try:
+            filename = _parse_source_filename(filename)
+        except ValueError as error:
+            yield SQLResult(status=str(error), is_error=True)
+            return
         if not filename:
-            yield SQLResult(status="Missing required argument: filename.")
+            yield SQLResult(status="Missing required argument: filename.", is_error=True)
             return
 
         try:

mycli/packages/completion_engine.py

@@ -703,6 +703,10 @@ def suggest_type(full_text: str, text_before_cursor: str) -> list[dict[str, Any]
     A scope for a column category will be a list of tables.
     """
 
+    stripped_text = text_before_cursor.lstrip()
+    if re.match(r'^(?:source|/source|\\\.|/\.)\s', stripped_text, re.IGNORECASE):
+        return suggest_special(text_before_cursor)
+
     word_before_cursor = last_word(text_before_cursor, include="many_punctuations")
 
     identifier: Identifier | None = None
@@ -817,7 +821,7 @@ def suggest_special(text: str) -> list[dict[str, Any]]:
         if not source_arguments:
             return [
                 {'type': 'special_subcommand', 'subcommands': source_options},
-                {'type': 'file_name'},
+                {'type': 'file_name', 'quote_spaces': True, 'source_filename': ''},
             ]
 
         used_options: set[str] = set()
@@ -826,17 +830,22 @@ def suggest_special(text: str) -> list[dict[str, Any]]:
             used_options.add(source_arguments[argument_index])
             argument_index += 1
         remaining_options = [option for option in source_options if option not in used_options]
+        source_filename = _arg
+        for _index in range(argument_index):
+            parsed_argument = source_filename.split(maxsplit=1)
+            source_filename = parsed_argument[1] if len(parsed_argument) == 2 else ''
+        file_suggestion = {'type': 'file_name', 'quote_spaces': True, 'source_filename': source_filename}
 
         if argument_index < len(source_arguments):
             if source_arguments[argument_index].startswith('-'):
                 return [{'type': 'special_subcommand', 'subcommands': remaining_options}]
-            return [{'type': 'file_name'}]
+            return [file_suggestion]
         if not text[-1].isspace():
             return []
-        suggestions = []
+        suggestions: list[dict[str, Any]] = []
         if remaining_options:
             suggestions.append({'type': 'special_subcommand', 'subcommands': remaining_options})
-        suggestions.append({'type': 'file_name'})
+        suggestions.append(file_suggestion)
         return suggestions
 
     if cmd.lower() in [

mycli/packages/filepaths.py

@@ -85,7 +85,7 @@ def suggest_path(root_dir: str) -> list[str]:
             *list_path(os.curdir),
         ]
 
-    if root_dir[0] not in ('/', '~') and root_dir[0:1] != './':
+    if root_dir[0] not in ('/', '~') and root_dir[0:2] != './':
         return list_path(os.curdir)
 
     if "~" in root_dir:

mycli/sqlcompleter.py

@@ -3,7 +3,10 @@
 from collections import Counter
 from enum import IntEnum
 import logging
+import os
 import re
+import shlex
+import subprocess
 from typing import Any, Collection, Generator, Iterable, Literal
 
 from jinja2 import TemplateError
@@ -12,6 +15,7 @@
 from pygments.lexers._mysql_builtins import MYSQL_DATATYPES, MYSQL_FUNCTIONS, MYSQL_KEYWORDS
 import rapidfuzz
 
+from mycli.compat import WIN
 from mycli.packages.completion_engine import is_inside_quotes, suggest_type
 from mycli.packages.filepaths import complete_path, parse_path, suggest_path
 from mycli.packages.special import llm
@@ -1454,6 +1458,7 @@ def get_completions(
         suggestions = suggest_type(document.text, document.text_before_cursor)
         rigid_sort = False
         length_based_on_path = False
+        source_file_completion_length: int | None = None
         config_property_length: int | None = None
         completion_filter_text = text_for_len
 
@@ -1719,7 +1724,26 @@ def get_completions(
                 completions.extend([(*x, rank) for x in formats_m])
 
             elif suggestion["type"] == "file_name":
-                file_names_m = self.find_files(word_before_cursor)
+                source_filename = suggestion.get('source_filename')
+                if source_filename is None:
+                    file_names_m = self.find_files(word_before_cursor)
+                else:
+                    source_file_completion_length = len(source_filename)
+                    quote = source_filename[0] if source_filename[:1] in ("'", '"') else None
+                    partial_path = source_filename[1:] if quote else source_filename
+                    if quote and partial_path.endswith(quote):
+                        partial_path = partial_path[:-1]
+                    base_path, _last_path, _position = parse_path(partial_path)
+                    file_names_m = (
+                        (
+                            self._quote_source_path(
+                                os.path.join(base_path, path) if base_path and not path.startswith('~') else path,
+                                quote,
+                            ),
+                            fuzziness,
+                        )
+                        for path, fuzziness in self.find_files(partial_path)
+                    )
                 completions.extend([(*x, rank) for x in file_names_m])
                 # for filenames we _really_ want directories to go last
                 rigid_sort = True
@@ -1805,6 +1829,8 @@ def completion_sort_key(item: tuple[str, int, int], text_for_len: str):
 
         if config_property_length is not None:
             return (Completion(x, -config_property_length) for x in uniq_completions_str)
+        elif source_file_completion_length is not None:
+            return (Completion(x, -source_file_completion_length) for x in uniq_completions_str)
         elif length_based_on_path:
             return (
                 Completion(
@@ -1842,6 +1868,19 @@ def find_files(self, word: str) -> Generator[tuple[str, int], None, None]:
             if suggestion:
                 yield (suggestion, Fuzziness.PERFECT)
 
+    @staticmethod
+    def _quote_source_path(path: str, quote: str | None) -> str:
+        is_directory = path.endswith(('/', os.sep))
+        if is_directory and any(character.isspace() for character in path) and not path.startswith(('/', '~', './')):
+            path = f'./{path}'
+        if quote:
+            return f'{quote}{path}' if is_directory else f'{quote}{path}{quote}'
+        if not any(character.isspace() for character in path):
+            return path
+        if is_directory:
+            return f'"{path}' if WIN else f"'{path}"
+        return subprocess.list2cmdline([path]) if WIN else shlex.quote(path)
+
     def populate_scoped_cols(self, scoped_tbls: list[tuple[str | None, str, str | None]]) -> list[str]:
         """Find all columns in a set of scoped_tables
         :param scoped_tbls: list of (schema, table, alias) tuples

test/pytests/test_client_commands.py

@@ -113,6 +113,55 @@ def test_parse_source_arguments(arg: str, expected: tuple[str, bool, bool, bool]
     assert client_commands._parse_source_arguments(arg) == expected
 
 
+@pytest.mark.parametrize(
+    ('filename', 'expected'),
+    [
+        ('query.sql', 'query.sql'),
+        ('"query file.sql"', 'query file.sql'),
+        ("'query file.sql'", 'query file.sql'),
+        ('prefix" query".sql', 'prefix query.sql'),
+    ],
+)
+def test_parse_source_filename(filename: str, expected: str) -> None:
+    assert client_commands._parse_source_filename(filename) == expected
+
+
+@pytest.mark.parametrize(
+    'filename',
+    [
+        'query file.sql',
+        r'query\ file.sql',
+        '"first file.sql" second.sql',
+    ],
+)
+def test_parse_source_filename_rejects_multiple_unquoted_arguments(filename: str) -> None:
+    with pytest.raises(ValueError, match='filenames containing spaces must be quoted'):
+        client_commands._parse_source_filename(filename)
+
+
+def test_parse_source_filename_rejects_unclosed_quote() -> None:
+    with pytest.raises(ValueError, match='No closing quotation'):
+        client_commands._parse_source_filename('"query file.sql')
+
+
+def test_parse_source_filename_rejects_missing_parsed_argument(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(client_commands.shlex, 'split', lambda *_args, **_kwargs: [])
+
+    with pytest.raises(ValueError, match='accepts exactly one filename'):
+        client_commands._parse_source_filename('query.sql')
+
+
+def test_source_filename_whitespace_scanner_allows_escaped_non_whitespace() -> None:
+    assert not client_commands._has_unquoted_whitespace(r'query\name.sql')
+
+
+def test_parse_source_filename_preserves_windows_backslashes(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(client_commands, 'WIN', True)
+
+    assert client_commands._parse_source_filename(r'C:\queries\query.sql') == r'C:\queries\query.sql'
+    assert client_commands._parse_source_filename(r'"C:\my queries\query.sql"') == r'C:\my queries\query.sql'
+
+
 def test_register_special_commands_registers_expected_commands(monkeypatch: pytest.MonkeyPatch) -> None:
     client = DummyClient()
     calls: list[tuple[Any, ...]] = []
@@ -502,7 +551,7 @@ def test_change_db_without_argument_reports_error(monkeypatch: pytest.MonkeyPatc
 def test_execute_from_file_requires_filename() -> None:
     client = DummyClient()
 
-    assert list(client.execute_from_file('')) == [SQLResult(status='Missing required argument: filename.')]
+    assert list(client.execute_from_file('')) == [SQLResult(status='Missing required argument: filename.', is_error=True)]
 
 
 def test_execute_from_file_reports_open_errors() -> None:
@@ -706,21 +755,46 @@ def open_file(path: str) -> IteratedFile:
         return file_h
 
     monkeypatch.setattr(client_commands, 'open', open_file, raising=False)
+    monkeypatch.setattr(client_commands.os.path, 'expanduser', lambda path: f'/expanded/{path.removeprefix("~/")}')
 
-    assert result_statuses(client.execute_from_file('--special query file.sql')) == ['ran select 1;']
-    assert opened_paths == ['query file.sql']
+    assert result_statuses(client.execute_from_file('--special "~/query file.sql"')) == ['ran select 1;']
+    assert opened_paths == ['/expanded/query file.sql']
 
 
 @pytest.mark.parametrize('options', ['--special', '--show', '--page', '--special --show --page'])
 def test_execute_from_file_reports_missing_filename_after_options(options: str) -> None:
     client = DummyClient()
 
-    expected = [SQLResult(status='Missing required argument: filename.')]
+    expected = [SQLResult(status='Missing required argument: filename.', is_error=True)]
     if '--page' in options:
         expected.insert(0, SQLResult(command={'name': 'source_page'}))
     assert list(client.execute_from_file(options)) == expected
 
 
+def test_execute_from_file_rejects_unquoted_filename_with_spaces(monkeypatch: pytest.MonkeyPatch) -> None:
+    client = DummyClient()
+    opened_paths: list[str] = []
+    monkeypatch.setattr(client_commands, 'open', lambda path: opened_paths.append(path), raising=False)
+
+    assert list(client.execute_from_file('query file.sql')) == [SQLResult(status=client_commands.INVALID_SOURCE_FILENAME, is_error=True)]
+    assert opened_paths == []
+
+
+def test_execute_from_file_pages_invalid_filename_error() -> None:
+    client = DummyClient()
+
+    assert list(client.execute_from_file('--page query file.sql')) == [
+        SQLResult(command={'name': 'source_page'}),
+        SQLResult(status=client_commands.INVALID_SOURCE_FILENAME, is_error=True),
+    ]
+
+
+def test_execute_from_file_treats_empty_quotes_as_missing_filename() -> None:
+    client = DummyClient()
+
+    assert list(client.execute_from_file('""')) == [SQLResult(status='Missing required argument: filename.', is_error=True)]
+
+
 def test_execute_from_file_runs_permitted_special_commands(capsys: pytest.CaptureFixture[str], tmp_path: Path) -> None:
     client = DummyClient()
     sql_file = tmp_path / 'query.sql'