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