ZpenrMessangerProject/venv/lib/python3.12/site-packages/sqlalchemy/dialects/mysql/reflection.py

# dialects/mysql/reflection.py
# Copyright (C) 2005-2025 the SQLAlchemy authors and contributors
# <see AUTHORS file>
#
# This module is part of SQLAlchemy and is released under
# the MIT License: https://www.opensource.org/licenses/mit-license.php
from __future__ import annotations

import re
from typing import Any
from typing import Callable
from typing import Dict
from typing import List
from typing import Optional
from typing import overload
from typing import Sequence
from typing import Tuple
from typing import TYPE_CHECKING
from typing import Union

from .enumerated import ENUM
from .enumerated import SET
from .types import DATETIME
from .types import TIME
from .types import TIMESTAMP
from ... import types as sqltypes
from ... import util
from ...util.typing import Literal

if TYPE_CHECKING:
    from .base import MySQLDialect
    from .base import MySQLIdentifierPreparer
    from ...engine.interfaces import ReflectedColumn


class ReflectedState:
    """Stores raw information about a SHOW CREATE TABLE statement."""

    charset: Optional[str]

    def __init__(self) -> None:
        self.columns: List[ReflectedColumn] = []
        self.table_options: Dict[str, str] = {}
        self.table_name: Optional[str] = None
        self.keys: List[Dict[str, Any]] = []
        self.fk_constraints: List[Dict[str, Any]] = []
        self.ck_constraints: List[Dict[str, Any]] = []


class MySQLTableDefinitionParser:
    """Parses the results of a SHOW CREATE TABLE statement."""

    def __init__(
        self, dialect: MySQLDialect, preparer: MySQLIdentifierPreparer
    ):
        self.dialect = dialect
        self.preparer = preparer
        self._prep_regexes()

    def parse(
        self, show_create: str, charset: Optional[str]
    ) -> ReflectedState:
        state = ReflectedState()
        state.charset = charset
        for line in re.split(r"\r?\n", show_create):
            if line.startswith("  " + self.preparer.initial_quote):
                self._parse_column(line, state)
            # a regular table options line
            elif line.startswith(") "):
                self._parse_table_options(line, state)
            # an ANSI-mode table options line
            elif line == ")":
                pass
            elif line.startswith("CREATE "):
                self._parse_table_name(line, state)
            elif "PARTITION" in line:
                self._parse_partition_options(line, state)
            # Not present in real reflection, but may be if
            # loading from a file.
            elif not line:
                pass
            else:
                type_, spec = self._parse_constraints(line)
                if type_ is None:
                    util.warn("Unknown schema content: %r" % line)
                elif type_ == "key":
                    state.keys.append(spec)  # type: ignore[arg-type]
                elif type_ == "fk_constraint":
                    state.fk_constraints.append(spec)  # type: ignore[arg-type]
                elif type_ == "ck_constraint":
                    state.ck_constraints.append(spec)  # type: ignore[arg-type]
                else:
                    pass
        return state

    def _check_view(self, sql: str) -> bool:
        return bool(self._re_is_view.match(sql))

    def _parse_constraints(self, line: str) -> Union[
        Tuple[None, str],
        Tuple[Literal["partition"], str],
        Tuple[
            Literal["ck_constraint", "fk_constraint", "key"], Dict[str, str]
        ],
    ]:
        """Parse a KEY or CONSTRAINT line.

        :param line: A line of SHOW CREATE TABLE output
        """

        # KEY
        m = self._re_key.match(line)
        if m:
            spec = m.groupdict()
            # convert columns into name, length pairs
            # NOTE: we may want to consider SHOW INDEX as the
            # format of indexes in MySQL becomes more complex
            spec["columns"] = self._parse_keyexprs(spec["columns"])
            if spec["version_sql"]:
                m2 = self._re_key_version_sql.match(spec["version_sql"])
                if m2 and m2.groupdict()["parser"]:
                    spec["parser"] = m2.groupdict()["parser"]
            if spec["parser"]:
                spec["parser"] = self.preparer.unformat_identifiers(
                    spec["parser"]
                )[0]
            return "key", spec

        # FOREIGN KEY CONSTRAINT
        m = self._re_fk_constraint.match(line)
        if m:
            spec = m.groupdict()
            spec["table"] = self.preparer.unformat_identifiers(spec["table"])
            spec["local"] = [c[0] for c in self._parse_keyexprs(spec["local"])]
            spec["foreign"] = [
                c[0] for c in self._parse_keyexprs(spec["foreign"])
            ]
            return "fk_constraint", spec

        # CHECK constraint
        m = self._re_ck_constraint.match(line)
        if m:
            spec = m.groupdict()
            return "ck_constraint", spec

        # PARTITION and SUBPARTITION
        m = self._re_partition.match(line)
        if m:
            # Punt!
            return "partition", line

        # No match.
        return (None, line)

    def _parse_table_name(self, line: str, state: ReflectedState) -> None:
        """Extract the table name.

        :param line: The first line of SHOW CREATE TABLE
        """

        regex, cleanup = self._pr_name
        m = regex.match(line)
        if m:
            state.table_name = cleanup(m.group("name"))

    def _parse_table_options(self, line: str, state: ReflectedState) -> None:
        """Build a dictionary of all reflected table-level options.

        :param line: The final line of SHOW CREATE TABLE output.
        """

        options = {}

        if line and line != ")":
            rest_of_line = line
            for regex, cleanup in self._pr_options:
                m = regex.search(rest_of_line)
                if not m:
                    continue
                directive, value = m.group("directive"), m.group("val")
                if cleanup:
                    value = cleanup(value)
                options[directive.lower()] = value
                rest_of_line = regex.sub("", rest_of_line)

        for nope in ("auto_increment", "data directory", "index directory"):
            options.pop(nope, None)

        for opt, val in options.items():
            state.table_options["%s_%s" % (self.dialect.name, opt)] = val

    def _parse_partition_options(
        self, line: str, state: ReflectedState
    ) -> None:
        options = {}
        new_line = line[:]

        while new_line.startswith("(") or new_line.startswith(" "):
            new_line = new_line[1:]

        for regex, cleanup in self._pr_options:
            m = regex.search(new_line)
            if not m or "PARTITION" not in regex.pattern:
                continue

            directive = m.group("directive")
            directive = directive.lower()
            is_subpartition = directive == "subpartition"

            if directive == "partition" or is_subpartition:
                new_line = new_line.replace(") */", "")
                new_line = new_line.replace(",", "")
                if is_subpartition and new_line.endswith(")"):
                    new_line = new_line[:-1]
                if self.dialect.name == "mariadb" and new_line.endswith(")"):
                    if (
                        "MAXVALUE" in new_line
                        or "MINVALUE" in new_line
                        or "ENGINE" in new_line
                    ):
                        # final line of MariaDB partition endswith ")"
                        new_line = new_line[:-1]

                defs = "%s_%s_definitions" % (self.dialect.name, directive)
                options[defs] = new_line

            else:
                directive = directive.replace(" ", "_")
                value = m.group("val")
                if cleanup:
                    value = cleanup(value)
                options[directive] = value
            break

        for opt, val in options.items():
            part_def = "%s_partition_definitions" % (self.dialect.name)
            subpart_def = "%s_subpartition_definitions" % (self.dialect.name)
            if opt == part_def or opt == subpart_def:
                # builds a string of definitions
                if opt not in state.table_options:
                    state.table_options[opt] = val
                else:
                    state.table_options[opt] = "%s, %s" % (
                        state.table_options[opt],
                        val,
                    )
            else:
                state.table_options["%s_%s" % (self.dialect.name, opt)] = val

    def _parse_column(self, line: str, state: ReflectedState) -> None:
        """Extract column details.

        Falls back to a 'minimal support' variant if full parse fails.

        :param line: Any column-bearing line from SHOW CREATE TABLE
        """

        spec = None
        m = self._re_column.match(line)
        if m:
            spec = m.groupdict()
            spec["full"] = True
        else:
            m = self._re_column_loose.match(line)
            if m:
                spec = m.groupdict()
                spec["full"] = False
        if not spec:
            util.warn("Unknown column definition %r" % line)
            return
        if not spec["full"]:
            util.warn("Incomplete reflection of column definition %r" % line)

        name, type_, args = spec["name"], spec["coltype"], spec["arg"]

        try:
            col_type = self.dialect.ischema_names[type_]
        except KeyError:
            util.warn(
                "Did not recognize type '%s' of column '%s'" % (type_, name)
            )
            col_type = sqltypes.NullType

        # Column type positional arguments eg. varchar(32)
        if args is None or args == "":
            type_args = []
        elif args[0] == "'" and args[-1] == "'":
            type_args = self._re_csv_str.findall(args)
        else:
            type_args = [int(v) for v in self._re_csv_int.findall(args)]

        # Column type keyword options
        type_kw = {}

        if issubclass(col_type, (DATETIME, TIME, TIMESTAMP)):
            if type_args:
                type_kw["fsp"] = type_args.pop(0)

        for kw in ("unsigned", "zerofill"):
            if spec.get(kw, False):
                type_kw[kw] = True
        for kw in ("charset", "collate"):
            if spec.get(kw, False):
                type_kw[kw] = spec[kw]
        if issubclass(col_type, (ENUM, SET)):
            type_args = _strip_values(type_args)

            if issubclass(col_type, SET) and "" in type_args:
                type_kw["retrieve_as_bitwise"] = True

        type_instance = col_type(*type_args, **type_kw)

        col_kw: Dict[str, Any] = {}

        # NOT NULL
        col_kw["nullable"] = True
        # this can be "NULL" in the case of TIMESTAMP
        if spec.get("notnull", False) == "NOT NULL":
            col_kw["nullable"] = False
        # For generated columns, the nullability is marked in a different place
        if spec.get("notnull_generated", False) == "NOT NULL":
            col_kw["nullable"] = False

        # AUTO_INCREMENT
        if spec.get("autoincr", False):
            col_kw["autoincrement"] = True
        elif issubclass(col_type, sqltypes.Integer):
            col_kw["autoincrement"] = False

        # DEFAULT
        default = spec.get("default", None)

        if default == "NULL":
            # eliminates the need to deal with this later.
            default = None

        comment = spec.get("comment", None)

        if comment is not None:
            comment = cleanup_text(comment)

        sqltext = spec.get("generated")
        if sqltext is not None:
            computed = dict(sqltext=sqltext)
            persisted = spec.get("persistence")
            if persisted is not None:
                computed["persisted"] = persisted == "STORED"
            col_kw["computed"] = computed

        col_d = dict(
            name=name, type=type_instance, default=default, comment=comment
        )
        col_d.update(col_kw)
        state.columns.append(col_d)  # type: ignore[arg-type]

    def _describe_to_create(
        self,
        table_name: str,
        columns: Sequence[Tuple[str, str, str, str, str, str]],
    ) -> str:
        """Re-format DESCRIBE output as a SHOW CREATE TABLE string.

        DESCRIBE is a much simpler reflection and is sufficient for
        reflecting views for runtime use.  This method formats DDL
        for columns only- keys are omitted.

        :param columns: A sequence of DESCRIBE or SHOW COLUMNS 6-tuples.
          SHOW FULL COLUMNS FROM rows must be rearranged for use with
          this function.
        """

        buffer = []
        for row in columns:
            (name, col_type, nullable, default, extra) = (
                row[i] for i in (0, 1, 2, 4, 5)
            )

            line = [" "]
            line.append(self.preparer.quote_identifier(name))
            line.append(col_type)
            if not nullable:
                line.append("NOT NULL")
            if default:
                if "auto_increment" in default:
                    pass
                elif col_type.startswith("timestamp") and default.startswith(
                    "C"
                ):
                    line.append("DEFAULT")
                    line.append(default)
                elif default == "NULL":
                    line.append("DEFAULT")
                    line.append(default)
                else:
                    line.append("DEFAULT")
                    line.append("'%s'" % default.replace("'", "''"))
            if extra:
                line.append(extra)

            buffer.append(" ".join(line))

        return "".join(
            [
                (
                    "CREATE TABLE %s (\n"
                    % self.preparer.quote_identifier(table_name)
                ),
                ",\n".join(buffer),
                "\n) ",
            ]
        )

    def _parse_keyexprs(
        self, identifiers: str
    ) -> List[Tuple[str, Optional[int], str]]:
        """Unpack '"col"(2),"col" ASC'-ish strings into components."""

        return [
            (colname, int(length) if length else None, modifiers)
            for colname, length, modifiers in self._re_keyexprs.findall(
                identifiers
            )
        ]

    def _prep_regexes(self) -> None:
        """Pre-compile regular expressions."""

        self._pr_options: List[
            Tuple[re.Pattern[Any], Optional[Callable[[str], str]]]
        ] = []

        _final = self.preparer.final_quote

        quotes = dict(
            zip(
                ("iq", "fq", "esc_fq"),
                [
                    re.escape(s)
                    for s in (
                        self.preparer.initial_quote,
                        _final,
                        self.preparer._escape_identifier(_final),
                    )
                ],
            )
        )

        self._pr_name = _pr_compile(
            r"^CREATE (?:\w+ +)?TABLE +"
            r"%(iq)s(?P<name>(?:%(esc_fq)s|[^%(fq)s])+)%(fq)s +\($" % quotes,
            self.preparer._unescape_identifier,
        )

        self._re_is_view = _re_compile(r"^CREATE(?! TABLE)(\s.*)?\sVIEW")

        # `col`,`col2`(32),`col3`(15) DESC
        #
        self._re_keyexprs = _re_compile(
            r"(?:"
            r"(?:%(iq)s((?:%(esc_fq)s|[^%(fq)s])+)%(fq)s)"
            r"(?:\((\d+)\))?(?: +(ASC|DESC))?(?=\,|$))+" % quotes
        )

        # 'foo' or 'foo','bar' or 'fo,o','ba''a''r'
        self._re_csv_str = _re_compile(r"\x27(?:\x27\x27|[^\x27])*\x27")

        # 123 or 123,456
        self._re_csv_int = _re_compile(r"\d+")

        # `colname` <type> [type opts]
        #  (NOT NULL | NULL)
        #   DEFAULT ('value' | CURRENT_TIMESTAMP...)
        #   COMMENT 'comment'
        #  COLUMN_FORMAT (FIXED|DYNAMIC|DEFAULT)
        #  STORAGE (DISK|MEMORY)
        self._re_column = _re_compile(
            r"  "
            r"%(iq)s(?P<name>(?:%(esc_fq)s|[^%(fq)s])+)%(fq)s +"
            r"(?P<coltype>\w+)"
            r"(?:\((?P<arg>(?:\d+|\d+,\d+|"
            r"(?:'(?:''|[^'])*',?)+))\))?"
            r"(?: +(?P<unsigned>UNSIGNED))?"
            r"(?: +(?P<zerofill>ZEROFILL))?"
            r"(?: +CHARACTER SET +(?P<charset>[\w_]+))?"
            r"(?: +COLLATE +(?P<collate>[\w_]+))?"
            r"(?: +(?P<notnull>(?:NOT )?NULL))?"
            r"(?: +DEFAULT +(?P<default>"
            r"(?:NULL|'(?:''|[^'])*'|\(.+?\)|[\-\w\.\(\)]+"
            r"(?: +ON UPDATE [\-\w\.\(\)]+)?)"
            r"))?"
            r"(?: +(?:GENERATED ALWAYS)? ?AS +(?P<generated>\("
            r".*\))? ?(?P<persistence>VIRTUAL|STORED)?"
            r"(?: +(?P<notnull_generated>(?:NOT )?NULL))?"
            r")?"
            r"(?: +(?P<autoincr>AUTO_INCREMENT))?"
            r"(?: +COMMENT +'(?P<comment>(?:''|[^'])*)')?"
            r"(?: +COLUMN_FORMAT +(?P<colfmt>\w+))?"
            r"(?: +STORAGE +(?P<storage>\w+))?"
            r"(?: +(?P<extra>.*))?"
            r",?$" % quotes
        )

        # Fallback, try to parse as little as possible
        self._re_column_loose = _re_compile(
            r"  "
            r"%(iq)s(?P<name>(?:%(esc_fq)s|[^%(fq)s])+)%(fq)s +"
            r"(?P<coltype>\w+)"
            r"(?:\((?P<arg>(?:\d+|\d+,\d+|\x27(?:\x27\x27|[^\x27])+\x27))\))?"
            r".*?(?P<notnull>(?:NOT )NULL)?" % quotes
        )

        # (PRIMARY|UNIQUE|FULLTEXT|SPATIAL) INDEX `name` (USING (BTREE|HASH))?
        # (`col` (ASC|DESC)?, `col` (ASC|DESC)?)
        # KEY_BLOCK_SIZE size | WITH PARSER name  /*!50100 WITH PARSER name */
        self._re_key = _re_compile(
            r"  "
            r"(?:(?P<type>\S+) )?KEY"
            r"(?: +%(iq)s(?P<name>(?:%(esc_fq)s|[^%(fq)s])+)%(fq)s)?"
            r"(?: +USING +(?P<using_pre>\S+))?"
            r" +\((?P<columns>.+?)\)"
            r"(?: +USING +(?P<using_post>\S+))?"
            r"(?: +KEY_BLOCK_SIZE *[ =]? *(?P<keyblock>\S+))?"
            r"(?: +WITH PARSER +(?P<parser>\S+))?"
            r"(?: +COMMENT +(?P<comment>(\x27\x27|\x27([^\x27])*?\x27)+))?"
            r"(?: +/\*(?P<version_sql>.+)\*/ *)?"
            r",?$" % quotes
        )

        # https://forums.mysql.com/read.php?20,567102,567111#msg-567111
        # It means if the MySQL version >= \d+, execute what's in the comment
        self._re_key_version_sql = _re_compile(
            r"\!\d+ " r"(?: *WITH PARSER +(?P<parser>\S+) *)?"
        )

        # CONSTRAINT `name` FOREIGN KEY (`local_col`)
        # REFERENCES `remote` (`remote_col`)
        # MATCH FULL | MATCH PARTIAL | MATCH SIMPLE
        # ON DELETE CASCADE ON UPDATE RESTRICT
        #
        # unique constraints come back as KEYs
        kw = quotes.copy()
        kw["on"] = "RESTRICT|CASCADE|SET NULL|NO ACTION|SET DEFAULT"
        self._re_fk_constraint = _re_compile(
            r"  "
            r"CONSTRAINT +"
            r"%(iq)s(?P<name>(?:%(esc_fq)s|[^%(fq)s])+)%(fq)s +"
            r"FOREIGN KEY +"
            r"\((?P<local>[^\)]+?)\) REFERENCES +"
            r"(?P<table>%(iq)s[^%(fq)s]+%(fq)s"
            r"(?:\.%(iq)s[^%(fq)s]+%(fq)s)?) +"
            r"\((?P<foreign>(?:%(iq)s[^%(fq)s]+%(fq)s(?: *, *)?)+)\)"
            r"(?: +(?P<match>MATCH \w+))?"
            r"(?: +ON DELETE (?P<ondelete>%(on)s))?"
            r"(?: +ON UPDATE (?P<onupdate>%(on)s))?" % kw
        )

        # CONSTRAINT `CONSTRAINT_1` CHECK (`x` > 5)'
        # testing on MariaDB 10.2 shows that the CHECK constraint
        # is returned on a line by itself, so to match without worrying
        # about parenthesis in the expression we go to the end of the line
        self._re_ck_constraint = _re_compile(
            r"  "
            r"CONSTRAINT +"
            r"%(iq)s(?P<name>(?:%(esc_fq)s|[^%(fq)s])+)%(fq)s +"
            r"CHECK +"
            r"\((?P<sqltext>.+)\),?" % kw
        )

        # PARTITION
        #
        # punt!
        self._re_partition = _re_compile(r"(?:.*)(?:SUB)?PARTITION(?:.*)")

        # Table-level options (COLLATE, ENGINE, etc.)
        # Do the string options first, since they have quoted
        # strings we need to get rid of.
        for option in _options_of_type_string:
            self._add_option_string(option)

        for option in (
            "ENGINE",
            "TYPE",
            "AUTO_INCREMENT",
            "AVG_ROW_LENGTH",
            "CHARACTER SET",
            "DEFAULT CHARSET",
            "CHECKSUM",
            "COLLATE",
            "DELAY_KEY_WRITE",
            "INSERT_METHOD",
            "MAX_ROWS",
            "MIN_ROWS",
            "PACK_KEYS",
            "ROW_FORMAT",
            "KEY_BLOCK_SIZE",
            "STATS_SAMPLE_PAGES",
        ):
            self._add_option_word(option)

        for option in (
            "PARTITION BY",
            "SUBPARTITION BY",
            "PARTITIONS",
            "SUBPARTITIONS",
            "PARTITION",
            "SUBPARTITION",
        ):
            self._add_partition_option_word(option)

        self._add_option_regex("UNION", r"\([^\)]+\)")
        self._add_option_regex("TABLESPACE", r".*? STORAGE DISK")
        self._add_option_regex(
            "RAID_TYPE",
            r"\w+\s+RAID_CHUNKS\s*\=\s*\w+RAID_CHUNKSIZE\s*=\s*\w+",
        )

    _optional_equals = r"(?:\s*(?:=\s*)|\s+)"

    def _add_option_string(self, directive: str) -> None:
        regex = r"(?P<directive>%s)%s" r"'(?P<val>(?:[^']|'')*?)'(?!')" % (
            re.escape(directive),
            self._optional_equals,
        )
        self._pr_options.append(_pr_compile(regex, cleanup_text))

    def _add_option_word(self, directive: str) -> None:
        regex = r"(?P<directive>%s)%s" r"(?P<val>\w+)" % (
            re.escape(directive),
            self._optional_equals,
        )
        self._pr_options.append(_pr_compile(regex))

    def _add_partition_option_word(self, directive: str) -> None:
        if directive == "PARTITION BY" or directive == "SUBPARTITION BY":
            regex = r"(?<!\S)(?P<directive>%s)%s" r"(?P<val>\w+.*)" % (
                re.escape(directive),
                self._optional_equals,
            )
        elif directive == "SUBPARTITIONS" or directive == "PARTITIONS":
            regex = r"(?<!\S)(?P<directive>%s)%s" r"(?P<val>\d+)" % (
                re.escape(directive),
                self._optional_equals,
            )
        else:
            regex = r"(?<!\S)(?P<directive>%s)(?!\S)" % (re.escape(directive),)
        self._pr_options.append(_pr_compile(regex))

    def _add_option_regex(self, directive: str, regex: str) -> None:
        regex = r"(?P<directive>%s)%s" r"(?P<val>%s)" % (
            re.escape(directive),
            self._optional_equals,
            regex,
        )
        self._pr_options.append(_pr_compile(regex))


_options_of_type_string = (
    "COMMENT",
    "DATA DIRECTORY",
    "INDEX DIRECTORY",
    "PASSWORD",
    "CONNECTION",
)


@overload
def _pr_compile(
    regex: str, cleanup: Callable[[str], str]
) -> Tuple[re.Pattern[Any], Callable[[str], str]]: ...


@overload
def _pr_compile(
    regex: str, cleanup: None = None
) -> Tuple[re.Pattern[Any], None]: ...


def _pr_compile(
    regex: str, cleanup: Optional[Callable[[str], str]] = None
) -> Tuple[re.Pattern[Any], Optional[Callable[[str], str]]]:
    """Prepare a 2-tuple of compiled regex and callable."""

    return (_re_compile(regex), cleanup)


def _re_compile(regex: str) -> re.Pattern[Any]:
    """Compile a string to regex, I and UNICODE."""

    return re.compile(regex, re.I | re.UNICODE)


def _strip_values(values: Sequence[str]) -> List[str]:
    "Strip reflected values quotes"
    strip_values: List[str] = []
    for a in values:
        if a[0:1] == '"' or a[0:1] == "'":
            # strip enclosing quotes and unquote interior
            a = a[1:-1].replace(a[0] * 2, a[0])
        strip_values.append(a)
    return strip_values


def cleanup_text(raw_text: str) -> str:
    if "\\" in raw_text:
        raw_text = re.sub(
            _control_char_regexp,
            lambda s: _control_char_map[s[0]],  # type: ignore[index]
            raw_text,
        )
    return raw_text.replace("''", "'")


_control_char_map = {
    "\\\\": "\\",
    "\\0": "\0",
    "\\a": "\a",
    "\\b": "\b",
    "\\t": "\t",
    "\\n": "\n",
    "\\v": "\v",
    "\\f": "\f",
    "\\r": "\r",
    # '\\e':'\e',
}
_control_char_regexp = re.compile(
    "|".join(re.escape(k) for k in _control_char_map)
)