Changed how the positional compilation is performed. It's rendered by the compiler
the same as the pyformat compilation. The string is then processed to replace
the placeholders with the correct ones, and to obtain the correct order of the
parameters.
This vastly simplifies the computation of the order of the parameters, that in
case of nested CTE is very hard to compute correctly.
Reworked how numeric paramstyle behavers:
- added support for repeated parameter, without duplicating them like in normal
positional dialects
- implement insertmany support. This requires that the dialect supports out of
order placehoders, since all parameters that are not part of the VALUES clauses
are placed at the beginning of the parameter tuple
- support for different identifiers for a numeric parameter. It's for example
possible to use postgresql style placeholder $1, $2, etc
Added two new dialect based on sqlite to test "numeric" fully using
both :1 style and $1 style. Includes a workaround for SQLite's
not-really-correct numeric implementation.
Changed parmstyle of asyncpg dialect to use numeric, rendering with its native
$ identifiers
Fixes: #8926
Fixes: #8849
Change-Id: I7c640467d49adfe6d795cc84296fc7403dcad4d6
--- /dev/null
+.. change::
+ :tags: bug, sql
+ :tickets: 8849
+
+ Reworked how numeric paramstyle behavers, in particular, fixed insertmany
+ behaviour that prior to this was non functional; added support for repeated
+ parameter without duplicating them like in other positional dialects;
+ introduced new numeric paramstyle called ``numeric_dollar`` that can be
+ used to render statements that use the PostgreSQL placeholder style (
+ i.e. ``$1, $2, $3``).
+ This change requires that the dialect supports out of order placehoders,
+ that may be used used in the statements, in particular when using
+ insert-many values with statement that have parameters in the returning
+ clause.
--- /dev/null
+.. change::
+ :tags: asyncpg
+ :tickets: 8926
+
+ Changed the paramstyle used by asyncpg from ``format`` to
+ ``numeric_dollar``. This has two main benefits since it does not require
+ additional processing of the statement and allows for duplicate parameters
+ to be present in the statements.
def _handle_exception(self, error):
self._adapt_connection._handle_exception(error)
- def _parameter_placeholders(self, params):
- return tuple(f"${idx:d}" for idx, _ in enumerate(params, 1))
-
async def _prepare_and_execute(self, operation, parameters):
adapt_connection = self._adapt_connection
if not adapt_connection._started:
await adapt_connection._start_transaction()
- if parameters is not None:
- operation = operation % self._parameter_placeholders(
- parameters
- )
- else:
+ if parameters is None:
parameters = ()
try:
if not adapt_connection._started:
await adapt_connection._start_transaction()
- operation = operation % self._parameter_placeholders(
- seq_of_parameters[0]
- )
-
try:
return await self._connection.executemany(
operation, seq_of_parameters
class AsyncAdapt_asyncpg_dbapi:
def __init__(self, asyncpg):
self.asyncpg = asyncpg
- self.paramstyle = "format"
+ self.paramstyle = "numeric_dollar"
def connect(self, *arg, **kw):
async_fallback = kw.pop("async_fallback", False)
render_bind_cast = True
has_terminate = True
- default_paramstyle = "format"
+ default_paramstyle = "numeric_dollar"
supports_sane_multi_rowcount = False
execution_ctx_cls = PGExecutionContext_asyncpg
statement_compiler = PGCompiler_asyncpg
# TODO: I can't get this to build dynamically with pytest-xdist procs
-_drivernames = {"pysqlite", "aiosqlite", "pysqlcipher"}
+_drivernames = {
+ "pysqlite",
+ "aiosqlite",
+ "pysqlcipher",
+ "pysqlite_numeric",
+ "pysqlite_dollar",
+}
@generate_driver_url.for_db("sqlite")
dialect = SQLiteDialect_pysqlite
+
+
+class _SQLiteDialect_pysqlite_numeric(SQLiteDialect_pysqlite):
+ """numeric dialect for testing only
+
+ internal use only. This dialect is **NOT** supported by SQLAlchemy
+ and may change at any time.
+
+ """
+
+ supports_statement_cache = True
+ default_paramstyle = "numeric"
+ driver = "pysqlite_numeric"
+
+ _first_bind = ":1"
+ _not_in_statement_regexp = None
+
+ def __init__(self, *arg, **kw):
+ kw.setdefault("paramstyle", "numeric")
+ super().__init__(*arg, **kw)
+
+ def create_connect_args(self, url):
+ arg, opts = super().create_connect_args(url)
+ opts["factory"] = self._fix_sqlite_issue_99953()
+ return arg, opts
+
+ def _fix_sqlite_issue_99953(self):
+ import sqlite3
+
+ first_bind = self._first_bind
+ if self._not_in_statement_regexp:
+ nis = self._not_in_statement_regexp
+
+ def _test_sql(sql):
+ m = nis.search(sql)
+ assert not m, f"Found {nis.pattern!r} in {sql!r}"
+
+ else:
+
+ def _test_sql(sql):
+ pass
+
+ def _numeric_param_as_dict(parameters):
+ if parameters:
+ assert isinstance(parameters, tuple)
+ return {
+ str(idx): value for idx, value in enumerate(parameters, 1)
+ }
+ else:
+ return ()
+
+ class SQLiteFix99953Cursor(sqlite3.Cursor):
+ def execute(self, sql, parameters=()):
+ _test_sql(sql)
+ if first_bind in sql:
+ parameters = _numeric_param_as_dict(parameters)
+ return super().execute(sql, parameters)
+
+ def executemany(self, sql, parameters):
+ _test_sql(sql)
+ if first_bind in sql:
+ parameters = [
+ _numeric_param_as_dict(p) for p in parameters
+ ]
+ return super().executemany(sql, parameters)
+
+ class SQLiteFix99953Connection(sqlite3.Connection):
+ def cursor(self, factory=None):
+ if factory is None:
+ factory = SQLiteFix99953Cursor
+ return super().cursor(factory=factory)
+
+ def execute(self, sql, parameters=()):
+ _test_sql(sql)
+ if first_bind in sql:
+ parameters = _numeric_param_as_dict(parameters)
+ return super().execute(sql, parameters)
+
+ def executemany(self, sql, parameters):
+ _test_sql(sql)
+ if first_bind in sql:
+ parameters = [
+ _numeric_param_as_dict(p) for p in parameters
+ ]
+ return super().executemany(sql, parameters)
+
+ return SQLiteFix99953Connection
+
+
+class _SQLiteDialect_pysqlite_dollar(_SQLiteDialect_pysqlite_numeric):
+ """numeric dialect that uses $ for testing only
+
+ internal use only. This dialect is **NOT** supported by SQLAlchemy
+ and may change at any time.
+
+ """
+
+ supports_statement_cache = True
+ default_paramstyle = "numeric_dollar"
+ driver = "pysqlite_dollar"
+
+ _first_bind = "$1"
+ _not_in_statement_regexp = re.compile(r"[^\d]:\d+")
+
+ def __init__(self, *arg, **kw):
+ kw.setdefault("paramstyle", "numeric_dollar")
+ super().__init__(*arg, **kw)
self.paramstyle = self.dbapi.paramstyle
else:
self.paramstyle = self.default_paramstyle
- self.positional = self.paramstyle in ("qmark", "format", "numeric")
+ self.positional = self.paramstyle in (
+ "qmark",
+ "format",
+ "numeric",
+ "numeric_dollar",
+ )
self.identifier_preparer = self.preparer(self)
self._on_connect_isolation_level = isolation_level
_ImmutableExecuteOptions = immutabledict[str, Any]
-_ParamStyle = Literal["qmark", "numeric", "named", "format", "pyformat"]
+_ParamStyle = Literal[
+ "qmark", "numeric", "named", "format", "pyformat", "numeric_dollar"
+]
_GenericSetInputSizesType = List[Tuple[str, Any, "TypeEngine[Any]"]]
BIND_PARAMS = re.compile(r"(?<![:\w\$\x5c]):([\w\$]+)(?![:\w\$])", re.UNICODE)
BIND_PARAMS_ESC = re.compile(r"\x5c(:[\w\$]*)(?![:\w\$])", re.UNICODE)
+_pyformat_template = "%%(%(name)s)s"
BIND_TEMPLATES = {
- "pyformat": "%%(%(name)s)s",
+ "pyformat": _pyformat_template,
"qmark": "?",
"format": "%%s",
"numeric": ":[_POSITION]",
+ "numeric_dollar": "$[_POSITION]",
"named": ":%(name)s",
}
num_positional_params_counted: int
+class CompilerState(IntEnum):
+ COMPILING = 0
+ """statement is present, compilation phase in progress"""
+
+ STRING_APPLIED = 1
+ """statement is present, string form of the statement has been applied.
+
+ Additional processors by subclasses may still be pending.
+
+ """
+
+ NO_STATEMENT = 2
+ """compiler does not have a statement to compile, is used
+ for method access"""
+
+
class Linting(IntEnum):
"""represent preferences for the 'SQL linting' feature.
defaults.
"""
+ statement: Optional[ClauseElement] = None
+ "The statement to compile."
+ string: str = ""
+ "The string representation of the ``statement``"
+
+ state: CompilerState
+ """description of the compiler's state"""
+
is_sql = False
is_ddl = False
"""
-
self.dialect = dialect
self.preparer = self.dialect.identifier_preparer
if schema_translate_map:
)
if statement is not None:
+ self.state = CompilerState.COMPILING
self.statement = statement
self.can_execute = statement.supports_execution
self._annotations = statement._annotations
self.string = self.preparer._render_schema_translates(
self.string, schema_translate_map
)
+
+ self.state = CompilerState.STRING_APPLIED
+ else:
+ self.state = CompilerState.NO_STATEMENT
+
self._gen_time = perf_counter()
def _execute_on_connection(
def __str__(self) -> str:
"""Return the string text of the generated SQL or DDL."""
- return self.string or ""
+ if self.state is CompilerState.STRING_APPLIED:
+ return self.string
+ else:
+ return ""
def construct_params(
self,
driver/DB enforces this
"""
+ bindtemplate: str
+ """template to render bound parameters based on paramstyle."""
+
+ compilation_bindtemplate: str
+ """template used by compiler to render parameters before positional
+ paramstyle application"""
+
+ _numeric_binds_identifier_char: str
+ """Character that's used to as the identifier of a numerical bind param.
+ For example if this char is set to ``$``, numerical binds will be rendered
+ in the form ``$1, $2, $3``.
+ """
+
_result_columns: List[ResultColumnsEntry]
"""relates label names in the final SQL to a tuple of local
column/label name, ColumnElement object (if any) and
and is combined with the :attr:`_sql.Compiled.params` dictionary to
render parameters.
+ This sequence always contains the unescaped name of the parameters.
+
.. seealso::
:ref:`faq_sql_expression_string` - includes a usage example for
debugging use cases.
"""
- positiontup_level: Optional[Dict[str, int]] = None
+ _values_bindparam: Optional[List[str]] = None
+
+ _visited_bindparam: Optional[List[str]] = None
inline: bool = False
level_name_by_cte: Dict[CTE, Tuple[int, str, selectable._CTEOpts]]
ctes_recursive: bool
- cte_positional: Dict[CTE, List[str]]
- cte_level: Dict[CTE, int]
- cte_order: Dict[Optional[CTE], List[CTE]]
+
+ _post_compile_pattern = re.compile(r"__\[POSTCOMPILE_(\S+?)(~~.+?~~)?\]")
+ _pyformat_pattern = re.compile(r"%\(([^)]+?)\)s")
+ _positional_pattern = re.compile(
+ f"{_pyformat_pattern.pattern}|{_post_compile_pattern.pattern}"
+ )
def __init__(
self,
# true if the paramstyle is positional
self.positional = dialect.positional
if self.positional:
- self.positiontup_level = {}
- self.positiontup = []
- self._numeric_binds = dialect.paramstyle == "numeric"
- self.bindtemplate = BIND_TEMPLATES[dialect.paramstyle]
+ self._numeric_binds = nb = dialect.paramstyle.startswith("numeric")
+ if nb:
+ self._numeric_binds_identifier_char = (
+ "$" if dialect.paramstyle == "numeric_dollar" else ":"
+ )
+
+ self.compilation_bindtemplate = _pyformat_template
+ else:
+ self.compilation_bindtemplate = BIND_TEMPLATES[dialect.paramstyle]
self.ctes = None
):
self.inline = True
- if self.positional and self._numeric_binds:
- self._apply_numbered_params()
+ self.bindtemplate = BIND_TEMPLATES[dialect.paramstyle]
+
+ if self.state is CompilerState.STRING_APPLIED:
+ if self.positional:
+ if self._numeric_binds:
+ self._process_numeric()
+ else:
+ self._process_positional()
- if self._render_postcompile:
- self._process_parameters_for_postcompile(_populate_self=True)
+ if self._render_postcompile:
+ self._process_parameters_for_postcompile(_populate_self=True)
@property
def insert_single_values_expr(self) -> Optional[str]:
"""
if self.implicit_returning:
return self.implicit_returning
- elif is_dml(self.statement):
+ elif self.statement is not None and is_dml(self.statement):
return [
c
for c in self.statement._all_selected_columns
self.level_name_by_cte = {}
self.ctes_recursive = False
- if self.positional:
- self.cte_positional = {}
- self.cte_level = {}
- self.cte_order = collections.defaultdict(list)
return ctes
ordered_columns,
)
- def _apply_numbered_params(self):
- poscount = itertools.count(1)
+ def _process_positional(self):
+ assert not self.positiontup
+ assert self.state is CompilerState.STRING_APPLIED
+ assert not self._numeric_binds
+
+ if self.dialect.paramstyle == "format":
+ placeholder = "%s"
+ else:
+ assert self.dialect.paramstyle == "qmark"
+ placeholder = "?"
+
+ positions = []
+
+ def find_position(m: re.Match[str]) -> str:
+ normal_bind = m.group(1)
+ if normal_bind:
+ positions.append(normal_bind)
+ return placeholder
+ else:
+ # this a post-compile bind
+ positions.append(m.group(2))
+ return m.group(0)
+
self.string = re.sub(
- r"\[_POSITION\]", lambda m: str(next(poscount)), self.string
+ self._positional_pattern, find_position, self.string
)
+ if self.escaped_bind_names:
+ reverse_escape = {v: k for k, v in self.escaped_bind_names.items()}
+ assert len(self.escaped_bind_names) == len(reverse_escape)
+ self.positiontup = [
+ reverse_escape.get(name, name) for name in positions
+ ]
+ else:
+ self.positiontup = positions
+
+ if self._insertmanyvalues:
+ positions = []
+ single_values_expr = re.sub(
+ self._positional_pattern,
+ find_position,
+ self._insertmanyvalues.single_values_expr,
+ )
+ insert_crud_params = [
+ (
+ v[0],
+ v[1],
+ re.sub(self._positional_pattern, find_position, v[2]),
+ v[3],
+ )
+ for v in self._insertmanyvalues.insert_crud_params
+ ]
+
+ self._insertmanyvalues = _InsertManyValues(
+ is_default_expr=self._insertmanyvalues.is_default_expr,
+ single_values_expr=single_values_expr,
+ insert_crud_params=insert_crud_params,
+ num_positional_params_counted=(
+ self._insertmanyvalues.num_positional_params_counted
+ ),
+ )
+
+ def _process_numeric(self):
+ assert self._numeric_binds
+ assert self.state is CompilerState.STRING_APPLIED
+
+ num = 1
+ param_pos: Dict[str, str] = {}
+ order: Iterable[str]
+ if self._insertmanyvalues and self._values_bindparam is not None:
+ # bindparams that are not in values are always placed first.
+ # this avoids the need of changing them when using executemany
+ # values () ()
+ order = itertools.chain(
+ (
+ name
+ for name in self.bind_names.values()
+ if name not in self._values_bindparam
+ ),
+ self.bind_names.values(),
+ )
+ else:
+ order = self.bind_names.values()
+
+ for bind_name in order:
+ if bind_name in param_pos:
+ continue
+ bind = self.binds[bind_name]
+ if (
+ bind in self.post_compile_params
+ or bind in self.literal_execute_params
+ ):
+ # set to None to just mark the in positiontup, it will not
+ # be replaced below.
+ param_pos[bind_name] = None # type: ignore
+ else:
+ ph = f"{self._numeric_binds_identifier_char}{num}"
+ num += 1
+ param_pos[bind_name] = ph
+
+ self.next_numeric_pos = num
+
+ self.positiontup = list(param_pos)
+ if self.escaped_bind_names:
+ reverse_escape = {v: k for k, v in self.escaped_bind_names.items()}
+ assert len(self.escaped_bind_names) == len(reverse_escape)
+ param_pos = {
+ self.escaped_bind_names.get(name, name): pos
+ for name, pos in param_pos.items()
+ }
+
+ # Can't use format here since % chars are not escaped.
+ self.string = self._pyformat_pattern.sub(
+ lambda m: param_pos[m.group(1)], self.string
+ )
+
+ if self._insertmanyvalues:
+ single_values_expr = (
+ # format is ok here since single_values_expr includes only
+ # place-holders
+ self._insertmanyvalues.single_values_expr
+ % param_pos
+ )
+ insert_crud_params = [
+ (v[0], v[1], "%s", v[3])
+ for v in self._insertmanyvalues.insert_crud_params
+ ]
+
+ self._insertmanyvalues = _InsertManyValues(
+ is_default_expr=self._insertmanyvalues.is_default_expr,
+ # This has the numbers (:1, :2)
+ single_values_expr=single_values_expr,
+ # The single binds are instead %s so they can be formatted
+ insert_crud_params=insert_crud_params,
+ num_positional_params_counted=(
+ self._insertmanyvalues.num_positional_params_counted
+ ),
+ )
+
@util.memoized_property
def _bind_processors(
self,
new_processors: Dict[str, _BindProcessorType[Any]] = {}
- if self.positional and self._numeric_binds:
- # I'm not familiar with any DBAPI that uses 'numeric'.
- # strategy would likely be to make use of numbers greater than
- # the highest number present; then for expanding parameters,
- # append them to the end of the parameter list. that way
- # we avoid having to renumber all the existing parameters.
- raise NotImplementedError(
- "'post-compile' bind parameters are not supported with "
- "the 'numeric' paramstyle at this time."
- )
-
replacement_expressions: Dict[str, Any] = {}
to_update_sets: Dict[str, Any] = {}
# notes:
# *unescaped* parameter names in:
- # self.bind_names, self.binds, self._bind_processors
+ # self.bind_names, self.binds, self._bind_processors, self.positiontup
#
# *escaped* parameter names in:
# construct_params(), replacement_expressions
+ numeric_positiontup: Optional[List[str]] = None
+
if self.positional and self.positiontup is not None:
names: Iterable[str] = self.positiontup
+ if self._numeric_binds:
+ numeric_positiontup = []
else:
names = self.bind_names.values()
+ ebn = self.escaped_bind_names
for name in names:
- escaped_name = (
- self.escaped_bind_names.get(name, name)
- if self.escaped_bind_names
- else name
- )
+ escaped_name = ebn.get(name, name) if ebn else name
parameter = self.binds[name]
+
if parameter in self.literal_execute_params:
if escaped_name not in replacement_expressions:
value = parameters.pop(escaped_name)
# in the escaped_bind_names dictionary.
values = parameters.pop(name)
- leep = self._literal_execute_expanding_parameter
- to_update, replacement_expr = leep(
+ leep_res = self._literal_execute_expanding_parameter(
escaped_name, parameter, values
)
+ (to_update, replacement_expr) = leep_res
to_update_sets[escaped_name] = to_update
replacement_expressions[escaped_name] = replacement_expr
for key, _ in to_update
if name in single_processors
)
- if positiontup is not None:
+ if numeric_positiontup is not None:
+ numeric_positiontup.extend(
+ name for name, _ in to_update
+ )
+ elif positiontup is not None:
+ # to_update has escaped names, but that's ok since
+ # these are new names, that aren't in the
+ # escaped_bind_names dict.
positiontup.extend(name for name, _ in to_update)
expanded_parameters[name] = [
expand_key for expand_key, _ in to_update
return expr
statement = re.sub(
- r"__\[POSTCOMPILE_(\S+?)(~~.+?~~)?\]",
- process_expanding,
- self.string,
+ self._post_compile_pattern, process_expanding, self.string
)
+ if numeric_positiontup is not None:
+ assert positiontup is not None
+ param_pos = {
+ key: f"{self._numeric_binds_identifier_char}{num}"
+ for num, key in enumerate(
+ numeric_positiontup, self.next_numeric_pos
+ )
+ }
+ # Can't use format here since % chars are not escaped.
+ statement = self._pyformat_pattern.sub(
+ lambda m: param_pos[m.group(1)], statement
+ )
+ positiontup.extend(numeric_positiontup)
+
expanded_state = ExpandedState(
statement,
parameters,
text = self.process(taf.element, **kw)
if self.ctes:
nesting_level = len(self.stack) if not toplevel else None
- text = (
- self._render_cte_clause(
- nesting_level=nesting_level,
- visiting_cte=kw.get("visiting_cte"),
- )
- + text
- )
+ text = self._render_cte_clause(nesting_level=nesting_level) + text
self.stack.pop(-1)
self._render_cte_clause(
nesting_level=nesting_level,
include_following_stack=True,
- visiting_cte=kwargs.get("visiting_cte"),
)
+ text
)
dialect = self.dialect
typ_dialect_impl = parameter.type._unwrapped_dialect_impl(dialect)
+ if self._numeric_binds:
+ bind_template = self.compilation_bindtemplate
+ else:
+ bind_template = self.bindtemplate
+
if (
self.dialect._bind_typing_render_casts
and typ_dialect_impl.render_bind_cast
return self.render_bind_cast(
parameter.type,
typ_dialect_impl,
- self.bindtemplate % {"name": name},
+ bind_template % {"name": name},
)
else:
def _render_bindtemplate(name):
- return self.bindtemplate % {"name": name}
+ return bind_template % {"name": name}
if not values:
to_update = []
def bindparam_string(
self,
name: str,
- positional_names: Optional[List[str]] = None,
post_compile: bool = False,
expanding: bool = False,
escaped_from: Optional[str] = None,
**kw: Any,
) -> str:
- if self.positional:
- if positional_names is not None:
- positional_names.append(name)
- else:
- self.positiontup.append(name) # type: ignore[union-attr]
- self.positiontup_level[name] = len(self.stack) # type: ignore[index] # noqa: E501
+ if self._visited_bindparam is not None:
+ self._visited_bindparam.append(name)
+
if not escaped_from:
if _BIND_TRANSLATE_RE.search(name):
if type_impl.render_literal_cast:
ret = self.render_bind_cast(bindparam_type, type_impl, ret)
return ret
+ elif self.state is CompilerState.COMPILING:
+ ret = self.compilation_bindtemplate % {"name": name}
else:
ret = self.bindtemplate % {"name": name}
self.level_name_by_cte[_reference_cte] = new_level_name + (
cte_opts,
)
- if self.positional:
- self.cte_level[cte] = cte_level
else:
cte_level = len(self.stack) if nesting else 1
self.level_name_by_cte[_reference_cte] = cte_level_name + (
cte_opts,
)
- if self.positional:
- self.cte_level[cte] = cte_level
if pre_alias_cte not in self.ctes:
self.visit_cte(pre_alias_cte, **kwargs)
)
)
- if self.positional:
- kwargs["positional_names"] = self.cte_positional[cte] = []
-
assert kwargs.get("subquery", False) is False
if not self.stack:
# In compound query, CTEs are shared at the compound level
if self.ctes and (not is_embedded_select or toplevel):
nesting_level = len(self.stack) if not toplevel else None
- text = (
- self._render_cte_clause(
- nesting_level=nesting_level,
- visiting_cte=kwargs.get("visiting_cte"),
- )
- + text
- )
+ text = self._render_cte_clause(nesting_level=nesting_level) + text
if select_stmt._suffixes:
text += " " + self._generate_prefixes(
self,
nesting_level=None,
include_following_stack=False,
- visiting_cte=None,
):
"""
include_following_stack
return ""
ctes_recursive = any([cte.recursive for cte in ctes])
- if self.positional:
- self.cte_order[visiting_cte].extend(ctes)
-
- if visiting_cte is None and self.cte_order:
- assert self.positiontup is not None
-
- def get_nested_positional(cte):
- if cte in self.cte_order:
- children = self.cte_order.pop(cte)
- to_add = list(
- itertools.chain.from_iterable(
- get_nested_positional(child_cte)
- for child_cte in children
- )
- )
- if cte in self.cte_positional:
- return reorder_positional(
- self.cte_positional[cte],
- to_add,
- self.cte_level[children[0]],
- )
- else:
- return to_add
- else:
- return self.cte_positional.get(cte, [])
-
- def reorder_positional(pos, to_add, level):
- if not level:
- return to_add + pos
- index = 0
- for index, name in enumerate(reversed(pos)):
- if self.positiontup_level[name] < level: # type: ignore[index] # noqa: E501
- break
- return pos[:-index] + to_add + pos[-index:]
-
- to_add = get_nested_positional(None)
- self.positiontup = reorder_positional(
- self.positiontup, to_add, nesting_level
- )
-
cte_text = self.get_cte_preamble(ctes_recursive) + " "
cte_text += ", \n".join([txt for txt in ctes.values()])
cte_text += "\n "
keys_to_replace = set()
base_parameters = {}
executemany_values_w_comma = f"({imv.single_values_expr}), "
+ if self._numeric_binds:
+ escaped = re.escape(self._numeric_binds_identifier_char)
+ executemany_values_w_comma = re.sub(
+ rf"{escaped}\d+", "%s", executemany_values_w_comma
+ )
while batches:
batch = batches[0:batch_size]
num_ins_params = imv.num_positional_params_counted
+ batch_iterator: Iterable[Tuple[Any, ...]]
if num_ins_params == len(batch[0]):
extra_params = ()
- batch_iterator: Iterable[Tuple[Any, ...]] = batch
- elif self.returning_precedes_values:
+ batch_iterator = batch
+ elif self.returning_precedes_values or self._numeric_binds:
extra_params = batch[0][:-num_ins_params]
batch_iterator = (b[-num_ins_params:] for b in batch)
else:
extra_params = batch[0][num_ins_params:]
batch_iterator = (b[:num_ins_params] for b in batch)
+ values_string = (executemany_values_w_comma * len(batch))[:-2]
+ if self._numeric_binds and num_ins_params > 0:
+ # need to format here, since statement may contain
+ # unescaped %, while values_string contains just (%s, %s)
+ start = len(extra_params) + 1
+ end = num_ins_params * len(batch) + start
+ positions = tuple(
+ f"{self._numeric_binds_identifier_char}{i}"
+ for i in range(start, end)
+ )
+ values_string = values_string % positions
+
replaced_statement = statement.replace(
- "__EXECMANY_TOKEN__",
- (executemany_values_w_comma * len(batch))[:-2],
+ "__EXECMANY_TOKEN__", values_string
)
replaced_parameters = tuple(
itertools.chain.from_iterable(batch_iterator)
)
- if self.returning_precedes_values:
+ if self.returning_precedes_values or self._numeric_binds:
replaced_parameters = extra_params + replaced_parameters
else:
replaced_parameters = replaced_parameters + extra_params
}
)
- positiontup_before = positiontup_after = 0
+ counted_bindparam = 0
# for positional, insertmanyvalues needs to know how many
# bound parameters are in the VALUES sequence; there's no simple
# rule because default expressions etc. can have zero or more
# params inside them. After multiple attempts to figure this out,
- # this very simplistic "count before, then count after" works and is
+ # this very simplistic "count after" works and is
# likely the least amount of callcounts, though looks clumsy
- if self.positiontup:
- positiontup_before = len(self.positiontup)
+ if self.positional:
+ self._visited_bindparam = []
crud_params_struct = crud._get_crud_params(
self, insert_stmt, compile_state, toplevel, **kw
)
- if self.positiontup:
- positiontup_after = len(self.positiontup)
+ if self.positional:
+ assert self._visited_bindparam is not None
+ counted_bindparam = len(self._visited_bindparam)
+ if self._numeric_binds:
+ if self._values_bindparam is not None:
+ self._values_bindparam += self._visited_bindparam
+ else:
+ self._values_bindparam = self._visited_bindparam
+ self._visited_bindparam = None
crud_params_single = crud_params_struct.single_params
if self.implicit_returning or insert_stmt._returning:
- # if returning clause is rendered first, capture bound parameters
- # while visiting and place them prior to the VALUES oriented
- # bound parameters, when using positional parameter scheme
- rpv = self.returning_precedes_values
- flip_pt = rpv and self.positional
- if flip_pt:
- pt: Optional[List[str]] = self.positiontup
- temp_pt: Optional[List[str]]
- self.positiontup = temp_pt = []
- else:
- temp_pt = pt = None
-
returning_clause = self.returning_clause(
insert_stmt,
self.implicit_returning or insert_stmt._returning,
populate_result_map=toplevel,
)
- if flip_pt:
- if TYPE_CHECKING:
- assert temp_pt is not None
- assert pt is not None
- self.positiontup = temp_pt + pt
-
- if rpv:
+ if self.returning_precedes_values:
text += " " + returning_clause
else:
self._render_cte_clause(
nesting_level=nesting_level,
include_following_stack=True,
- visiting_cte=kw.get("visiting_cte"),
),
select_text,
)
cast(
"List[crud._CrudParamElementStr]", crud_params_single
),
- (positiontup_after - positiontup_before),
+ counted_bindparam,
)
elif compile_state._has_multi_parameters:
text += " VALUES %s" % (
"List[crud._CrudParamElementStr]",
crud_params_single,
),
- positiontup_after - positiontup_before,
+ counted_bindparam,
)
if insert_stmt._post_values_clause is not None:
self._render_cte_clause(
nesting_level=nesting_level,
include_following_stack=True,
- visiting_cte=kw.get("visiting_cte"),
)
+ text
)
if self.ctes:
nesting_level = len(self.stack) if not toplevel else None
- text = (
- self._render_cte_clause(
- nesting_level=nesting_level,
- visiting_cte=kw.get("visiting_cte"),
- )
- + text
- )
+ text = self._render_cte_clause(nesting_level=nesting_level) + text
self.stack.pop(-1)
if self.ctes:
nesting_level = len(self.stack) if not toplevel else None
- text = (
- self._render_cte_clause(
- nesting_level=nesting_level,
- visiting_cte=kw.get("visiting_cte"),
- )
- + text
- )
+ text = self._render_cte_clause(nesting_level=nesting_level) + text
self.stack.pop(-1)
]
_CrudParamElementStr = Tuple[
"KeyedColumnElement[Any]",
- str,
- str,
+ str, # column name
+ str, # placeholder
Iterable[str],
]
_CrudParamElementSQLExpr = Tuple[
import collections
import contextlib
+import itertools
import re
from .. import event
return received_stmt, execute_observed.context.compiled_parameters
- def _dialect_adjusted_statement(self, paramstyle):
+ def _dialect_adjusted_statement(self, dialect):
+ paramstyle = dialect.paramstyle
stmt = re.sub(r"[\n\t]", "", self.statement)
# temporarily escape out PG double colons
repl = "?"
elif paramstyle == "format":
repl = r"%s"
- elif paramstyle == "numeric":
- repl = None
+ elif paramstyle.startswith("numeric"):
+ counter = itertools.count(1)
+
+ num_identifier = "$" if paramstyle == "numeric_dollar" else ":"
+
+ def repl(m):
+ return f"{num_identifier}{next(counter)}"
+
stmt = re.sub(r":([\w_]+)", repl, stmt)
# put them back
return stmt
def _compare_sql(self, execute_observed, received_statement):
- paramstyle = execute_observed.context.dialect.paramstyle
- stmt = self._dialect_adjusted_statement(paramstyle)
+ stmt = self._dialect_adjusted_statement(
+ execute_observed.context.dialect
+ )
return received_statement == stmt
def _failure_message(self, execute_observed, expected_params):
- paramstyle = execute_observed.context.dialect.paramstyle
return (
"Testing for compiled statement\n%r partial params %s, "
"received\n%%(received_statement)r with params "
"%%(received_parameters)r"
% (
- self._dialect_adjusted_statement(paramstyle).replace(
- "%", "%%"
- ),
+ self._dialect_adjusted_statement(
+ execute_observed.context.dialect
+ ).replace("%", "%%"),
repr(expected_params).replace("%", "%%"),
)
)
elif querytyp.legacy_query:
stmt = Session.query(Thing)
else:
- assert False
+ querytyp.fail()
The variable provided is a slots object of boolean variables, as well
options = opt
+@pre
+def _register_sqlite_numeric_dialect(opt, file_config):
+ from sqlalchemy.dialects import registry
+
+ registry.register(
+ "sqlite.pysqlite_numeric",
+ "sqlalchemy.dialects.sqlite.pysqlite",
+ "_SQLiteDialect_pysqlite_numeric",
+ )
+ registry.register(
+ "sqlite.pysqlite_dollar",
+ "sqlalchemy.dialects.sqlite.pysqlite",
+ "_SQLiteDialect_pysqlite_dollar",
+ )
+
+
@post
def __ensure_cext(opt, file_config):
if os.environ.get("REQUIRE_SQLALCHEMY_CEXT", "0") == "1":
[db]
default = sqlite:///:memory:
sqlite = sqlite:///:memory:
+sqlite_numeric = sqlite+pysqlite_numeric:///:memory:
+sqlite_dollar = sqlite+pysqlite_dollar:///:memory:
aiosqlite = sqlite+aiosqlite:///:memory:
sqlite_file = sqlite:///querytest.db
aiosqlite_file = sqlite+aiosqlite:///async_querytest.db
"matchtable.title @@ plainto_tsquery(%(title_1)s)",
)
- @testing.requires.format_paramstyle
+ @testing.only_if("+asyncpg")
def test_expression_positional(self, connection):
matchtable = self.tables.matchtable
if self._strs_render_bind_casts(connection):
self.assert_compile(
matchtable.c.title.match("somstr"),
- "matchtable.title @@ plainto_tsquery(%s::VARCHAR(200))",
+ "matchtable.title @@ plainto_tsquery($1::VARCHAR(200))",
)
else:
self.assert_compile(
matchtable.c.title.match("somstr"),
- "matchtable.title @@ plainto_tsquery(%s)",
+ "matchtable.title @@ plainto_tsquery($1)",
)
def test_simple_match(self, connection):
)
@testing.combinations("control", "excluded", "dict")
+ @testing.skip_if("+pysqlite_numeric")
+ @testing.skip_if("+pysqlite_dollar")
def test_set_excluded(self, scenario):
"""test #8014, sending all of .excluded to set"""
class LogParamsTest(fixtures.TestBase):
- __only_on__ = "sqlite"
+ __only_on__ = "sqlite+pysqlite"
__requires__ = ("ad_hoc_engines",)
def setup_test(self):
class TransactionContextLoggingTest(fixtures.TestBase):
- __only_on__ = "sqlite"
+ __only_on__ = "sqlite+pysqlite"
@testing.fixture()
def plain_assert_buf(self, plain_logging_engine):
BulkDMLReturningInhTest, fixtures.DeclarativeMappedTest
):
- __requires__ = ("insert_returning",)
+ __requires__ = ("insert_returning", "insert_executemany_returning")
__backend__ = True
@classmethod
class BulkDMLReturningSingleInhTest(
BulkDMLReturningInhTest, fixtures.DeclarativeMappedTest
):
- __requires__ = ("insert_returning",)
+ __requires__ = ("insert_returning", "insert_executemany_returning")
__backend__ = True
@classmethod
class BulkDMLReturningConcreteInhTest(
BulkDMLReturningInhTest, fixtures.DeclarativeMappedTest
):
- __requires__ = ("insert_returning",)
+ __requires__ = ("insert_returning", "insert_executemany_returning")
__backend__ = True
@classmethod
"INSERT INTO users (name) VALUES (:name)",
[{"name": "x"}],
),
- CompiledSQL(
- "INSERT INTO addresses (user_id, email_address) "
- "VALUES (:user_id, :email_address) "
- "RETURNING addresses.id",
+ Conditional(
+ testing.requires.insert_executemany_returning.enabled,
+ [
+ CompiledSQL(
+ "INSERT INTO addresses "
+ "(user_id, email_address) "
+ "VALUES (:user_id, :email_address) "
+ "RETURNING addresses.id",
+ [
+ {"user_id": uid, "email_address": "e1"},
+ {"user_id": uid, "email_address": "e2"},
+ {"user_id": uid, "email_address": "e3"},
+ ],
+ )
+ ],
[
- {"user_id": uid, "email_address": "e1"},
- {"user_id": uid, "email_address": "e2"},
- {"user_id": uid, "email_address": "e3"},
+ CompiledSQL(
+ "INSERT INTO addresses "
+ "(user_id, email_address) "
+ "VALUES (:user_id, :email_address)",
+ param,
+ )
+ for param in [
+ {"user_id": uid, "email_address": "e1"},
+ {"user_id": uid, "email_address": "e2"},
+ {"user_id": uid, "email_address": "e3"},
+ ]
],
),
],
"INSERT INTO users (name) VALUES (:name)",
[{"name": "x"}],
),
- CompiledSQL(
- "INSERT INTO addresses (user_id, email_address) "
- "VALUES (:user_id, :email_address) "
- "RETURNING addresses.id",
+ Conditional(
+ testing.requires.insert_executemany_returning.enabled,
+ [
+ CompiledSQL(
+ "INSERT INTO addresses "
+ "(user_id, email_address) "
+ "VALUES (:user_id, :email_address) "
+ "RETURNING addresses.id",
+ [
+ {"user_id": uid, "email_address": "e1"},
+ {"user_id": uid, "email_address": "e2"},
+ {"user_id": uid, "email_address": "e3"},
+ ],
+ )
+ ],
[
- {"user_id": uid, "email_address": "e1"},
- {"user_id": uid, "email_address": "e2"},
- {"user_id": uid, "email_address": "e3"},
+ CompiledSQL(
+ "INSERT INTO addresses "
+ "(user_id, email_address) "
+ "VALUES (:user_id, :email_address)",
+ param,
+ )
+ for param in [
+ {"user_id": uid, "email_address": "e1"},
+ {"user_id": uid, "email_address": "e2"},
+ {"user_id": uid, "email_address": "e3"},
+ ]
],
),
],
)
attrname = "user"
else:
- assert False
+ direction.fail()
assert attrname in obj_to_merge.__dict__
asserter.assert_(
Conditional(
- testing.db.dialect.insert_executemany_returning,
+ testing.db.dialect.insert_returning,
[
CompiledSQL(
"INSERT INTO test (id) VALUES (:id) "
"mariadb+pymysql",
"mariadb+cymysql",
"mariadb+mysqlconnector",
- "postgresql+asyncpg",
"postgresql+pg8000",
]
)
]
)
+ @property
+ def predictable_gc(self):
+ """target platform must remove all cycles unconditionally when
+ gc.collect() is called, as well as clean out unreferenced subclasses.
+
+ """
+ return self.cpython + skip_if("+aiosqlite")
+
@property
def memory_process_intensive(self):
"""Driver is able to handle the memory tests which run in a subprocess
"mariadb",
"sqlite+aiosqlite",
"sqlite+pysqlite",
+ "sqlite+pysqlite_numeric",
+ "sqlite+pysqlite_dollar",
"sqlite+pysqlcipher",
"mssql",
)
from sqlalchemy.sql.elements import BooleanClauseList
from sqlalchemy.sql.elements import ColumnElement
from sqlalchemy.sql.elements import CompilerColumnElement
+from sqlalchemy.sql.elements import Grouping
from sqlalchemy.sql.expression import ClauseElement
from sqlalchemy.sql.expression import ClauseList
from sqlalchemy.sql.selectable import LABEL_STYLE_NONE
dialect="default",
)
- @standalone_escape
- @testing.variation("use_assert_compile", [True, False])
@testing.variation("use_positional", [True, False])
- def test_standalone_bindparam_escape_expanding(
- self, paramname, expected, use_assert_compile, use_positional
+ def test_standalone_bindparam_escape_collision(self, use_positional):
+ """this case is currently not supported
+
+ it's kinda bad since positional takes the unescaped param
+ while non positional takes the escaped one.
+ """
+ stmt = select(table1.c.myid).where(
+ table1.c.name == bindparam("[brackets]", value="x"),
+ table1.c.description == bindparam("_brackets_", value="y"),
+ )
+
+ if use_positional:
+ self.assert_compile(
+ stmt,
+ "SELECT mytable.myid FROM mytable WHERE mytable.name = ? "
+ "AND mytable.description = ?",
+ params={"[brackets]": "a", "_brackets_": "b"},
+ checkpositional=("a", "a"),
+ dialect="sqlite",
+ )
+ else:
+ self.assert_compile(
+ stmt,
+ "SELECT mytable.myid FROM mytable WHERE mytable.name = "
+ ":_brackets_ AND mytable.description = :_brackets_",
+ params={"[brackets]": "a", "_brackets_": "b"},
+ checkparams={"_brackets_": "b"},
+ dialect="default",
+ )
+
+ paramstyle = testing.variation("paramstyle", ["named", "qmark", "numeric"])
+
+ @standalone_escape
+ @paramstyle
+ def test_standalone_bindparam_escape_expanding_compile(
+ self, paramname, expected, paramstyle
):
stmt = select(table1.c.myid).where(
table1.c.name.in_(bindparam(paramname, value=["a", "b"]))
)
- if use_assert_compile:
- if use_positional:
- self.assert_compile(
- stmt,
- "SELECT mytable.myid FROM mytable "
- "WHERE mytable.name IN (?, ?)",
- params={paramname: ["y", "z"]},
- # NOTE: this is what render_postcompile will do right now
- # if you run construct_params(). render_postcompile mode
- # is not actually used by the execution internals, it's for
- # user-facing compilation code. So this is likely a
- # current limitation of construct_params() which is not
- # doing the full blown postcompile; just assert that's
- # what it does for now. it likely should be corrected
- # to make more sense.
- checkpositional=(["y", "z"], ["y", "z"]),
- dialect="sqlite",
- render_postcompile=True,
- )
- else:
- self.assert_compile(
- stmt,
- "SELECT mytable.myid FROM mytable WHERE mytable.name IN "
- "(:%s_1, :%s_2)" % (expected, expected),
- params={paramname: ["y", "z"]},
- # NOTE: this is what render_postcompile will do right now
- # if you run construct_params(). render_postcompile mode
- # is not actually used by the execution internals, it's for
- # user-facing compilation code. So this is likely a
- # current limitation of construct_params() which is not
- # doing the full blown postcompile; just assert that's
- # what it does for now. it likely should be corrected
- # to make more sense.
- checkparams={
- "%s_1" % expected: ["y", "z"],
- "%s_2" % expected: ["y", "z"],
- },
- dialect="default",
- render_postcompile=True,
- )
+ # NOTE: below the rendered params are just what
+ # render_postcompile will do right now
+ # if you run construct_params(). render_postcompile mode
+ # is not actually used by the execution internals, it's for
+ # user-facing compilation code. So this is likely a
+ # current limitation of construct_params() which is not
+ # doing the full blown postcompile; just assert that's
+ # what it does for now. it likely should be corrected
+ # to make more sense.
+ if paramstyle.qmark:
+ self.assert_compile(
+ stmt,
+ "SELECT mytable.myid FROM mytable "
+ "WHERE mytable.name IN (?, ?)",
+ params={paramname: ["y", "z"]},
+ checkpositional=(["y", "z"], ["y", "z"]),
+ dialect="sqlite",
+ render_postcompile=True,
+ )
+ elif paramstyle.numeric:
+ self.assert_compile(
+ stmt,
+ "SELECT mytable.myid FROM mytable "
+ "WHERE mytable.name IN (:1, :2)",
+ params={paramname: ["y", "z"]},
+ checkpositional=(["y", "z"], ["y", "z"]),
+ dialect=sqlite.dialect(paramstyle="numeric"),
+ render_postcompile=True,
+ )
+ elif paramstyle.named:
+ self.assert_compile(
+ stmt,
+ "SELECT mytable.myid FROM mytable WHERE mytable.name IN "
+ "(:%s_1, :%s_2)" % (expected, expected),
+ params={paramname: ["y", "z"]},
+ checkparams={
+ "%s_1" % expected: ["y", "z"],
+ "%s_2" % expected: ["y", "z"],
+ },
+ dialect="default",
+ render_postcompile=True,
+ )
else:
- # this is what DefaultDialect actually does.
- # this should be matched to DefaultDialect._init_compiled()
- if use_positional:
- compiled = stmt.compile(
- dialect=default.DefaultDialect(paramstyle="qmark")
- )
- else:
- compiled = stmt.compile(dialect=default.DefaultDialect())
+ paramstyle.fail()
- checkparams = compiled.construct_params(
- {paramname: ["y", "z"]}, escape_names=False
- )
+ @standalone_escape
+ @paramstyle
+ def test_standalone_bindparam_escape_expanding(
+ self, paramname, expected, paramstyle
+ ):
+ stmt = select(table1.c.myid).where(
+ table1.c.name.in_(bindparam(paramname, value=["a", "b"]))
+ )
+ # this is what DefaultDialect actually does.
+ # this should be matched to DefaultDialect._init_compiled()
+ if paramstyle.qmark:
+ dialect = default.DefaultDialect(paramstyle="qmark")
+ elif paramstyle.numeric:
+ dialect = default.DefaultDialect(paramstyle="numeric")
+ else:
+ dialect = default.DefaultDialect()
- # nothing actually happened. if the compiler had
- # render_postcompile set, the
- # above weird param thing happens
- eq_(checkparams, {paramname: ["y", "z"]})
+ compiled = stmt.compile(dialect=dialect)
+ checkparams = compiled.construct_params(
+ {paramname: ["y", "z"]}, escape_names=False
+ )
- expanded_state = compiled._process_parameters_for_postcompile(
- checkparams
- )
+ # nothing actually happened. if the compiler had
+ # render_postcompile set, the
+ # above weird param thing happens
+ eq_(checkparams, {paramname: ["y", "z"]})
+
+ expanded_state = compiled._process_parameters_for_postcompile(
+ checkparams
+ )
+ eq_(
+ expanded_state.additional_parameters,
+ {f"{expected}_1": "y", f"{expected}_2": "z"},
+ )
+
+ if paramstyle.qmark or paramstyle.numeric:
eq_(
- expanded_state.additional_parameters,
- {f"{expected}_1": "y", f"{expected}_2": "z"},
+ expanded_state.positiontup,
+ [f"{expected}_1", f"{expected}_2"],
)
- if use_positional:
- eq_(
- expanded_state.positiontup,
- [f"{expected}_1", f"{expected}_2"],
+ @paramstyle
+ def test_expanding_in_repeated(self, paramstyle):
+ stmt = (
+ select(table1)
+ .where(
+ table1.c.name.in_(
+ bindparam("uname", value=["h", "e"], expanding=True)
+ )
+ | table1.c.name.in_(
+ bindparam("uname2", value=["y"], expanding=True)
+ )
+ )
+ .where(table1.c.myid == 8)
+ )
+ stmt = stmt.union(
+ select(table1)
+ .where(
+ table1.c.name.in_(
+ bindparam("uname", value=["h", "e"], expanding=True)
+ )
+ | table1.c.name.in_(
+ bindparam("uname2", value=["y"], expanding=True)
)
+ )
+ .where(table1.c.myid == 9)
+ ).order_by("myid")
+
+ # NOTE: below the rendered params are just what
+ # render_postcompile will do right now
+ # if you run construct_params(). render_postcompile mode
+ # is not actually used by the execution internals, it's for
+ # user-facing compilation code. So this is likely a
+ # current limitation of construct_params() which is not
+ # doing the full blown postcompile; just assert that's
+ # what it does for now. it likely should be corrected
+ # to make more sense.
+
+ if paramstyle.qmark:
+ self.assert_compile(
+ stmt,
+ "SELECT mytable.myid, mytable.name, mytable.description "
+ "FROM mytable "
+ "WHERE (mytable.name IN (?, ?) OR "
+ "mytable.name IN (?)) "
+ "AND mytable.myid = ? "
+ "UNION SELECT mytable.myid, mytable.name, mytable.description "
+ "FROM mytable "
+ "WHERE (mytable.name IN (?, ?) OR "
+ "mytable.name IN (?)) "
+ "AND mytable.myid = ? ORDER BY myid",
+ params={"uname": ["y", "z"], "uname2": ["a"]},
+ checkpositional=(
+ ["y", "z"],
+ ["y", "z"],
+ ["a"],
+ 8,
+ ["y", "z"],
+ ["y", "z"],
+ ["a"],
+ 9,
+ ),
+ dialect="sqlite",
+ render_postcompile=True,
+ )
+ elif paramstyle.numeric:
+ self.assert_compile(
+ stmt,
+ "SELECT mytable.myid, mytable.name, mytable.description "
+ "FROM mytable "
+ "WHERE (mytable.name IN (:3, :4) OR "
+ "mytable.name IN (:5)) "
+ "AND mytable.myid = :1 "
+ "UNION SELECT mytable.myid, mytable.name, mytable.description "
+ "FROM mytable "
+ "WHERE (mytable.name IN (:3, :4) OR "
+ "mytable.name IN (:5)) "
+ "AND mytable.myid = :2 ORDER BY myid",
+ params={"uname": ["y", "z"], "uname2": ["a"]},
+ checkpositional=(8, 9, ["y", "z"], ["y", "z"], ["a"]),
+ dialect=sqlite.dialect(paramstyle="numeric"),
+ render_postcompile=True,
+ )
+ elif paramstyle.named:
+ self.assert_compile(
+ stmt,
+ "SELECT mytable.myid, mytable.name, mytable.description "
+ "FROM mytable "
+ "WHERE (mytable.name IN (:uname_1, :uname_2) OR "
+ "mytable.name IN (:uname2_1)) "
+ "AND mytable.myid = :myid_1 "
+ "UNION SELECT mytable.myid, mytable.name, mytable.description "
+ "FROM mytable "
+ "WHERE (mytable.name IN (:uname_1, :uname_2) OR "
+ "mytable.name IN (:uname2_1)) "
+ "AND mytable.myid = :myid_2 ORDER BY myid",
+ params={"uname": ["y", "z"], "uname2": ["a"]},
+ checkparams={
+ "uname": ["y", "z"],
+ "uname2": ["a"],
+ "uname_1": ["y", "z"],
+ "uname_2": ["y", "z"],
+ "uname2_1": ["a"],
+ "myid_1": 8,
+ "myid_2": 9,
+ },
+ dialect="default",
+ render_postcompile=True,
+ )
+ else:
+ paramstyle.fail()
+
+ def test_numeric_dollar_bindparam(self):
+ stmt = table1.select().where(
+ table1.c.name == "a", table1.c.myid.in_([1, 2])
+ )
+ self.assert_compile(
+ stmt,
+ "SELECT mytable.myid, mytable.name, mytable.description "
+ "FROM mytable "
+ "WHERE mytable.name = $1 "
+ "AND mytable.myid IN ($2, $3)",
+ checkpositional=("a", 1, 2),
+ dialect=default.DefaultDialect(paramstyle="numeric_dollar"),
+ render_postcompile=True,
+ )
class UnsupportedTest(fixtures.TestBase):
"INSERT INTO mytable (myid) VALUES (:myid_m0), (:myid_m1)",
)
+ def test_multirow_insert_positional(self):
+ stmt = table1.insert().values([{"myid": 1}, {"myid": 2}])
+ eq_ignore_whitespace(
+ stmt.compile(dialect=sqlite.dialect()).string,
+ "INSERT INTO mytable (myid) VALUES (?), (?)",
+ )
+
+ def test_multirow_insert_numeric(self):
+ stmt = table1.insert().values([{"myid": 1}, {"myid": 2}])
+ eq_ignore_whitespace(
+ stmt.compile(dialect=sqlite.dialect(paramstyle="numeric")).string,
+ "INSERT INTO mytable (myid) VALUES (:1), (:2)",
+ )
+
+ def test_insert_noparams_numeric(self):
+ ii = table1.insert().returning(table1.c.myid)
+ eq_ignore_whitespace(
+ ii.compile(dialect=sqlite.dialect(paramstyle="numeric")).string,
+ "INSERT INTO mytable (myid, name, description) VALUES "
+ "(:1, :2, :3) RETURNING myid",
+ )
+
def test_cte(self):
# stringify of these was supported anyway by defaultdialect.
stmt = select(table1.c.myid).cte()
"SELECT CAST(mytable.myid AS MyType()) AS myid FROM mytable",
)
+ def test_dialect_sub_compile(self):
+ class Widget(ClauseElement):
+ __visit_name__ = "widget"
+ stringify_dialect = "sqlite"
+
+ def visit_widget(self, element, **kw):
+ return "widget"
+
+ with mock.patch(
+ "sqlalchemy.dialects.sqlite.base.SQLiteCompiler.visit_widget",
+ visit_widget,
+ create=True,
+ ):
+ eq_(str(Grouping(Widget())), "(widget)")
+
+ def test_dialect_sub_compile_w_binds(self):
+ """test sub-compile into a new compiler where
+ state != CompilerState.COMPILING, but we have to render a bindparam
+ string. has to render the correct template.
+
+ """
+
+ class Widget(ClauseElement):
+ __visit_name__ = "widget"
+ stringify_dialect = "sqlite"
+
+ def visit_widget(self, element, **kw):
+ return f"widget {self.process(bindparam('q'), **kw)}"
+
+ with mock.patch(
+ "sqlalchemy.dialects.sqlite.base.SQLiteCompiler.visit_widget",
+ visit_widget,
+ create=True,
+ ):
+ eq_(str(Grouping(Widget())), "(widget ?)")
+
def test_within_group(self):
# stringify of these was supported anyway by defaultdialect.
from sqlalchemy import within_group
self.assert_compile(
s,
'WITH regional_sales AS (SELECT orders."order" '
- 'AS "order", :1 AS anon_2 FROM orders) SELECT '
- 'regional_sales."order", :2 AS anon_1 FROM regional_sales',
- checkpositional=("x", "y"),
+ 'AS "order", :2 AS anon_2 FROM orders) SELECT '
+ 'regional_sales."order", :1 AS anon_1 FROM regional_sales',
+ checkpositional=("y", "x"),
dialect=dialect,
)
self.assert_compile(
s.union(s),
'WITH regional_sales AS (SELECT orders."order" '
- 'AS "order", :1 AS anon_2 FROM orders) SELECT '
- 'regional_sales."order", :2 AS anon_1 FROM regional_sales '
- 'UNION SELECT regional_sales."order", :3 AS anon_1 '
+ 'AS "order", :2 AS anon_2 FROM orders) SELECT '
+ 'regional_sales."order", :1 AS anon_1 FROM regional_sales '
+ 'UNION SELECT regional_sales."order", :1 AS anon_1 '
"FROM regional_sales",
- checkpositional=("x", "y", "y"),
+ checkpositional=("y", "x"),
dialect=dialect,
)
self.assert_compile(
s3,
'WITH regional_sales_1 AS (SELECT orders."order" AS "order" '
- 'FROM orders WHERE orders."order" = :1), regional_sales_2 AS '
- '(SELECT orders."order" = :2 AS anon_1, '
+ 'FROM orders WHERE orders."order" = :2), regional_sales_2 AS '
+ '(SELECT orders."order" = :1 AS anon_1, '
'anon_2."order" AS "order", '
'orders."order" AS order_1, '
'regional_sales_1."order" AS order_2 FROM orders, '
'WHERE orders."order" = :3) SELECT regional_sales_2.anon_1, '
'regional_sales_2."order", regional_sales_2.order_1, '
"regional_sales_2.order_2 FROM regional_sales_2",
- checkpositional=("x", "y", "z"),
+ checkpositional=("y", "x", "z"),
dialect=dialect,
)
dialect=postgresql.dialect(),
)
- def test_heterogeneous_multi_values(self):
+ @testing.variation("paramstyle", ["pg", "qmark", "numeric", "dollar"])
+ def test_heterogeneous_multi_values(self, paramstyle):
"""for #6047, originally I thought we'd take any insert().values()
and be able to convert it to a "many" style execution that we can
cache.
]
)
+ pos_par = (
+ 1,
+ 1,
+ 2,
+ 2,
+ 1,
+ 2,
+ 2,
+ 3,
+ 1,
+ 2,
+ 2,
+ 10,
+ )
+
# SQL expressions in the params at arbitrary locations means
# we have to scan them at compile time, and the shape of the bound
# parameters is not predictable. so for #6047 where I originally
# thought all of values() could be rewritten, this makes it not
# really worth it.
- self.assert_compile(
- stmt,
- "INSERT INTO t (x, y, z) VALUES "
- "(%(x_m0)s, sum(%(sum_1)s, %(sum_2)s), %(z_m0)s), "
- "(sum(%(sum_3)s, %(sum_4)s), %(y_m1)s, %(z_m1)s), "
- "(sum(%(sum_5)s, %(sum_6)s), %(y_m2)s, foo(%(foo_1)s))",
- checkparams={
- "x_m0": 1,
- "sum_1": 1,
- "sum_2": 2,
- "z_m0": 2,
- "sum_3": 1,
- "sum_4": 2,
- "y_m1": 2,
- "z_m1": 3,
- "sum_5": 1,
- "sum_6": 2,
- "y_m2": 2,
- "foo_1": 10,
- },
- dialect=postgresql.dialect(),
- )
+ if paramstyle.pg:
+ self.assert_compile(
+ stmt,
+ "INSERT INTO t (x, y, z) VALUES "
+ "(%(x_m0)s, sum(%(sum_1)s, %(sum_2)s), %(z_m0)s), "
+ "(sum(%(sum_3)s, %(sum_4)s), %(y_m1)s, %(z_m1)s), "
+ "(sum(%(sum_5)s, %(sum_6)s), %(y_m2)s, foo(%(foo_1)s))",
+ checkparams={
+ "x_m0": 1,
+ "sum_1": 1,
+ "sum_2": 2,
+ "z_m0": 2,
+ "sum_3": 1,
+ "sum_4": 2,
+ "y_m1": 2,
+ "z_m1": 3,
+ "sum_5": 1,
+ "sum_6": 2,
+ "y_m2": 2,
+ "foo_1": 10,
+ },
+ dialect=postgresql.dialect(),
+ )
+ elif paramstyle.qmark:
+ self.assert_compile(
+ stmt,
+ "INSERT INTO t (x, y, z) VALUES "
+ "(?, sum(?, ?), ?), "
+ "(sum(?, ?), ?, ?), "
+ "(sum(?, ?), ?, foo(?))",
+ checkpositional=pos_par,
+ dialect=sqlite.dialect(),
+ )
+ elif paramstyle.numeric:
+ self.assert_compile(
+ stmt,
+ "INSERT INTO t (x, y, z) VALUES "
+ "(:1, sum(:2, :3), :4), "
+ "(sum(:5, :6), :7, :8), "
+ "(sum(:9, :10), :11, foo(:12))",
+ checkpositional=pos_par,
+ dialect=sqlite.dialect(paramstyle="numeric"),
+ )
+ elif paramstyle.dollar:
+ self.assert_compile(
+ stmt,
+ "INSERT INTO t (x, y, z) VALUES "
+ "($1, sum($2, $3), $4), "
+ "(sum($5, $6), $7, $8), "
+ "(sum($9, $10), $11, foo($12))",
+ checkpositional=pos_par,
+ dialect=sqlite.dialect(paramstyle="numeric_dollar"),
+ )
+ else:
+ paramstyle.fail()
def test_insert_seq_pk_multi_values_seq_not_supported(self):
m = MetaData()
Column("y", String(50)),
)
- @testing.requires.insert_returning
+ @testing.requires.insert_executemany_returning
def test_splice_horizontally(self, connection):
users = self.tables.users
addresses = self.tables.addresses
elif expression_type.right_side:
expr = (column("x", Integer) == Widget(52)).right
else:
- assert False
+ expression_type.fail()
if secondary_adapt:
is_(expr.type._type_affinity, String)
dialect=dialect,
)
- def test_update_bound_ordering(self):
+ @testing.variation("paramstyle", ["qmark", "format", "numeric"])
+ def test_update_bound_ordering(self, paramstyle):
"""test that bound parameters between the UPDATE and FROM clauses
order correctly in different SQL compilation scenarios.
.values(name="foo")
)
- dialect = default.StrCompileDialect()
- dialect.positional = True
- self.assert_compile(
- upd,
- "UPDATE mytable SET name=:name FROM (SELECT "
- "myothertable.otherid AS otherid, "
- "myothertable.othername AS othername "
- "FROM myothertable "
- "WHERE myothertable.otherid = :otherid_1) AS anon_1 "
- "WHERE mytable.name = anon_1.othername",
- checkpositional=("foo", 5),
- dialect=dialect,
- )
+ if paramstyle.qmark:
- self.assert_compile(
- upd,
- "UPDATE mytable, (SELECT myothertable.otherid AS otherid, "
- "myothertable.othername AS othername "
- "FROM myothertable "
- "WHERE myothertable.otherid = %s) AS anon_1 SET mytable.name=%s "
- "WHERE mytable.name = anon_1.othername",
- checkpositional=(5, "foo"),
- dialect=mysql.dialect(),
- )
+ dialect = default.StrCompileDialect(paramstyle="qmark")
+ self.assert_compile(
+ upd,
+ "UPDATE mytable SET name=? FROM (SELECT "
+ "myothertable.otherid AS otherid, "
+ "myothertable.othername AS othername "
+ "FROM myothertable "
+ "WHERE myothertable.otherid = ?) AS anon_1 "
+ "WHERE mytable.name = anon_1.othername",
+ checkpositional=("foo", 5),
+ dialect=dialect,
+ )
+ elif paramstyle.format:
+ self.assert_compile(
+ upd,
+ "UPDATE mytable, (SELECT myothertable.otherid AS otherid, "
+ "myothertable.othername AS othername "
+ "FROM myothertable "
+ "WHERE myothertable.otherid = %s) AS anon_1 "
+ "SET mytable.name=%s "
+ "WHERE mytable.name = anon_1.othername",
+ checkpositional=(5, "foo"),
+ dialect=mysql.dialect(),
+ )
+ elif paramstyle.numeric:
+ dialect = default.StrCompileDialect(paramstyle="numeric")
+ self.assert_compile(
+ upd,
+ "UPDATE mytable SET name=:1 FROM (SELECT "
+ "myothertable.otherid AS otherid, "
+ "myothertable.othername AS othername "
+ "FROM myothertable "
+ "WHERE myothertable.otherid = :2) AS anon_1 "
+ "WHERE mytable.name = anon_1.othername",
+ checkpositional=("foo", 5),
+ dialect=dialect,
+ )
+ else:
+ paramstyle.fail()
class UpdateFromCompileTest(
sqlite: SQLITE={env:TOX_SQLITE:--db sqlite}
sqlite_file: SQLITE={env:TOX_SQLITE_FILE:--db sqlite_file}
- py3{,7,8,9,10,11}-sqlite: EXTRA_SQLITE_DRIVERS={env:EXTRA_SQLITE_DRIVERS:--dbdriver sqlite --dbdriver aiosqlite}
+ py3{,7,8,9,10,11}-sqlite: EXTRA_SQLITE_DRIVERS={env:EXTRA_SQLITE_DRIVERS:--dbdriver sqlite --dbdriver pysqlite_numeric --dbdriver aiosqlite}
py3{,7,8,9}-sqlite_file: EXTRA_SQLITE_DRIVERS={env:EXTRA_SQLITE_DRIVERS:--dbdriver sqlite --dbdriver aiosqlite --dbdriver pysqlcipher}
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
