1from __future__ import annotations
 2
 3import signal
 4from typing import Any
 5
 6from plain.models.backends.base.client import BaseDatabaseClient
 7
 8
 9class DatabaseClient(BaseDatabaseClient):
10    executable_name = "mysql"
11
12    @classmethod
13    def settings_to_cmd_args_env(
14        cls, settings_dict: dict[str, Any], parameters: list[str]
15    ) -> tuple[list[str], dict[str, str] | None]:
16        args = [cls.executable_name]
17        env = None
18        database = settings_dict["OPTIONS"].get(
19            "database",
20            settings_dict["OPTIONS"].get("db", settings_dict["NAME"]),
21        )
22        user = settings_dict["OPTIONS"].get("user", settings_dict["USER"])
23        password = settings_dict["OPTIONS"].get(
24            "password",
25            settings_dict["OPTIONS"].get("passwd", settings_dict["PASSWORD"]),
26        )
27        host = settings_dict["OPTIONS"].get("host", settings_dict["HOST"])
28        port = settings_dict["OPTIONS"].get("port", settings_dict["PORT"])
29        server_ca = settings_dict["OPTIONS"].get("ssl", {}).get("ca")
30        client_cert = settings_dict["OPTIONS"].get("ssl", {}).get("cert")
31        client_key = settings_dict["OPTIONS"].get("ssl", {}).get("key")
32        defaults_file = settings_dict["OPTIONS"].get("read_default_file")
33        charset = settings_dict["OPTIONS"].get("charset")
34        # Seems to be no good way to set sql_mode with CLI.
35
36        if defaults_file:
37            args += [f"--defaults-file={defaults_file}"]
38        if user:
39            args += [f"--user={user}"]
40        if password:
41            # The MYSQL_PWD environment variable usage is discouraged per
42            # MySQL's documentation due to the possibility of exposure through
43            # `ps` on old Unix flavors but --password suffers from the same
44            # flaw on even more systems. Usage of an environment variable also
45            # prevents password exposure if the subprocess.run(check=True) call
46            # raises a CalledProcessError since the string representation of
47            # the latter includes all of the provided `args`.
48            env = {"MYSQL_PWD": password}
49        if host:
50            if "/" in host:
51                args += [f"--socket={host}"]
52            else:
53                args += [f"--host={host}"]
54        if port:
55            args += [f"--port={port}"]
56        if server_ca:
57            args += [f"--ssl-ca={server_ca}"]
58        if client_cert:
59            args += [f"--ssl-cert={client_cert}"]
60        if client_key:
61            args += [f"--ssl-key={client_key}"]
62        if charset:
63            args += [f"--default-character-set={charset}"]
64        if database:
65            args += [database]
66        args.extend(parameters)
67        return args, env
68
69    def runshell(self, parameters: list[str]) -> None:
70        sigint_handler = signal.getsignal(signal.SIGINT)
71        try:
72            # Allow SIGINT to pass to mysql to abort queries.
73            signal.signal(signal.SIGINT, signal.SIG_IGN)
74            super().runshell(parameters)
75        finally:
76            # Restore the original SIGINT handler.
77            signal.signal(signal.SIGINT, sigint_handler)