diff --git a/venv/Lib/site-packages/sqlalchemy/dialects/mysql/dml.py b/venv/Lib/site-packages/sqlalchemy/dialects/mysql/dml.py new file mode 100644 index 0000000..9d3258b --- /dev/null +++ b/venv/Lib/site-packages/sqlalchemy/dialects/mysql/dml.py @@ -0,0 +1,224 @@ +# dialects/mysql/dml.py +# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors +# +# +# 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 + +from typing import Any +from typing import Dict +from typing import List +from typing import Mapping +from typing import Optional +from typing import Tuple +from typing import Union + +from ... import exc +from ... import util +from ...sql._typing import _DMLTableArgument +from ...sql.base import _exclusive_against +from ...sql.base import _generative +from ...sql.base import ColumnCollection +from ...sql.base import ReadOnlyColumnCollection +from ...sql.dml import Insert as StandardInsert +from ...sql.elements import ClauseElement +from ...sql.elements import KeyedColumnElement +from ...sql.expression import alias +from ...sql.selectable import NamedFromClause +from ...util.typing import Self + +__all__ = ("Insert", "insert") + + +def insert(table: _DMLTableArgument) -> Insert: + """Construct a MySQL/MariaDB-specific variant :class:`_mysql.Insert` + construct. + + .. container:: inherited_member + + The :func:`sqlalchemy.dialects.mysql.insert` function creates + a :class:`sqlalchemy.dialects.mysql.Insert`. This class is based + on the dialect-agnostic :class:`_sql.Insert` construct which may + be constructed using the :func:`_sql.insert` function in + SQLAlchemy Core. + + The :class:`_mysql.Insert` construct includes additional methods + :meth:`_mysql.Insert.on_duplicate_key_update`. + + """ + return Insert(table) + + +class Insert(StandardInsert): + """MySQL-specific implementation of INSERT. + + Adds methods for MySQL-specific syntaxes such as ON DUPLICATE KEY UPDATE. + + The :class:`~.mysql.Insert` object is created using the + :func:`sqlalchemy.dialects.mysql.insert` function. + + .. versionadded:: 1.2 + + """ + + stringify_dialect = "mysql" + inherit_cache = False + + @property + def inserted( + self, + ) -> ReadOnlyColumnCollection[str, KeyedColumnElement[Any]]: + """Provide the "inserted" namespace for an ON DUPLICATE KEY UPDATE + statement + + MySQL's ON DUPLICATE KEY UPDATE clause allows reference to the row + that would be inserted, via a special function called ``VALUES()``. + This attribute provides all columns in this row to be referenceable + such that they will render within a ``VALUES()`` function inside the + ON DUPLICATE KEY UPDATE clause. The attribute is named ``.inserted`` + so as not to conflict with the existing + :meth:`_expression.Insert.values` method. + + .. tip:: The :attr:`_mysql.Insert.inserted` attribute is an instance + of :class:`_expression.ColumnCollection`, which provides an + interface the same as that of the :attr:`_schema.Table.c` + collection described at :ref:`metadata_tables_and_columns`. + With this collection, ordinary names are accessible like attributes + (e.g. ``stmt.inserted.some_column``), but special names and + dictionary method names should be accessed using indexed access, + such as ``stmt.inserted["column name"]`` or + ``stmt.inserted["values"]``. See the docstring for + :class:`_expression.ColumnCollection` for further examples. + + .. seealso:: + + :ref:`mysql_insert_on_duplicate_key_update` - example of how + to use :attr:`_expression.Insert.inserted` + + """ + return self.inserted_alias.columns + + @util.memoized_property + def inserted_alias(self) -> NamedFromClause: + return alias(self.table, name="inserted") + + @_generative + @_exclusive_against( + "_post_values_clause", + msgs={ + "_post_values_clause": "This Insert construct already " + "has an ON DUPLICATE KEY clause present" + }, + ) + def on_duplicate_key_update(self, *args: _UpdateArg, **kw: Any) -> Self: + r""" + Specifies the ON DUPLICATE KEY UPDATE clause. + + :param \**kw: Column keys linked to UPDATE values. The + values may be any SQL expression or supported literal Python + values. + + .. warning:: This dictionary does **not** take into account + Python-specified default UPDATE values or generation functions, + e.g. those specified using :paramref:`_schema.Column.onupdate`. + These values will not be exercised for an ON DUPLICATE KEY UPDATE + style of UPDATE, unless values are manually specified here. + + :param \*args: As an alternative to passing key/value parameters, + a dictionary or list of 2-tuples can be passed as a single positional + argument. + + Passing a single dictionary is equivalent to the keyword argument + form:: + + insert().on_duplicate_key_update({"name": "some name"}) + + Passing a list of 2-tuples indicates that the parameter assignments + in the UPDATE clause should be ordered as sent, in a manner similar + to that described for the :class:`_expression.Update` + construct overall + in :ref:`tutorial_parameter_ordered_updates`:: + + insert().on_duplicate_key_update( + [ + ("name", "some name"), + ("value", "some value"), + ] + ) + + .. versionchanged:: 1.3 parameters can be specified as a dictionary + or list of 2-tuples; the latter form provides for parameter + ordering. + + + .. versionadded:: 1.2 + + .. seealso:: + + :ref:`mysql_insert_on_duplicate_key_update` + + """ + if args and kw: + raise exc.ArgumentError( + "Can't pass kwargs and positional arguments simultaneously" + ) + + if args: + if len(args) > 1: + raise exc.ArgumentError( + "Only a single dictionary or list of tuples " + "is accepted positionally." + ) + values = args[0] + else: + values = kw + + self._post_values_clause = OnDuplicateClause( + self.inserted_alias, values + ) + return self + + +class OnDuplicateClause(ClauseElement): + __visit_name__ = "on_duplicate_key_update" + + _parameter_ordering: Optional[List[str]] = None + + update: Dict[str, Any] + stringify_dialect = "mysql" + + def __init__( + self, inserted_alias: NamedFromClause, update: _UpdateArg + ) -> None: + self.inserted_alias = inserted_alias + + # auto-detect that parameters should be ordered. This is copied from + # Update._proces_colparams(), however we don't look for a special flag + # in this case since we are not disambiguating from other use cases as + # we are in Update.values(). + if isinstance(update, list) and ( + update and isinstance(update[0], tuple) + ): + self._parameter_ordering = [key for key, value in update] + update = dict(update) + + if isinstance(update, dict): + if not update: + raise ValueError( + "update parameter dictionary must not be empty" + ) + elif isinstance(update, ColumnCollection): + update = dict(update) + else: + raise ValueError( + "update parameter must be a non-empty dictionary " + "or a ColumnCollection such as the `.c.` collection " + "of a Table object" + ) + self.update = update + + +_UpdateArg = Union[ + Mapping[Any, Any], List[Tuple[str, Any]], ColumnCollection[Any, Any] +] diff --git a/venv/Lib/site-packages/sqlalchemy/dialects/mysql/enumerated.py b/venv/Lib/site-packages/sqlalchemy/dialects/mysql/enumerated.py new file mode 100644 index 0000000..4eba21e --- /dev/null +++ b/venv/Lib/site-packages/sqlalchemy/dialects/mysql/enumerated.py @@ -0,0 +1,282 @@ +# dialects/mysql/enumerated.py +# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors +# +# +# 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 enum +import re +from typing import Any +from typing import Dict +from typing import Optional +from typing import Set +from typing import Type +from typing import TYPE_CHECKING +from typing import Union + +from .types import _StringType +from ... import exc +from ... import sql +from ... import util +from ...sql import sqltypes +from ...sql import type_api + +if TYPE_CHECKING: + from ...engine.interfaces import Dialect + from ...sql.elements import ColumnElement + from ...sql.type_api import _BindProcessorType + from ...sql.type_api import _ResultProcessorType + from ...sql.type_api import TypeEngine + from ...sql.type_api import TypeEngineMixin + + +class ENUM(type_api.NativeForEmulated, sqltypes.Enum, _StringType): + """MySQL ENUM type.""" + + __visit_name__ = "ENUM" + + native_enum = True + + def __init__(self, *enums: Union[str, Type[enum.Enum]], **kw: Any) -> None: + """Construct an ENUM. + + E.g.:: + + Column("myenum", ENUM("foo", "bar", "baz")) + + :param enums: The range of valid values for this ENUM. Values in + enums are not quoted, they will be escaped and surrounded by single + quotes when generating the schema. This object may also be a + PEP-435-compliant enumerated type. + + .. versionadded: 1.1 added support for PEP-435-compliant enumerated + types. + + :param strict: This flag has no effect. + + .. versionchanged:: The MySQL ENUM type as well as the base Enum + type now validates all Python data values. + + :param charset: Optional, a column-level character set for this string + value. Takes precedence to 'ascii' or 'unicode' short-hand. + + :param collation: Optional, a column-level collation for this string + value. Takes precedence to 'binary' short-hand. + + :param ascii: Defaults to False: short-hand for the ``latin1`` + character set, generates ASCII in schema. + + :param unicode: Defaults to False: short-hand for the ``ucs2`` + character set, generates UNICODE in schema. + + :param binary: Defaults to False: short-hand, pick the binary + collation type that matches the column's character set. Generates + BINARY in schema. This does not affect the type of data stored, + only the collation of character data. + + """ + kw.pop("strict", None) + self._enum_init(enums, kw) # type: ignore[arg-type] + _StringType.__init__(self, length=self.length, **kw) + + @classmethod + def adapt_emulated_to_native( + cls, + impl: Union[TypeEngine[Any], TypeEngineMixin], + **kw: Any, + ) -> ENUM: + """Produce a MySQL native :class:`.mysql.ENUM` from plain + :class:`.Enum`. + + """ + if TYPE_CHECKING: + assert isinstance(impl, ENUM) + kw.setdefault("validate_strings", impl.validate_strings) + kw.setdefault("values_callable", impl.values_callable) + kw.setdefault("omit_aliases", impl._omit_aliases) + return cls(**kw) + + def _object_value_for_elem(self, elem: str) -> Union[str, enum.Enum]: + # mysql sends back a blank string for any value that + # was persisted that was not in the enums; that is, it does no + # validation on the incoming data, it "truncates" it to be + # the blank string. Return it straight. + if elem == "": + return elem + else: + return super()._object_value_for_elem(elem) + + def __repr__(self) -> str: + return util.generic_repr( + self, to_inspect=[ENUM, _StringType, sqltypes.Enum] + ) + + +# TODO: SET is a string as far as configuration but does not act like +# a string at the python level. We either need to make a py-type agnostic +# version of String as a base to be used for this, make this some kind of +# TypeDecorator, or just vendor it out as its own type. +class SET(_StringType): + """MySQL SET type.""" + + __visit_name__ = "SET" + + def __init__(self, *values: str, **kw: Any): + """Construct a SET. + + E.g.:: + + Column("myset", SET("foo", "bar", "baz")) + + The list of potential values is required in the case that this + set will be used to generate DDL for a table, or if the + :paramref:`.SET.retrieve_as_bitwise` flag is set to True. + + :param values: The range of valid values for this SET. The values + are not quoted, they will be escaped and surrounded by single + quotes when generating the schema. + + :param convert_unicode: Same flag as that of + :paramref:`.String.convert_unicode`. + + :param collation: same as that of :paramref:`.String.collation` + + :param charset: same as that of :paramref:`.VARCHAR.charset`. + + :param ascii: same as that of :paramref:`.VARCHAR.ascii`. + + :param unicode: same as that of :paramref:`.VARCHAR.unicode`. + + :param binary: same as that of :paramref:`.VARCHAR.binary`. + + :param retrieve_as_bitwise: if True, the data for the set type will be + persisted and selected using an integer value, where a set is coerced + into a bitwise mask for persistence. MySQL allows this mode which + has the advantage of being able to store values unambiguously, + such as the blank string ``''``. The datatype will appear + as the expression ``col + 0`` in a SELECT statement, so that the + value is coerced into an integer value in result sets. + This flag is required if one wishes + to persist a set that can store the blank string ``''`` as a value. + + .. warning:: + + When using :paramref:`.mysql.SET.retrieve_as_bitwise`, it is + essential that the list of set values is expressed in the + **exact same order** as exists on the MySQL database. + + """ + self.retrieve_as_bitwise = kw.pop("retrieve_as_bitwise", False) + self.values = tuple(values) + if not self.retrieve_as_bitwise and "" in values: + raise exc.ArgumentError( + "Can't use the blank value '' in a SET without " + "setting retrieve_as_bitwise=True" + ) + if self.retrieve_as_bitwise: + self._inversed_bitmap: Dict[str, int] = { + value: 2**idx for idx, value in enumerate(self.values) + } + self._bitmap: Dict[int, str] = { + 2**idx: value for idx, value in enumerate(self.values) + } + length = max([len(v) for v in values] + [0]) + kw.setdefault("length", length) + super().__init__(**kw) + + def column_expression( + self, colexpr: ColumnElement[Any] + ) -> ColumnElement[Any]: + if self.retrieve_as_bitwise: + return sql.type_coerce( + sql.type_coerce(colexpr, sqltypes.Integer) + 0, self + ) + else: + return colexpr + + def result_processor( + self, dialect: Dialect, coltype: Any + ) -> Optional[_ResultProcessorType[Any]]: + if self.retrieve_as_bitwise: + + def process(value: Union[str, int, None]) -> Optional[Set[str]]: + if value is not None: + value = int(value) + + return set(util.map_bits(self._bitmap.__getitem__, value)) + else: + return None + + else: + super_convert = super().result_processor(dialect, coltype) + + def process(value: Union[str, Set[str], None]) -> Optional[Set[str]]: # type: ignore[misc] # noqa: E501 + if isinstance(value, str): + # MySQLdb returns a string, let's parse + if super_convert: + value = super_convert(value) + assert value is not None + if TYPE_CHECKING: + assert isinstance(value, str) + return set(re.findall(r"[^,]+", value)) + else: + # mysql-connector-python does a naive + # split(",") which throws in an empty string + if value is not None: + value.discard("") + return value + + return process + + def bind_processor( + self, dialect: Dialect + ) -> _BindProcessorType[Union[str, int]]: + super_convert = super().bind_processor(dialect) + if self.retrieve_as_bitwise: + + def process( + value: Union[str, int, set[str], None], + ) -> Union[str, int, None]: + if value is None: + return None + elif isinstance(value, (int, str)): + if super_convert: + return super_convert(value) # type: ignore[arg-type, no-any-return] # noqa: E501 + else: + return value + else: + int_value = 0 + for v in value: + int_value |= self._inversed_bitmap[v] + return int_value + + else: + + def process( + value: Union[str, int, set[str], None], + ) -> Union[str, int, None]: + # accept strings and int (actually bitflag) values directly + if value is not None and not isinstance(value, (int, str)): + value = ",".join(value) + if super_convert: + return super_convert(value) # type: ignore + else: + return value + + return process + + def adapt(self, cls: type, **kw: Any) -> Any: + kw["retrieve_as_bitwise"] = self.retrieve_as_bitwise + return util.constructor_copy(self, cls, *self.values, **kw) + + def __repr__(self) -> str: + return util.generic_repr( + self, + to_inspect=[SET, _StringType], + additional_kw=[ + ("retrieve_as_bitwise", False), + ], + ) diff --git a/venv/Lib/site-packages/sqlalchemy/dialects/mysql/expression.py b/venv/Lib/site-packages/sqlalchemy/dialects/mysql/expression.py new file mode 100644 index 0000000..8795290 --- /dev/null +++ b/venv/Lib/site-packages/sqlalchemy/dialects/mysql/expression.py @@ -0,0 +1,146 @@ +# dialects/mysql/expression.py +# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors +# +# +# 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 + +from typing import Any + +from ... import exc +from ... import util +from ...sql import coercions +from ...sql import elements +from ...sql import operators +from ...sql import roles +from ...sql.base import _generative +from ...sql.base import Generative +from ...util.typing import Self + + +class match(Generative, elements.BinaryExpression[Any]): + """Produce a ``MATCH (X, Y) AGAINST ('TEXT')`` clause. + + E.g.:: + + from sqlalchemy import desc + from sqlalchemy.dialects.mysql import match + + match_expr = match( + users_table.c.firstname, + users_table.c.lastname, + against="Firstname Lastname", + ) + + stmt = ( + select(users_table) + .where(match_expr.in_boolean_mode()) + .order_by(desc(match_expr)) + ) + + Would produce SQL resembling: + + .. sourcecode:: sql + + SELECT id, firstname, lastname + FROM user + WHERE MATCH(firstname, lastname) AGAINST (:param_1 IN BOOLEAN MODE) + ORDER BY MATCH(firstname, lastname) AGAINST (:param_2) DESC + + The :func:`_mysql.match` function is a standalone version of the + :meth:`_sql.ColumnElement.match` method available on all + SQL expressions, as when :meth:`_expression.ColumnElement.match` is + used, but allows to pass multiple columns + + :param cols: column expressions to match against + + :param against: expression to be compared towards + + :param in_boolean_mode: boolean, set "boolean mode" to true + + :param in_natural_language_mode: boolean , set "natural language" to true + + :param with_query_expansion: boolean, set "query expansion" to true + + .. versionadded:: 1.4.19 + + .. seealso:: + + :meth:`_expression.ColumnElement.match` + + """ + + __visit_name__ = "mysql_match" + + inherit_cache = True + modifiers: util.immutabledict[str, Any] + + def __init__(self, *cols: elements.ColumnElement[Any], **kw: Any): + if not cols: + raise exc.ArgumentError("columns are required") + + against = kw.pop("against", None) + + if against is None: + raise exc.ArgumentError("against is required") + against = coercions.expect( + roles.ExpressionElementRole, + against, + ) + + left = elements.BooleanClauseList._construct_raw( + operators.comma_op, + clauses=cols, + ) + left.group = False + + flags = util.immutabledict( + { + "mysql_boolean_mode": kw.pop("in_boolean_mode", False), + "mysql_natural_language": kw.pop( + "in_natural_language_mode", False + ), + "mysql_query_expansion": kw.pop("with_query_expansion", False), + } + ) + + if kw: + raise exc.ArgumentError("unknown arguments: %s" % (", ".join(kw))) + + super().__init__(left, against, operators.match_op, modifiers=flags) + + @_generative + def in_boolean_mode(self) -> Self: + """Apply the "IN BOOLEAN MODE" modifier to the MATCH expression. + + :return: a new :class:`_mysql.match` instance with modifications + applied. + """ + + self.modifiers = self.modifiers.union({"mysql_boolean_mode": True}) + return self + + @_generative + def in_natural_language_mode(self) -> Self: + """Apply the "IN NATURAL LANGUAGE MODE" modifier to the MATCH + expression. + + :return: a new :class:`_mysql.match` instance with modifications + applied. + """ + + self.modifiers = self.modifiers.union({"mysql_natural_language": True}) + return self + + @_generative + def with_query_expansion(self) -> Self: + """Apply the "WITH QUERY EXPANSION" modifier to the MATCH expression. + + :return: a new :class:`_mysql.match` instance with modifications + applied. + """ + + self.modifiers = self.modifiers.union({"mysql_query_expansion": True}) + return self diff --git a/venv/Lib/site-packages/sqlalchemy/dialects/mysql/json.py b/venv/Lib/site-packages/sqlalchemy/dialects/mysql/json.py new file mode 100644 index 0000000..9ed3ffe --- /dev/null +++ b/venv/Lib/site-packages/sqlalchemy/dialects/mysql/json.py @@ -0,0 +1,91 @@ +# dialects/mysql/json.py +# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors +# +# +# 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 + +from typing import Any +from typing import TYPE_CHECKING + +from ... import types as sqltypes + +if TYPE_CHECKING: + from ...engine.interfaces import Dialect + from ...sql.type_api import _BindProcessorType + from ...sql.type_api import _LiteralProcessorType + + +class JSON(sqltypes.JSON): + """MySQL JSON type. + + MySQL supports JSON as of version 5.7. + MariaDB supports JSON (as an alias for LONGTEXT) as of version 10.2. + + :class:`_mysql.JSON` is used automatically whenever the base + :class:`_types.JSON` datatype is used against a MySQL or MariaDB backend. + + .. seealso:: + + :class:`_types.JSON` - main documentation for the generic + cross-platform JSON datatype. + + The :class:`.mysql.JSON` type supports persistence of JSON values + as well as the core index operations provided by :class:`_types.JSON` + datatype, by adapting the operations to render the ``JSON_EXTRACT`` + function at the database level. + + """ + + pass + + +class _FormatTypeMixin: + def _format_value(self, value: Any) -> str: + raise NotImplementedError() + + def bind_processor(self, dialect: Dialect) -> _BindProcessorType[Any]: + super_proc = self.string_bind_processor(dialect) # type: ignore[attr-defined] # noqa: E501 + + def process(value: Any) -> Any: + value = self._format_value(value) + if super_proc: + value = super_proc(value) + return value + + return process + + def literal_processor( + self, dialect: Dialect + ) -> _LiteralProcessorType[Any]: + super_proc = self.string_literal_processor(dialect) # type: ignore[attr-defined] # noqa: E501 + + def process(value: Any) -> str: + value = self._format_value(value) + if super_proc: + value = super_proc(value) + return value # type: ignore[no-any-return] + + return process + + +class JSONIndexType(_FormatTypeMixin, sqltypes.JSON.JSONIndexType): + def _format_value(self, value: Any) -> str: + if isinstance(value, int): + formatted_value = "$[%s]" % value + else: + formatted_value = '$."%s"' % value + return formatted_value + + +class JSONPathType(_FormatTypeMixin, sqltypes.JSON.JSONPathType): + def _format_value(self, value: Any) -> str: + return "$%s" % ( + "".join( + [ + "[%s]" % elem if isinstance(elem, int) else '."%s"' % elem + for elem in value + ] + ) + ) diff --git a/venv/Lib/site-packages/sqlalchemy/dialects/mysql/mariadb.py b/venv/Lib/site-packages/sqlalchemy/dialects/mysql/mariadb.py new file mode 100644 index 0000000..7c33abd --- /dev/null +++ b/venv/Lib/site-packages/sqlalchemy/dialects/mysql/mariadb.py @@ -0,0 +1,72 @@ +# dialects/mysql/mariadb.py +# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors +# +# +# 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 + +from typing import Any + +from .base import MariaDBIdentifierPreparer +from .base import MySQLDialect +from .base import MySQLIdentifierPreparer +from .base import MySQLTypeCompiler +from ...sql import sqltypes + + +class INET4(sqltypes.TypeEngine[str]): + """INET4 column type for MariaDB + + .. versionadded:: 2.0.37 + """ + + __visit_name__ = "INET4" + + +class INET6(sqltypes.TypeEngine[str]): + """INET6 column type for MariaDB + + .. versionadded:: 2.0.37 + """ + + __visit_name__ = "INET6" + + +class MariaDBTypeCompiler(MySQLTypeCompiler): + def visit_INET4(self, type_: INET4, **kwargs: Any) -> str: + return "INET4" + + def visit_INET6(self, type_: INET6, **kwargs: Any) -> str: + return "INET6" + + +class MariaDBDialect(MySQLDialect): + is_mariadb = True + supports_statement_cache = True + name = "mariadb" + preparer: type[MySQLIdentifierPreparer] = MariaDBIdentifierPreparer + type_compiler_cls = MariaDBTypeCompiler + + +def loader(driver: str) -> type[MariaDBDialect]: + dialect_mod = __import__( + "sqlalchemy.dialects.mysql.%s" % driver + ).dialects.mysql + + driver_mod = getattr(dialect_mod, driver) + if hasattr(driver_mod, "mariadb_dialect"): + driver_cls = driver_mod.mariadb_dialect + return driver_cls # type: ignore[no-any-return] + else: + driver_cls = driver_mod.dialect + + return type( + "MariaDBDialect_%s" % driver, + ( + MariaDBDialect, + driver_cls, + ), + {"supports_statement_cache": True}, + )