Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions changelog.md
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@ Features
* Let the `--dsn` argument accept literal DSNs as well as aliases.
* Accept `--character-set` as an alias for `--charset` at the CLI.
* Add SSL/TLS version to `status` output.
* Add prompt format string for SSL/TLS version of the connection.


Bug Fixes
Expand Down
8 changes: 8 additions & 0 deletions mycli/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -66,6 +66,7 @@
from mycli.packages.prompt_utils import confirm, confirm_destructive_query
from mycli.packages.special.favoritequeries import FavoriteQueries
from mycli.packages.special.main import ArgType
from mycli.packages.special.utils import get_ssl_version
from mycli.packages.sqlresult import SQLResult
from mycli.packages.tabular_output import sql_format
from mycli.packages.toolkit.history import FileHistoryWithTimestamp
Expand Down Expand Up @@ -1478,6 +1479,13 @@ def get_prompt(self, string: str) -> str:
string = string.replace("\\K", sqlexecute.socket or str(sqlexecute.port))
string = string.replace("\\A", self.dsn_alias or "(none)")
string = string.replace("\\_", " ")
# jump through hoops for the test environment and for efficiency
if hasattr(sqlexecute, 'conn') and sqlexecute.conn is not None:
if '\\T' in string:
with sqlexecute.conn.cursor() as cur:
string = string.replace('\\T', get_ssl_version(cur) or '(none)')
else:
string = string.replace('\\T', '(none)')
return string

def run_query(
Expand Down
1 change: 1 addition & 0 deletions mycli/myclirc
Original file line number Diff line number Diff line change
Expand Up @@ -113,6 +113,7 @@ wider_completion_menu = False
# * \J - full connection socket path
# * \k - connection socket basename OR the port
# * \K - full connection socket path OR the port
# * \T - connection SSL/TLS version
# * \t - database vendor (Percona, MySQL, MariaDB, TiDB)
# * \u - username
# * \A - DSN alias
Expand Down
25 changes: 16 additions & 9 deletions mycli/packages/special/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,11 +2,12 @@
import os
import subprocess

import pymysql
from pymysql.cursors import Cursor

logger = logging.getLogger(__name__)

CACHED_SSL_VERSION: dict[int, str | None] = {}
CACHED_SSL_VERSION: dict[tuple, str | None] = {}


def handle_cd_command(arg: str) -> tuple[bool, str | None]:
Expand Down Expand Up @@ -56,18 +57,24 @@ def format_uptime(uptime_in_seconds: str) -> str:


def get_ssl_version(cur: Cursor) -> str | None:
if cur.connection.thread_id() in CACHED_SSL_VERSION:
return CACHED_SSL_VERSION[cur.connection.thread_id()] or None
cache_key = (id(cur.connection), cur.connection.thread_id())

if cache_key in CACHED_SSL_VERSION:
return CACHED_SSL_VERSION[cache_key] or None

query = 'SHOW STATUS LIKE "Ssl_version"'
logger.debug(query)
cur.execute(query)

ssl_version = None
if one := cur.fetchone():
CACHED_SSL_VERSION[cur.connection.thread_id()] = one[1]
ssl_version = one[1] or None
else:
CACHED_SSL_VERSION[cur.connection.thread_id()] = ''

try:
cur.execute(query)
if one := cur.fetchone():
CACHED_SSL_VERSION[cache_key] = one[1]
ssl_version = one[1] or None
else:
CACHED_SSL_VERSION[cache_key] = ''
except pymysql.err.OperationalError:
pass

return ssl_version
1 change: 1 addition & 0 deletions test/myclirc
Original file line number Diff line number Diff line change
Expand Up @@ -111,6 +111,7 @@ wider_completion_menu = False
# * \J - full connection socket path
# * \k - connection socket basename OR the port
# * \K - full connection socket path OR the port
# * \T - connection SSL/TLS version
# * \t - database vendor (Percona, MySQL, MariaDB, TiDB)
# * \u - username
# * \A - DSN alias
Expand Down