Add a /ping special command

changelog.md

@@ -1,3 +1,11 @@
+Upcoming (TBD)
+==============
+
+Features
+--------
+* Add `/ping` special command.
+
+
 2.18.5 (2026/08/31)
 ==============
 

mycli/TIPS

@@ -118,6 +118,8 @@ use /dsn to manage saved DSNs!
 
 use /config to inspect persistent configuration from the REPL!
 
+/ping checks the health of the connection!
+
 ###
 ### environment variables
 ###

mycli/client_commands.py

@@ -44,6 +44,7 @@
     'help',
     'l',
     'nowarnings',
+    'ping',
     'prompt',
     'redirectformat',
     'rehash',

mycli/packages/special/__init__.py

@@ -10,6 +10,7 @@
 from mycli.packages.special.dbcommands import (
     list_databases,
     list_tables,
+    ping,
     status,
 )
 from mycli.packages.special.iocommands import (
@@ -118,6 +119,7 @@ def sql_using_llm(*args, **kwargs):  # type: ignore[no-redef, misc]
     'list_tables',
     'open_external_editor',
     'parse_special_command',
+    'ping',
     'register_special_command',
     'run_post_redirect_hook',
     'set_delimiter',

mycli/packages/special/dbcommands.py

@@ -2,7 +2,7 @@
 import os
 import platform
 
-from pymysql import ProgrammingError
+from pymysql import Error, ProgrammingError
 from pymysql.cursors import Cursor
 
 from mycli import __version__
@@ -80,6 +80,24 @@ def list_databases(cur: Cursor, **_) -> list[SQLResult]:
         return [SQLResult()]
 
 
+@special_command(
+    r'\ping',
+    '/ping',
+    'Check the connection.',
+    arg_type=ArgType.PARSED_QUERY,
+    completion_snippet='check connection',
+)
+def ping(cur: Cursor, arg: str | None = None, **_) -> list[SQLResult]:
+    if arg:
+        return [SQLResult(status='Syntax: /ping.')]
+
+    try:
+        cur.connection.ping(reconnect=False)
+    except Error:
+        return [SQLResult(status='Not connected')]
+    return [SQLResult(status='Connected')]
+
+
 @special_command(
     "status",
     "/status",

test/features/fixture_data/help_commands.txt

@@ -24,6 +24,7 @@
 | /nowarnings     | /w       | /nowarnings                              | Disable automatic warnings display.                         |
 | /once           | /o       | /once [-o] <file>                        | Append next result to an output file (overwrite using -o).  |
 | /pager          | /P       | /pager [command]                         | Set pager to [command]. Print query results via pager.      |
+| /ping           | <null>   | /ping                                    | Check the connection.                                       |
 | /pipe_once      | /|       | /pipe_once <command>                     | Send next result to a subprocess.                           |
 | /prompt         | /R       | /prompt [string]                         | Show or change prompt format.                               |
 | /quit           | /q       | /quit                                    | Quit.                                                       |

test/pytests/test_client_commands.py

@@ -891,6 +891,7 @@ def test_execute_from_file_requires_semicolon_for_special_commands(tmp_path: Pat
     ('command', 'arg', 'expected'),
     [
         ('status', '', True),
+        ('ping', '', True),
         ('connect', 'db', True),
         ('config', 'get main.prompt', True),
         ('config', 'edit', False),

test/pytests/test_special_dbcommands.py

@@ -2,11 +2,14 @@
 
 from unittest.mock import MagicMock
 
-from pymysql import ProgrammingError
+from pymysql import Error, ProgrammingError
+import pytest
 
 from mycli.packages.completion_engine import suggest_type
 from mycli.packages.special import dbcommands
-from mycli.packages.special.dbcommands import list_databases, list_tables, status
+from mycli.packages.special import main as special_main
+from mycli.packages.special.dbcommands import list_databases, list_tables, ping, status
+from mycli.packages.sqlresult import SQLResult
 from test.pytests.test_completion_engine import sorted_dicts
 
 
@@ -19,16 +22,24 @@ def __init__(
         host_info: str = 'Localhost via UNIX socket',
         unix_socket: str | None = None,
         thread_id_value: int = 42,
+        ping_error: Exception | None = None,
     ) -> None:
         self.host = host
         self.port = port
         self.host_info = host_info
         self.unix_socket = unix_socket
         self._thread_id_value = thread_id_value
+        self.ping_error = ping_error
+        self.ping_calls: list[bool] = []
 
     def thread_id(self) -> int:
         return self._thread_id_value
 
+    def ping(self, reconnect: bool = True) -> None:
+        self.ping_calls.append(reconnect)
+        if self.ping_error is not None:
+            raise self.ping_error
+
 
 class FakeCursor:
     def __init__(
@@ -177,6 +188,48 @@ def test_list_databases_with_and_without_description() -> None:
     assert empty[0].rows is None
 
 
+def test_ping_reports_connected_without_reconnecting() -> None:
+    connection = FakeConnection()
+    cursor = FakeCursor(query_results={}, connection=connection)
+
+    assert ping(cursor) == [SQLResult(status='Connected')]
+    assert connection.ping_calls == [False]
+
+
+def test_ping_reports_not_connected_on_pymysql_error() -> None:
+    connection = FakeConnection(ping_error=Error('connection lost'))
+    cursor = FakeCursor(query_results={}, connection=connection)
+
+    assert ping(cursor) == [SQLResult(status='Not connected')]
+    assert connection.ping_calls == [False]
+
+
+def test_ping_propagates_unrelated_errors() -> None:
+    connection = FakeConnection(ping_error=RuntimeError('unexpected'))
+    cursor = FakeCursor(query_results={}, connection=connection)
+
+    with pytest.raises(RuntimeError, match='unexpected'):
+        ping(cursor)
+
+
+def test_ping_rejects_arguments_without_contacting_server() -> None:
+    connection = FakeConnection()
+    cursor = FakeCursor(query_results={}, connection=connection)
+
+    assert ping(cursor, arg='unexpected') == [SQLResult(status='Syntax: /ping.')]
+    assert connection.ping_calls == []
+
+
+def test_ping_command_registration() -> None:
+    command = special_main.COMMANDS[r'\ping']
+
+    assert command.handler is ping
+    assert command.usage == '/ping'
+    assert command.description == 'Check the connection.'
+    assert command.completion_snippet == 'check connection'
+    assert special_main.COMMANDS['/ping'].handler is ping
+
+
 def test_status_uses_global_queries_decodes_bytes_and_formats_stats(monkeypatch) -> None:
     monkeypatch.setattr(dbcommands, '__version__', '9.9.9')
     monkeypatch.setattr(dbcommands.platform, 'python_implementation', lambda: 'CPython')

test/pytests/test_special_init.py

@@ -68,6 +68,7 @@ def test_special_init_reexports_dbcommands(load_special: Callable[[bool], Module
 
     assert special.list_databases is dbcommands.list_databases
     assert special.list_tables is dbcommands.list_tables
+    assert special.ping is dbcommands.ping
     assert special.status is dbcommands.status