From: Federico Caselli Date: Thu, 16 Jul 2020 19:32:52 +0000 (+0200) Subject: Add support for regular expression on supported backend. X-Git-Tag: rel_1_4_0b1~146^2 X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=b1b97ed1fcac777c4f42fdf84e05f8d59f63b679;p=thirdparty%2Fsqlalchemy%2Fsqlalchemy.git Add support for regular expression on supported backend. Two operations have been defined: * :meth:`~.ColumnOperators.regexp_match` implementing a regular expression match like function. * :meth:`~.ColumnOperators.regexp_replace` implementing a regular expression string replace function. Fixes: #1390 Change-Id: I44556846e4668ccf329023613bd26861d5c674e6 --- diff --git a/doc/build/changelog/migration_14.rst b/doc/build/changelog/migration_14.rst index e21c53ef6e..e25d1e8bf2 100644 --- a/doc/build/changelog/migration_14.rst +++ b/doc/build/changelog/migration_14.rst @@ -373,6 +373,42 @@ instead. :ticket:`4689` + +.. _change_1390: + +Support for SQL Regular Expression operators +-------------------------------------------- + +A long awaited feature to add rudimentary support for database regular +expression operators, to complement the :meth:`_sql.ColumnOperators.like` and +:meth:`_sql.ColumnOperators.match` suites of operations. The new features +include :meth:`_sql.ColumnOperators.regexp_match` implementing a regular +expression match like function, and :meth:`_sql.ColumnOperators.regexp_replace` +implementing a regular expression string replace function. + +Supported backends include SQLite, PostgreSQL, MySQL / MariaDB, and Oracle. +The SQLite backend only supports "regexp_match" but not "regexp_replace". + +The regular expression syntaxes and flags are **not backend agnostic**. +A future feature will allow multiple regular expression syntaxes to be +specified at once to switch between different backends on the fly. + +For SQLite, Python's ``re.match()`` function with no additional arguments +is established as the implementation. + +.. seealso:: + + + :meth:`_sql.ColumnOperators.regexp_match` + + :meth:`_sql.ColumnOperators.regexp_replace` + + :ref:`pysqlite_regexp` - SQLite implementation notes + + +:ticket:`1390` + + API and Behavioral Changes - Core ================================== diff --git a/doc/build/changelog/unreleased_14/1390.rst b/doc/build/changelog/unreleased_14/1390.rst new file mode 100644 index 0000000000..85fc189e05 --- /dev/null +++ b/doc/build/changelog/unreleased_14/1390.rst @@ -0,0 +1,17 @@ +.. change:: + :tags: usecase, sql + :tickets: 1390 + + Add support for regular expression on supported backends. + Two operations have been defined: + + * :meth:`_sql.ColumnOperators.regexp_match` implementing a regular + expression match like function. + * :meth:`_sql.ColumnOperators.regexp_replace` implementing a regular + expression string replace function. + + Supported backends include SQLite, PostgreSQL, MySQL / MariaDB, and Oracle. + + .. seealso:: + + :ref:`change_1390` diff --git a/doc/build/conf.py b/doc/build/conf.py index 857bec6495..f281862e04 100644 --- a/doc/build/conf.py +++ b/doc/build/conf.py @@ -101,6 +101,7 @@ autodocmods_convert_modname = { "sqlalchemy.sql.dml": "sqlalchemy.sql.expression", "sqlalchemy.sql.ddl": "sqlalchemy.schema", "sqlalchemy.sql.base": "sqlalchemy.sql.expression", + "sqlalchemy.sql.operators": "sqlalchemy.sql.expression", "sqlalchemy.event.base": "sqlalchemy.event", "sqlalchemy.engine.base": "sqlalchemy.engine", "sqlalchemy.engine.url": "sqlalchemy.engine", diff --git a/doc/build/core/defaults.rst b/doc/build/core/defaults.rst index eaf323b34f..b0bf3d0b99 100644 --- a/doc/build/core/defaults.rst +++ b/doc/build/core/defaults.rst @@ -643,7 +643,7 @@ DDL is emitted to the database. .. _identity_ddl: Identity Columns (GENERATED { ALWAYS | BY DEFAULT } AS IDENTITY) --------------------------------------------------------------- +----------------------------------------------------------------- .. versionadded:: 1.4 diff --git a/doc/build/core/sqlelement.rst b/doc/build/core/sqlelement.rst index 3e2c1d7fbf..7d9f55645f 100644 --- a/doc/build/core/sqlelement.rst +++ b/doc/build/core/sqlelement.rst @@ -30,7 +30,7 @@ is placed in the FROM clause of a SELECT statement. .. autofunction:: cast -.. autofunction:: sqlalchemy.sql.expression.column +.. autofunction:: column .. autofunction:: collate @@ -114,7 +114,7 @@ is placed in the FROM clause of a SELECT statement. :inherited-members: :undoc-members: -.. autoclass:: sqlalchemy.sql.operators.ColumnOperators +.. autoclass:: ColumnOperators :members: :special-members: :inherited-members: @@ -125,7 +125,7 @@ is placed in the FROM clause of a SELECT statement. .. autoclass:: Extract :members: -.. autoclass:: sqlalchemy.sql.elements.False_ +.. autoclass:: False_ :members: .. autoclass:: FunctionFilter @@ -137,7 +137,7 @@ is placed in the FROM clause of a SELECT statement. .. autoclass:: LambdaElement :members: -.. autoclass:: sqlalchemy.sql.elements.Null +.. autoclass:: Null :members: .. autoclass:: Over @@ -158,20 +158,20 @@ is placed in the FROM clause of a SELECT statement. .. autoclass:: sqlalchemy.sql.elements.WrapsColumnExpression :members: -.. autoclass:: sqlalchemy.sql.elements.True_ +.. autoclass:: True_ :members: .. autoclass:: TypeCoerce :members: -.. autoclass:: sqlalchemy.sql.operators.custom_op +.. autoclass:: custom_op :members: -.. autoclass:: sqlalchemy.sql.operators.Operators +.. autoclass:: Operators :members: :special-members: -.. autoclass:: sqlalchemy.sql.elements.quoted_name +.. autoclass:: quoted_name .. attribute:: quote diff --git a/lib/sqlalchemy/dialects/mysql/base.py b/lib/sqlalchemy/dialects/mysql/base.py index 46529636d1..0c9859e79d 100644 --- a/lib/sqlalchemy/dialects/mysql/base.py +++ b/lib/sqlalchemy/dialects/mysql/base.py @@ -1759,6 +1759,62 @@ class MySQLCompiler(compiler.SQLCompiler): self.process(binary.right), ) + def _mariadb_regexp_flags(self, flags, pattern, **kw): + return "CONCAT('(?', %s, ')', %s)" % ( + self.process(flags, **kw), + self.process(pattern, **kw), + ) + + def _regexp_match(self, op_string, binary, operator, **kw): + flags = binary.modifiers["flags"] + if flags is None: + return self._generate_generic_binary(binary, op_string, **kw) + elif self.dialect.is_mariadb: + return "%s%s%s" % ( + self.process(binary.left, **kw), + op_string, + self._mariadb_regexp_flags(flags, binary.right), + ) + else: + text = "REGEXP_LIKE(%s, %s, %s)" % ( + self.process(binary.left, **kw), + self.process(binary.right, **kw), + self.process(flags, **kw), + ) + if op_string == " NOT REGEXP ": + return "NOT %s" % text + else: + return text + + def visit_regexp_match_op_binary(self, binary, operator, **kw): + return self._regexp_match(" REGEXP ", binary, operator, **kw) + + def visit_not_regexp_match_op_binary(self, binary, operator, **kw): + return self._regexp_match(" NOT REGEXP ", binary, operator, **kw) + + def visit_regexp_replace_op_binary(self, binary, operator, **kw): + flags = binary.modifiers["flags"] + replacement = binary.modifiers["replacement"] + if flags is None: + return "REGEXP_REPLACE(%s, %s, %s)" % ( + self.process(binary.left, **kw), + self.process(binary.right, **kw), + self.process(replacement, **kw), + ) + elif self.dialect.is_mariadb: + return "REGEXP_REPLACE(%s, %s, %s)" % ( + self.process(binary.left, **kw), + self._mariadb_regexp_flags(flags, binary.right), + self.process(replacement, **kw), + ) + else: + return "REGEXP_REPLACE(%s, %s, %s, %s)" % ( + self.process(binary.left, **kw), + self.process(binary.right, **kw), + self.process(replacement, **kw), + self.process(flags, **kw), + ) + class MySQLDDLCompiler(compiler.DDLCompiler): def get_column_specification(self, column, **kw): diff --git a/lib/sqlalchemy/dialects/oracle/base.py b/lib/sqlalchemy/dialects/oracle/base.py index 2e5ce25818..65580b4e15 100644 --- a/lib/sqlalchemy/dialects/oracle/base.py +++ b/lib/sqlalchemy/dialects/oracle/base.py @@ -23,7 +23,7 @@ available, which are the use of IDENTITY columns (Oracle 12 and above only) or the association of a SEQUENCE with the column. Specifying GENERATED AS IDENTITY (Oracle 12 and above) -^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ Starting from version 12 Oracle can make use of identity columns using the :class:`_sql.Identity` to specify the autoincrementing behavior:: @@ -53,7 +53,7 @@ setting :paramref:`_schema.Identity.on_null` to ``True`` to specify ON NULL in conjunction with a 'BY DEFAULT' identity column. Using a SEQUENCE (all Oracle versions) -^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ Older version of Oracle had no "autoincrement" feature, SQLAlchemy relies upon sequences to produce these values. With the @@ -1211,6 +1211,43 @@ class OracleCompiler(compiler.SQLCompiler): self.process(binary.right), ) + def _get_regexp_args(self, binary, kw): + string = self.process(binary.left, **kw) + pattern = self.process(binary.right, **kw) + flags = binary.modifiers["flags"] + if flags is not None: + flags = self.process(flags, **kw) + return string, pattern, flags + + def visit_regexp_match_op_binary(self, binary, operator, **kw): + string, pattern, flags = self._get_regexp_args(binary, kw) + if flags is None: + return "REGEXP_LIKE(%s, %s)" % (string, pattern) + else: + return "REGEXP_LIKE(%s, %s, %s)" % (string, pattern, flags) + + def visit_not_regexp_match_op_binary(self, binary, operator, **kw): + return "NOT %s" % self.visit_regexp_match_op_binary( + binary, operator, **kw + ) + + def visit_regexp_replace_op_binary(self, binary, operator, **kw): + string, pattern, flags = self._get_regexp_args(binary, kw) + replacement = self.process(binary.modifiers["replacement"], **kw) + if flags is None: + return "REGEXP_REPLACE(%s, %s, %s)" % ( + string, + pattern, + replacement, + ) + else: + return "REGEXP_REPLACE(%s, %s, %s, %s)" % ( + string, + pattern, + replacement, + flags, + ) + class OracleDDLCompiler(compiler.DDLCompiler): def define_constraint_cascades(self, constraint): diff --git a/lib/sqlalchemy/dialects/postgresql/base.py b/lib/sqlalchemy/dialects/postgresql/base.py index 07405e6d15..8eb116111d 100644 --- a/lib/sqlalchemy/dialects/postgresql/base.py +++ b/lib/sqlalchemy/dialects/postgresql/base.py @@ -1796,6 +1796,53 @@ class PGCompiler(compiler.SQLCompiler): else "" ) + def _regexp_match(self, base_op, binary, operator, kw): + flags = binary.modifiers["flags"] + if flags is None: + return self._generate_generic_binary( + binary, " %s " % base_op, **kw + ) + if isinstance(flags, elements.BindParameter) and flags.value == "i": + return self._generate_generic_binary( + binary, " %s* " % base_op, **kw + ) + flags = self.process(flags, **kw) + string = self.process(binary.left, **kw) + pattern = self.process(binary.right, **kw) + return "%s %s CONCAT('(?', %s, ')', %s)" % ( + string, + base_op, + flags, + pattern, + ) + + def visit_regexp_match_op_binary(self, binary, operator, **kw): + return self._regexp_match("~", binary, operator, kw) + + def visit_not_regexp_match_op_binary(self, binary, operator, **kw): + return self._regexp_match("!~", binary, operator, kw) + + def visit_regexp_replace_op_binary(self, binary, operator, **kw): + string = self.process(binary.left, **kw) + pattern = self.process(binary.right, **kw) + flags = binary.modifiers["flags"] + if flags is not None: + flags = self.process(flags, **kw) + replacement = self.process(binary.modifiers["replacement"], **kw) + if flags is None: + return "REGEXP_REPLACE(%s, %s, %s)" % ( + string, + pattern, + replacement, + ) + else: + return "REGEXP_REPLACE(%s, %s, %s, %s)" % ( + string, + pattern, + replacement, + flags, + ) + def visit_empty_set_expr(self, element_types): # cast the empty set to the type we are comparing against. if # we are comparing against the null type, pick an arbitrary diff --git a/lib/sqlalchemy/dialects/sqlite/base.py b/lib/sqlalchemy/dialects/sqlite/base.py index 3a8ffa23d7..8ef35514ab 100644 --- a/lib/sqlalchemy/dialects/sqlite/base.py +++ b/lib/sqlalchemy/dialects/sqlite/base.py @@ -1077,6 +1077,12 @@ class SQLiteCompiler(compiler.SQLCompiler): ", ".join("1" for type_ in element_types or [INTEGER()]), ) + def visit_regexp_match_op_binary(self, binary, operator, **kw): + return self._generate_generic_binary(binary, " REGEXP ", **kw) + + def visit_not_regexp_match_op_binary(self, binary, operator, **kw): + return self._generate_generic_binary(binary, " NOT REGEXP ", **kw) + class SQLiteDDLCompiler(compiler.DDLCompiler): def get_column_specification(self, column, **kwargs): diff --git a/lib/sqlalchemy/dialects/sqlite/pysqlite.py b/lib/sqlalchemy/dialects/sqlite/pysqlite.py index 3c88dab8ee..a8b2349216 100644 --- a/lib/sqlalchemy/dialects/sqlite/pysqlite.py +++ b/lib/sqlalchemy/dialects/sqlite/pysqlite.py @@ -120,6 +120,46 @@ that creates a Python sqlite3 driver level connection directly. `Uniform Resource Identifiers `_ - in the SQLite documentation +.. _pysqlite_regexp: + +Regular Expression Support +--------------------------- + +.. versionadded:: 1.4 + +Support for the :meth:`_sql.ColumnOperators.regexp_match` operator is provided +using Python's re.match_ function. SQLite itself does not include a working +regular expression operator; instead, it includes a non-implemented placeholder +operator ``REGEXP`` that calls a user-defined function that must be provided. + +SQLAlchemy's implementation makes use of the pysqlite create_function_ hook +as follows:: + + + def regexp(a, b): + return bool(re.match(a, b)) + + sqlite_connection.create_function( + "regexp", 2, regexp, + ) + +There is currently no support for regular expression flags as a separate +argument, as these are not supported by SQLite's REGEXP operator, however these +may be included inline within the regular expression string. See `Python regular expressions`_ for +details. + +.. seealso:: + + `Python regular expressions`_: Documentation for Python's regular expression syntax. + +.. _create_function: https://docs.python.org/3/library/sqlite3.html#sqlite3.Connection.create_function + +.. _re.match: https://docs.python.org/3/library/re.html#re.match + +.. _Python regular expressions: https://docs.python.org/3/library/re.html#re.match + + + Compatibility with sqlite3 "native" date and datetime types ----------------------------------------------------------- @@ -362,6 +402,7 @@ by adding the desired locking mode to our ``"BEGIN"``:: """ # noqa import os +import re from .base import DATE from .base import DATETIME @@ -461,6 +502,37 @@ class SQLiteDialect_pysqlite(SQLiteDialect): connection, level ) + def on_connect(self): + connect = super(SQLiteDialect_pysqlite, self).on_connect() + + def regexp(a, b): + return bool(re.match(a, b)) + + def set_regexp(connection): + if hasattr(connection, "connection"): + dbapi_connection = connection.connection + else: + dbapi_connection = connection + + dbapi_connection.create_function( + "regexp", 2, regexp, + ) + + fns = [set_regexp] + + if self.isolation_level is not None: + + def iso_level(conn): + self.set_isolation_level(conn, self.isolation_level) + + fns.append(iso_level) + + def connect(conn): + for fn in fns: + fn(conn) + + return connect + def create_connect_args(self, url): if url.username or url.password or url.host or url.port: raise exc.ArgumentError( diff --git a/lib/sqlalchemy/sql/compiler.py b/lib/sqlalchemy/sql/compiler.py index 7b917e6612..ec1a579355 100644 --- a/lib/sqlalchemy/sql/compiler.py +++ b/lib/sqlalchemy/sql/compiler.py @@ -2109,6 +2109,24 @@ class SQLCompiler(Compiled): **kw ) + def visit_regexp_match_op_binary(self, binary, operator, **kw): + raise exc.CompileError( + "%s dialect does not support regular expressions" + % self.dialect.name + ) + + def visit_not_regexp_match_op_binary(self, binary, operator, **kw): + raise exc.CompileError( + "%s dialect does not support regular expressions" + % self.dialect.name + ) + + def visit_regexp_replace_op_binary(self, binary, operator, **kw): + raise exc.CompileError( + "%s dialect does not support regular expression replacements" + % self.dialect.name + ) + def visit_bindparam( self, bindparam, @@ -3671,6 +3689,20 @@ class StrSQLCompiler(SQLCompiler): def get_from_hint_text(self, table, text): return "[%s]" % text + def visit_regexp_match_op_binary(self, binary, operator, **kw): + return self._generate_generic_binary(binary, " ", **kw) + + def visit_not_regexp_match_op_binary(self, binary, operator, **kw): + return self._generate_generic_binary(binary, " ", **kw) + + def visit_regexp_replace_op_binary(self, binary, operator, **kw): + replacement = binary.modifiers["replacement"] + return "(%s, %s, %s)" % ( + binary.left._compiler_dispatch(self, **kw), + binary.right._compiler_dispatch(self, **kw), + replacement._compiler_dispatch(self, **kw), + ) + class DDLCompiler(Compiled): @util.memoized_property diff --git a/lib/sqlalchemy/sql/default_comparator.py b/lib/sqlalchemy/sql/default_comparator.py index 6f1a256705..eec174e8b5 100644 --- a/lib/sqlalchemy/sql/default_comparator.py +++ b/lib/sqlalchemy/sql/default_comparator.py @@ -252,6 +252,45 @@ def _collate_impl(expr, op, other, **kw): return collate(expr, other) +def _regexp_match_impl(expr, op, pattern, flags, **kw): + if flags is not None: + flags = coercions.expect( + roles.BinaryElementRole, + flags, + expr=expr, + operator=operators.regexp_replace_op, + ) + return _boolean_compare( + expr, + op, + pattern, + flags=flags, + negate=operators.not_regexp_match_op + if op is operators.regexp_match_op + else operators.regexp_match_op, + **kw + ) + + +def _regexp_replace_impl(expr, op, pattern, replacement, flags, **kw): + replacement = coercions.expect( + roles.BinaryElementRole, + replacement, + expr=expr, + operator=operators.regexp_replace_op, + ) + if flags is not None: + flags = coercions.expect( + roles.BinaryElementRole, + flags, + expr=expr, + operator=operators.regexp_replace_op, + ) + return _binary_operate( + expr, op, pattern, replacement=replacement, flags=flags, **kw + ) + + # a mapping of operators with the method they use, along with # their negated operator for comparison operators operator_lookup = { @@ -304,4 +343,7 @@ operator_lookup = { "lshift": (_unsupported_impl,), "rshift": (_unsupported_impl,), "contains": (_unsupported_impl,), + "regexp_match_op": (_regexp_match_impl,), + "not_regexp_match_op": (_regexp_match_impl,), + "regexp_replace_op": (_regexp_replace_impl,), } diff --git a/lib/sqlalchemy/sql/expression.py b/lib/sqlalchemy/sql/expression.py index d60c633631..31584f0729 100644 --- a/lib/sqlalchemy/sql/expression.py +++ b/lib/sqlalchemy/sql/expression.py @@ -44,6 +44,7 @@ __all__ = [ "case", "cast", "column", + "custom_op", "cte", "delete", "desc", @@ -81,6 +82,7 @@ __all__ = [ "union", "union_all", "update", + "quoted_name", "within_group", "Subquery", "TableSample", @@ -141,6 +143,9 @@ from .functions import modifier # noqa from .lambdas import lambda_stmt # noqa from .lambdas import LambdaElement # noqa from .lambdas import StatementLambdaElement # noqa +from .operators import ColumnOperators # noqa +from .operators import custom_op # noqa +from .operators import Operators # noqa from .selectable import Alias # noqa from .selectable import AliasedReturnsRows # noqa from .selectable import CompoundSelect # noqa diff --git a/lib/sqlalchemy/sql/operators.py b/lib/sqlalchemy/sql/operators.py index bb773e2812..91a0792c3a 100644 --- a/lib/sqlalchemy/sql/operators.py +++ b/lib/sqlalchemy/sql/operators.py @@ -913,7 +913,7 @@ class ColumnOperators(Operators): def match(self, other, **kwargs): """Implements a database-specific 'match' operator. - :meth:`~.ColumnOperators.match` attempts to resolve to + :meth:`_sql.ColumnOperators.match` attempts to resolve to a MATCH-like function or operator provided by the backend. Examples include: @@ -928,6 +928,96 @@ class ColumnOperators(Operators): """ return self.operate(match_op, other, **kwargs) + def regexp_match(self, pattern, flags=None): + """Implements a database-specific 'regexp match' operator. + + E.g.:: + + stmt = select(table.c.some_column).where( + table.c.some_column.regexp_match('^(b|c)') + ) + + :meth:`_sql.ColumnOperators.regexp_match` attempts to resolve to + a REGEXP-like function or operator provided by the backend, however + the specific regular expression syntax and flags available are + **not backend agnostic**. + + Examples include: + + * PostgreSQL - renders ``x ~ y`` or ``x !~ y`` when negated. + * Oracle - renders ``REGEXP_LIKE(x, y)`` + * SQLite - uses SQLite's ``REGEXP`` placeholder operator and calls into + the Python ``re.match()`` builtin. + * other backends may provide special implementations. + * Backends without any special implementation will emit + the operator as "REGEXP" or "NOT REGEXP". This is compatible with + SQLite and MySQL, for example. + + Regular expression support is currently implemented for Oracle, + PostgreSQL, MySQL and MariaDB. Partial support is available for + SQLite. Support among third-party dialects may vary. + + :param pattern: The regular expression pattern string or column + clause. + :param flags: Any regular expression string flags to apply. Flags + tend to be backend specific. It can be a string or a column clause. + Some backends, like PostgreSQL and MariaDB, may alternatively + specify the flags as part of the pattern. + When using the ignore case flag 'i' in PostgreSQL, the ignore case + regexp match operator ``~*`` or ``!~*`` will be used. + + .. versionadded:: 1.4 + + .. seealso:: + + :meth:`_sql.ColumnOperators.regexp_replace` + + + """ + return self.operate(regexp_match_op, pattern, flags=flags) + + def regexp_replace(self, pattern, replacement, flags=None): + """Implements a database-specific 'regexp replace' operator. + + E.g.:: + + stmt = select( + table.c.some_column.regexp_replace( + 'b(..)', + 'X\1Y', + flags='g' + ) + ) + + :meth:`_sql.ColumnOperators.regexp_replace` attempts to resolve to + a REGEXP_REPLACE-like function provided by the backend, that + usually emit the function ``REGEXP_REPLACE()``. However, + the specific regular expression syntax and flags available are + **not backend agnostic**. + + Regular expression replacement support is currently implemented for + Oracle, PostgreSQL, MySQL 8 or greater and MariaDB. Support among + third-party dialects may vary. + + :param pattern: The regular expression pattern string or column + clause. + :param pattern: The replacement string or column clause. + :param flags: Any regular expression string flags to apply. Flags + tend to be backend specific. It can be a string or a column clause. + Some backends, like PostgreSQL and MariaDB, may alternatively + specify the flags as part of the pattern. + + .. versionadded:: 1.4 + + .. seealso:: + + :meth:`_sql.ColumnOperators.regexp_match` + + """ + return self.operate( + regexp_replace_op, pattern, replacement=replacement, flags=flags + ) + def desc(self): """Produce a :func:`_expression.desc` clause against the parent object.""" @@ -1298,6 +1388,20 @@ def match_op(a, b, **kw): return a.match(b, **kw) +@comparison_op +def regexp_match_op(a, b, flags=None): + return a.regexp_match(b, flags=flags) + + +@comparison_op +def not_regexp_match_op(a, b, flags=None): + return ~a.regexp_match(b, flags=flags) + + +def regexp_replace_op(a, b, replacement, flags=None): + return a.regexp_replace(b, replacement=replacement, flags=flags) + + @comparison_op def notmatch_op(a, b, **kw): return a.notmatch(b, **kw) @@ -1417,6 +1521,9 @@ _PRECEDENCE = { filter_op: 6, match_op: 5, notmatch_op: 5, + regexp_match_op: 5, + not_regexp_match_op: 5, + regexp_replace_op: 5, ilike_op: 5, notilike_op: 5, like_op: 5, diff --git a/lib/sqlalchemy/testing/requirements.py b/lib/sqlalchemy/testing/requirements.py index 7b0ddafe30..9b8caac2ef 100644 --- a/lib/sqlalchemy/testing/requirements.py +++ b/lib/sqlalchemy/testing/requirements.py @@ -1280,3 +1280,22 @@ class SuiteRequirements(Requirements): This is mainly to exclude MSSql. """ return exclusions.closed() + + @property + def regexp_match(self): + """backend supports the regexp_match operator. + + .. versionadded:: 1.4 + + """ + return exclusions.closed() + + @property + def regexp_replace(self): + """backend supports the regexp_replace operator. + + .. versionadded:: 1.4 + + + """ + return exclusions.closed() diff --git a/lib/sqlalchemy/testing/suite/test_select.py b/lib/sqlalchemy/testing/suite/test_select.py index 675fac609e..7e03371465 100644 --- a/lib/sqlalchemy/testing/suite/test_select.py +++ b/lib/sqlalchemy/testing/suite/test_select.py @@ -1004,6 +1004,23 @@ class LikeFunctionsTest(fixtures.TablesTest): self._test(col.contains("b%cd", autoescape=True, escape="#"), {3}) self._test(col.contains("b#cd", autoescape=True, escape="#"), {7}) + @testing.requires.regexp_match + def test_regexp_match(self): + col = self.tables.some_table.c.data + self._test(col.regexp_match("a.cde"), {1, 5, 6, 9}) + + @testing.requires.regexp_match + def test_not_regexp_match(self): + col = self.tables.some_table.c.data + self._test(~col.regexp_match("a.cde"), {2, 3, 4, 7, 8, 10}) + + @testing.requires.regexp_replace + def test_regexp_replace(self): + col = self.tables.some_table.c.data + self._test( + col.regexp_replace("a.cde", "FOO").contains("FOO"), {1, 5, 6, 9} + ) + class ComputedColumnTest(fixtures.TablesTest): __backend__ = True diff --git a/test/dialect/mysql/test_compiler.py b/test/dialect/mysql/test_compiler.py index aca1db33ce..cca6a27ded 100644 --- a/test/dialect/mysql/test_compiler.py +++ b/test/dialect/mysql/test_compiler.py @@ -960,3 +960,134 @@ class InsertOnDuplicateTest(fixtures.TestBase, AssertsCompiledSQL): "baz_1": "some literal", }, ) + + +class RegexpCommon(testing.AssertsCompiledSQL): + def setUp(self): + self.table = table( + "mytable", column("myid", Integer), column("name", String) + ) + + def test_regexp_match(self): + self.assert_compile( + self.table.c.myid.regexp_match("pattern"), + "mytable.myid REGEXP %s", + checkpositional=("pattern",), + ) + + def test_regexp_match_column(self): + self.assert_compile( + self.table.c.myid.regexp_match(self.table.c.name), + "mytable.myid REGEXP mytable.name", + checkpositional=(), + ) + + def test_regexp_match_str(self): + self.assert_compile( + literal("string").regexp_match(self.table.c.name), + "%s REGEXP mytable.name", + checkpositional=("string",), + ) + + def test_not_regexp_match(self): + self.assert_compile( + ~self.table.c.myid.regexp_match("pattern"), + "mytable.myid NOT REGEXP %s", + checkpositional=("pattern",), + ) + + def test_not_regexp_match_column(self): + self.assert_compile( + ~self.table.c.myid.regexp_match(self.table.c.name), + "mytable.myid NOT REGEXP mytable.name", + checkpositional=(), + ) + + def test_not_regexp_match_str(self): + self.assert_compile( + ~literal("string").regexp_match(self.table.c.name), + "%s NOT REGEXP mytable.name", + checkpositional=("string",), + ) + + def test_regexp_replace(self): + self.assert_compile( + self.table.c.myid.regexp_replace("pattern", "replacement"), + "REGEXP_REPLACE(mytable.myid, %s, %s)", + checkpositional=("pattern", "replacement"), + ) + + def test_regexp_replace_column(self): + self.assert_compile( + self.table.c.myid.regexp_replace("pattern", self.table.c.name), + "REGEXP_REPLACE(mytable.myid, %s, mytable.name)", + checkpositional=("pattern",), + ) + + def test_regexp_replace_column2(self): + self.assert_compile( + self.table.c.myid.regexp_replace(self.table.c.name, "replacement"), + "REGEXP_REPLACE(mytable.myid, mytable.name, %s)", + checkpositional=("replacement",), + ) + + def test_regexp_replace_string(self): + self.assert_compile( + literal("string").regexp_replace("pattern", self.table.c.name), + "REGEXP_REPLACE(%s, %s, mytable.name)", + checkpositional=("string", "pattern"), + ) + + +class RegexpTestMySql(fixtures.TestBase, RegexpCommon): + __dialect__ = "mysql" + + def test_regexp_match_flags(self): + self.assert_compile( + self.table.c.myid.regexp_match("pattern", flags="ig"), + "REGEXP_LIKE(mytable.myid, %s, %s)", + checkpositional=("pattern", "ig"), + ) + + def test_not_regexp_match_flags(self): + self.assert_compile( + ~self.table.c.myid.regexp_match("pattern", flags="ig"), + "NOT REGEXP_LIKE(mytable.myid, %s, %s)", + checkpositional=("pattern", "ig"), + ) + + def test_regexp_replace_flags(self): + self.assert_compile( + self.table.c.myid.regexp_replace( + "pattern", "replacement", flags="ig" + ), + "REGEXP_REPLACE(mytable.myid, %s, %s, %s)", + checkpositional=("pattern", "replacement", "ig"), + ) + + +class RegexpTestMariaDb(fixtures.TestBase, RegexpCommon): + __dialect__ = "mariadb" + + def test_regexp_match_flags(self): + self.assert_compile( + self.table.c.myid.regexp_match("pattern", flags="ig"), + "mytable.myid REGEXP CONCAT('(?', %s, ')', %s)", + checkpositional=("ig", "pattern"), + ) + + def test_not_regexp_match_flags(self): + self.assert_compile( + ~self.table.c.myid.regexp_match("pattern", flags="ig"), + "mytable.myid NOT REGEXP CONCAT('(?', %s, ')', %s)", + checkpositional=("ig", "pattern"), + ) + + def test_regexp_replace_flags(self): + self.assert_compile( + self.table.c.myid.regexp_replace( + "pattern", "replacement", flags="ig" + ), + "REGEXP_REPLACE(mytable.myid, CONCAT('(?', %s, ')', %s), %s)", + checkpositional=("ig", "pattern", "replacement"), + ) diff --git a/test/dialect/oracle/test_compiler.py b/test/dialect/oracle/test_compiler.py index 97a2046307..a4a8cd99f7 100644 --- a/test/dialect/oracle/test_compiler.py +++ b/test/dialect/oracle/test_compiler.py @@ -1307,3 +1307,134 @@ class SequenceTest(fixtures.TestBase, AssertsCompiledSQL): dialect.identifier_preparer.format_sequence(seq) == '"Some_Schema"."My_Seq"' ) + + +class RegexpTest(fixtures.TestBase, testing.AssertsCompiledSQL): + __dialect__ = "oracle" + + def setUp(self): + self.table = table( + "mytable", column("myid", Integer), column("name", String) + ) + + def test_regexp_match(self): + self.assert_compile( + self.table.c.myid.regexp_match("pattern"), + "REGEXP_LIKE(mytable.myid, :myid_1)", + checkparams={"myid_1": "pattern"}, + ) + + def test_regexp_match_column(self): + self.assert_compile( + self.table.c.myid.regexp_match(self.table.c.name), + "REGEXP_LIKE(mytable.myid, mytable.name)", + checkparams={}, + ) + + def test_regexp_match_str(self): + self.assert_compile( + literal("string").regexp_match(self.table.c.name), + "REGEXP_LIKE(:param_1, mytable.name)", + checkparams={"param_1": "string"}, + ) + + def test_regexp_match_flags(self): + self.assert_compile( + self.table.c.myid.regexp_match("pattern", flags="ig"), + "REGEXP_LIKE(mytable.myid, :myid_1, :myid_2)", + checkparams={"myid_1": "pattern", "myid_2": "ig"}, + ) + + def test_regexp_match_flags_col(self): + self.assert_compile( + self.table.c.myid.regexp_match("pattern", flags=self.table.c.name), + "REGEXP_LIKE(mytable.myid, :myid_1, mytable.name)", + checkparams={"myid_1": "pattern"}, + ) + + def test_not_regexp_match(self): + self.assert_compile( + ~self.table.c.myid.regexp_match("pattern"), + "NOT REGEXP_LIKE(mytable.myid, :myid_1)", + checkparams={"myid_1": "pattern"}, + ) + + def test_not_regexp_match_column(self): + self.assert_compile( + ~self.table.c.myid.regexp_match(self.table.c.name), + "NOT REGEXP_LIKE(mytable.myid, mytable.name)", + checkparams={}, + ) + + def test_not_regexp_match_str(self): + self.assert_compile( + ~literal("string").regexp_match(self.table.c.name), + "NOT REGEXP_LIKE(:param_1, mytable.name)", + checkparams={"param_1": "string"}, + ) + + def test_not_regexp_match_flags_col(self): + self.assert_compile( + ~self.table.c.myid.regexp_match( + "pattern", flags=self.table.c.name + ), + "NOT REGEXP_LIKE(mytable.myid, :myid_1, mytable.name)", + checkparams={"myid_1": "pattern"}, + ) + + def test_not_regexp_match_flags(self): + self.assert_compile( + ~self.table.c.myid.regexp_match("pattern", flags="ig"), + "NOT REGEXP_LIKE(mytable.myid, :myid_1, :myid_2)", + checkparams={"myid_1": "pattern", "myid_2": "ig"}, + ) + + def test_regexp_replace(self): + self.assert_compile( + self.table.c.myid.regexp_replace("pattern", "replacement"), + "REGEXP_REPLACE(mytable.myid, :myid_1, :myid_2)", + checkparams={"myid_1": "pattern", "myid_2": "replacement"}, + ) + + def test_regexp_replace_column(self): + self.assert_compile( + self.table.c.myid.regexp_replace("pattern", self.table.c.name), + "REGEXP_REPLACE(mytable.myid, :myid_1, mytable.name)", + checkparams={"myid_1": "pattern"}, + ) + + def test_regexp_replace_column2(self): + self.assert_compile( + self.table.c.myid.regexp_replace(self.table.c.name, "replacement"), + "REGEXP_REPLACE(mytable.myid, mytable.name, :myid_1)", + checkparams={"myid_1": "replacement"}, + ) + + def test_regexp_replace_string(self): + self.assert_compile( + literal("string").regexp_replace("pattern", self.table.c.name), + "REGEXP_REPLACE(:param_1, :param_2, mytable.name)", + checkparams={"param_2": "pattern", "param_1": "string"}, + ) + + def test_regexp_replace_flags(self): + self.assert_compile( + self.table.c.myid.regexp_replace( + "pattern", "replacement", flags="ig" + ), + "REGEXP_REPLACE(mytable.myid, :myid_1, :myid_3, :myid_2)", + checkparams={ + "myid_1": "pattern", + "myid_3": "replacement", + "myid_2": "ig", + }, + ) + + def test_regexp_replace_flags_col(self): + self.assert_compile( + self.table.c.myid.regexp_replace( + "pattern", "replacement", flags=self.table.c.name + ), + "REGEXP_REPLACE(mytable.myid, :myid_1, :myid_2, mytable.name)", + checkparams={"myid_1": "pattern", "myid_2": "replacement"}, + ) diff --git a/test/dialect/postgresql/test_compiler.py b/test/dialect/postgresql/test_compiler.py index 6196b52f28..556601fc62 100644 --- a/test/dialect/postgresql/test_compiler.py +++ b/test/dialect/postgresql/test_compiler.py @@ -12,6 +12,7 @@ from sqlalchemy import func from sqlalchemy import Identity from sqlalchemy import Index from sqlalchemy import Integer +from sqlalchemy import literal from sqlalchemy import MetaData from sqlalchemy import null from sqlalchemy import schema @@ -2346,3 +2347,149 @@ class FullTextSearchTest(fixtures.TestBase, AssertsCompiledSQL): "WHERE to_tsvector(%(to_tsvector_1)s, mytable.title) @@ " """to_tsquery('english', %(to_tsvector_2)s)""", ) + + +class RegexpTest(fixtures.TestBase, testing.AssertsCompiledSQL): + __dialect__ = "postgresql" + + def setUp(self): + self.table = table( + "mytable", column("myid", Integer), column("name", String) + ) + + def test_regexp_match(self): + self.assert_compile( + self.table.c.myid.regexp_match("pattern"), + "mytable.myid ~ %(myid_1)s", + checkparams={"myid_1": "pattern"}, + ) + + def test_regexp_match_column(self): + self.assert_compile( + self.table.c.myid.regexp_match(self.table.c.name), + "mytable.myid ~ mytable.name", + checkparams={}, + ) + + def test_regexp_match_str(self): + self.assert_compile( + literal("string").regexp_match(self.table.c.name), + "%(param_1)s ~ mytable.name", + checkparams={"param_1": "string"}, + ) + + def test_regexp_match_flags(self): + self.assert_compile( + self.table.c.myid.regexp_match("pattern", flags="ig"), + "mytable.myid ~ CONCAT('(?', %(myid_1)s, ')', %(myid_2)s)", + checkparams={"myid_2": "pattern", "myid_1": "ig"}, + ) + + def test_regexp_match_flags_ignorecase(self): + self.assert_compile( + self.table.c.myid.regexp_match("pattern", flags="i"), + "mytable.myid ~* %(myid_1)s", + checkparams={"myid_1": "pattern"}, + ) + + def test_regexp_match_flags_col(self): + self.assert_compile( + self.table.c.myid.regexp_match("pattern", flags=self.table.c.name), + "mytable.myid ~ CONCAT('(?', mytable.name, ')', %(myid_1)s)", + checkparams={"myid_1": "pattern"}, + ) + + def test_not_regexp_match(self): + self.assert_compile( + ~self.table.c.myid.regexp_match("pattern"), + "mytable.myid !~ %(myid_1)s", + checkparams={"myid_1": "pattern"}, + ) + + def test_not_regexp_match_column(self): + self.assert_compile( + ~self.table.c.myid.regexp_match(self.table.c.name), + "mytable.myid !~ mytable.name", + checkparams={}, + ) + + def test_not_regexp_match_str(self): + self.assert_compile( + ~literal("string").regexp_match(self.table.c.name), + "%(param_1)s !~ mytable.name", + checkparams={"param_1": "string"}, + ) + + def test_not_regexp_match_flags(self): + self.assert_compile( + ~self.table.c.myid.regexp_match("pattern", flags="ig"), + "mytable.myid !~ CONCAT('(?', %(myid_1)s, ')', %(myid_2)s)", + checkparams={"myid_2": "pattern", "myid_1": "ig"}, + ) + + def test_not_regexp_match_flags_ignorecase(self): + self.assert_compile( + ~self.table.c.myid.regexp_match("pattern", flags="i"), + "mytable.myid !~* %(myid_1)s", + checkparams={"myid_1": "pattern"}, + ) + + def test_not_regexp_match_flags_col(self): + self.assert_compile( + ~self.table.c.myid.regexp_match( + "pattern", flags=self.table.c.name + ), + "mytable.myid !~ CONCAT('(?', mytable.name, ')', %(myid_1)s)", + checkparams={"myid_1": "pattern"}, + ) + + def test_regexp_replace(self): + self.assert_compile( + self.table.c.myid.regexp_replace("pattern", "replacement"), + "REGEXP_REPLACE(mytable.myid, %(myid_1)s, %(myid_2)s)", + checkparams={"myid_1": "pattern", "myid_2": "replacement"}, + ) + + def test_regexp_replace_column(self): + self.assert_compile( + self.table.c.myid.regexp_replace("pattern", self.table.c.name), + "REGEXP_REPLACE(mytable.myid, %(myid_1)s, mytable.name)", + checkparams={"myid_1": "pattern"}, + ) + + def test_regexp_replace_column2(self): + self.assert_compile( + self.table.c.myid.regexp_replace(self.table.c.name, "replacement"), + "REGEXP_REPLACE(mytable.myid, mytable.name, %(myid_1)s)", + checkparams={"myid_1": "replacement"}, + ) + + def test_regexp_replace_string(self): + self.assert_compile( + literal("string").regexp_replace("pattern", self.table.c.name), + "REGEXP_REPLACE(%(param_1)s, %(param_2)s, mytable.name)", + checkparams={"param_2": "pattern", "param_1": "string"}, + ) + + def test_regexp_replace_flags(self): + self.assert_compile( + self.table.c.myid.regexp_replace( + "pattern", "replacement", flags="ig" + ), + "REGEXP_REPLACE(mytable.myid, %(myid_1)s, %(myid_3)s, %(myid_2)s)", + checkparams={ + "myid_1": "pattern", + "myid_3": "replacement", + "myid_2": "ig", + }, + ) + + def test_regexp_replace_flags_col(self): + self.assert_compile( + self.table.c.myid.regexp_replace( + "pattern", "replacement", flags=self.table.c.name + ), + "REGEXP_REPLACE(mytable.myid, %(myid_1)s," + " %(myid_2)s, mytable.name)", + checkparams={"myid_1": "pattern", "myid_2": "replacement"}, + ) diff --git a/test/dialect/test_sqlite.py b/test/dialect/test_sqlite.py index b4813926f8..cb418e99d6 100644 --- a/test/dialect/test_sqlite.py +++ b/test/dialect/test_sqlite.py @@ -20,6 +20,7 @@ from sqlalchemy import ForeignKey from sqlalchemy import func from sqlalchemy import Index from sqlalchemy import inspect +from sqlalchemy import literal from sqlalchemy import MetaData from sqlalchemy import pool from sqlalchemy import PrimaryKeyConstraint @@ -27,6 +28,7 @@ from sqlalchemy import schema from sqlalchemy import select from sqlalchemy import sql from sqlalchemy import Table +from sqlalchemy import table from sqlalchemy import testing from sqlalchemy import text from sqlalchemy import tuple_ @@ -2608,3 +2610,76 @@ class TypeReflectionTest(fixtures.TestBase): def test_round_trip_direct_type_affinity(self): self._test_round_trip(self._type_affinity_fixture()) + + +class RegexpTest(fixtures.TestBase, testing.AssertsCompiledSQL): + __dialect__ = "sqlite" + + def setUp(self): + self.table = table( + "mytable", column("myid", Integer), column("name", String) + ) + + def test_regexp_match(self): + self.assert_compile( + self.table.c.myid.regexp_match("pattern"), + "mytable.myid REGEXP ?", + checkpositional=("pattern",), + ) + + def test_regexp_match_column(self): + self.assert_compile( + self.table.c.myid.regexp_match(self.table.c.name), + "mytable.myid REGEXP mytable.name", + checkparams={}, + ) + + def test_regexp_match_str(self): + self.assert_compile( + literal("string").regexp_match(self.table.c.name), + "? REGEXP mytable.name", + checkpositional=("string",), + ) + + def test_regexp_match_flags(self): + self.assert_compile( + self.table.c.myid.regexp_match("pattern", flags="ig"), + "mytable.myid REGEXP ?", + checkpositional=("pattern",), + ) + + def test_not_regexp_match(self): + self.assert_compile( + ~self.table.c.myid.regexp_match("pattern"), + "mytable.myid NOT REGEXP ?", + checkpositional=("pattern",), + ) + + def test_not_regexp_match_flags(self): + self.assert_compile( + ~self.table.c.myid.regexp_match("pattern", flags="ig"), + "mytable.myid NOT REGEXP ?", + checkpositional=("pattern",), + ) + + def test_not_regexp_match_column(self): + self.assert_compile( + ~self.table.c.myid.regexp_match(self.table.c.name), + "mytable.myid NOT REGEXP mytable.name", + checkparams={}, + ) + + def test_not_regexp_match_str(self): + self.assert_compile( + ~literal("string").regexp_match(self.table.c.name), + "? NOT REGEXP mytable.name", + checkpositional=("string",), + ) + + def test_regexp_replace(self): + assert_raises_message( + exc.CompileError, + "sqlite dialect does not support regular expression replacements", + self.table.c.myid.regexp_replace("pattern", "rep").compile, + dialect=sqlite.dialect(), + ) diff --git a/test/requirements.py b/test/requirements.py index c2fd65b05d..6d4b3dc90c 100644 --- a/test/requirements.py +++ b/test/requirements.py @@ -1650,6 +1650,14 @@ class DefaultRequirements(SuiteRequirements): def computed_columns_reflect_persisted(self): return self.computed_columns + skip_if("oracle") + @property + def regexp_match(self): + return only_on(["postgresql", "mysql", "mariadb", "oracle", "sqlite"]) + + @property + def regexp_replace(self): + return only_on(["postgresql", "mysql>=8", "mariadb", "oracle"]) + @property def supports_distinct_on(self): """If a backend supports the DISTINCT ON in a select""" diff --git a/test/sql/test_operators.py b/test/sql/test_operators.py index fcf40ebbd2..2e5ce8daa6 100644 --- a/test/sql/test_operators.py +++ b/test/sql/test_operators.py @@ -2365,6 +2365,181 @@ class MatchTest(fixtures.TestBase, testing.AssertsCompiledSQL): ) +class RegexpTest(fixtures.TestBase, testing.AssertsCompiledSQL): + __dialect__ = "default" + + def setUp(self): + self.table = table( + "mytable", column("myid", Integer), column("name", String) + ) + + def test_regexp_match(self): + assert_raises_message( + exc.CompileError, + "default dialect does not support regular expressions", + self.table.c.myid.regexp_match("pattern").compile, + dialect=default.DefaultDialect(), + ) + + def test_not_regexp_match(self): + assert_raises_message( + exc.CompileError, + "default dialect does not support regular expressions", + (~self.table.c.myid.regexp_match("pattern")).compile, + dialect=default.DefaultDialect(), + ) + + def test_regexp_replace(self): + assert_raises_message( + exc.CompileError, + "default dialect does not support regular expression replacements", + self.table.c.myid.regexp_replace("pattern", "rep").compile, + dialect=default.DefaultDialect(), + ) + + +class RegexpTestStrCompiler(fixtures.TestBase, testing.AssertsCompiledSQL): + __dialect__ = "default_enhanced" + + def setUp(self): + self.table = table( + "mytable", column("myid", Integer), column("name", String) + ) + + def test_regexp_match(self): + self.assert_compile( + self.table.c.myid.regexp_match("pattern"), + "mytable.myid :myid_1", + checkparams={"myid_1": "pattern"}, + ) + + def test_regexp_match_column(self): + self.assert_compile( + self.table.c.myid.regexp_match(self.table.c.name), + "mytable.myid mytable.name", + checkparams={}, + ) + + def test_regexp_match_str(self): + self.assert_compile( + literal("string").regexp_match(self.table.c.name), + ":param_1 mytable.name", + checkparams={"param_1": "string"}, + ) + + def test_regexp_match_flags(self): + self.assert_compile( + self.table.c.myid.regexp_match("pattern", flags="ig"), + "mytable.myid :myid_1", + checkparams={"myid_1": "pattern"}, + ) + + def test_not_regexp_match(self): + self.assert_compile( + ~self.table.c.myid.regexp_match("pattern"), + "mytable.myid :myid_1", + checkparams={"myid_1": "pattern"}, + ) + + def test_not_regexp_match_column(self): + self.assert_compile( + ~self.table.c.myid.regexp_match(self.table.c.name), + "mytable.myid mytable.name", + checkparams={}, + ) + + def test_not_regexp_match_str(self): + self.assert_compile( + ~literal("string").regexp_match(self.table.c.name), + ":param_1 mytable.name", + checkparams={"param_1": "string"}, + ) + + def test_not_regexp_match_flags(self): + self.assert_compile( + ~self.table.c.myid.regexp_match("pattern", flags="ig"), + "mytable.myid :myid_1", + checkparams={"myid_1": "pattern"}, + ) + + def test_regexp_replace(self): + self.assert_compile( + self.table.c.myid.regexp_replace("pattern", "replacement"), + "(mytable.myid, :myid_1, :myid_2)", + checkparams={"myid_1": "pattern", "myid_2": "replacement"}, + ) + + def test_regexp_replace_column(self): + self.assert_compile( + self.table.c.myid.regexp_replace("pattern", self.table.c.name), + "(mytable.myid, :myid_1, mytable.name)", + checkparams={"myid_1": "pattern"}, + ) + + def test_regexp_replace_column2(self): + self.assert_compile( + self.table.c.myid.regexp_replace(self.table.c.name, "replacement"), + "(mytable.myid, mytable.name, :myid_1)", + checkparams={"myid_1": "replacement"}, + ) + + def test_regexp_replace_string(self): + self.assert_compile( + literal("string").regexp_replace("pattern", self.table.c.name), + "(:param_1, :param_2, mytable.name)", + checkparams={"param_2": "pattern", "param_1": "string"}, + ) + + def test_regexp_replace_flags(self): + self.assert_compile( + self.table.c.myid.regexp_replace( + "pattern", "replacement", flags="ig" + ), + "(mytable.myid, :myid_1, :myid_2)", + checkparams={"myid_1": "pattern", "myid_2": "replacement"}, + ) + + def test_regexp_precedence_1(self): + self.assert_compile( + and_( + self.table.c.myid.match("foo"), + self.table.c.myid.regexp_match("xx"), + ), + "mytable.myid MATCH :myid_1 AND " "mytable.myid :myid_2", + ) + self.assert_compile( + and_( + self.table.c.myid.match("foo"), + ~self.table.c.myid.regexp_match("xx"), + ), + "mytable.myid MATCH :myid_1 AND " + "mytable.myid :myid_2", + ) + self.assert_compile( + and_( + self.table.c.myid.match("foo"), + self.table.c.myid.regexp_replace("xx", "yy"), + ), + "mytable.myid MATCH :myid_1 AND " + "(mytable.myid, :myid_2, :myid_3)", + ) + + def test_regexp_precedence_2(self): + self.assert_compile( + self.table.c.myid + self.table.c.myid.regexp_match("xx"), + "mytable.myid + (mytable.myid :myid_1)", + ) + self.assert_compile( + self.table.c.myid + ~self.table.c.myid.regexp_match("xx"), + "mytable.myid + (mytable.myid :myid_1)", + ) + self.assert_compile( + self.table.c.myid + self.table.c.myid.regexp_replace("xx", "yy"), + "mytable.myid + (" + "(mytable.myid, :myid_1, :myid_2))", + ) + + class ComposedLikeOperatorsTest(fixtures.TestBase, testing.AssertsCompiledSQL): __dialect__ = "default"