"""
- _schema_translate_map = None
_is_future = False
_sqla_logger_namespace = "sqlalchemy.engine.Connection"
self.should_close_with_result = False
self.dispatch = _dispatch
self._has_events = _branch_from._has_events
- self._schema_translate_map = _branch_from._schema_translate_map
else:
self._dbapi_connection = (
connection
if self._has_events or self.engine._has_events:
self.dispatch.engine_connect(self, _branch_from is not None)
+ @property
+ def _schema_translate_map(self):
+ return self._execution_options.get("schema_translate_map", None)
+
def schema_for_object(self, obj):
"""return the schema name for the given schema item taking into
account current schema translate map.
"""
name = obj.schema
- schema_translate_map = self._schema_translate_map
+ schema_translate_map = self._execution_options.get(
+ "schema_translate_map", None
+ )
if (
schema_translate_map
self, ddl, multiparams, params, execution_options
)
+ exec_opts = self._execution_options.merge_with(execution_options)
+ schema_translate_map = exec_opts.get("schema_translate_map", None)
+
dialect = self.dialect
compiled = ddl.compile(
- dialect=dialect, schema_translate_map=self._schema_translate_map
+ dialect=dialect, schema_translate_map=schema_translate_map
)
ret = self._execute_context(
dialect,
dialect = self.dialect
- exec_opts = self._execution_options
- if execution_options:
- exec_opts = exec_opts.union(execution_options)
+ exec_opts = self._execution_options.merge_with(execution_options)
+
+ schema_translate_map = exec_opts.get("schema_translate_map", None)
if "compiled_cache" in exec_opts:
elem_cache_key = elem._generate_cache_key()
dialect,
cache_key,
tuple(sorted(keys)),
- bool(self._schema_translate_map),
+ bool(schema_translate_map),
len(distilled_params) > 1,
)
cache = exec_opts["compiled_cache"]
cache_key=elem_cache_key,
column_keys=keys,
inline=len(distilled_params) > 1,
- schema_translate_map=self._schema_translate_map,
+ schema_translate_map=schema_translate_map,
linting=self.dialect.compiler_linting
| compiler.WARN_LINTING,
)
dialect=dialect,
column_keys=keys,
inline=len(distilled_params) > 1,
- schema_translate_map=self._schema_translate_map,
+ schema_translate_map=schema_translate_map,
linting=self.dialect.compiler_linting | compiler.WARN_LINTING,
)
# the only feature that branching provides
self = self.__branch_from
- if execution_options:
- dialect.set_exec_execution_options(self, execution_options)
-
try:
conn = self._dbapi_connection
if conn is None:
if not branch:
self._set_connection_isolation(connection, isolation_level)
- if "schema_translate_map" in opts:
- engine._schema_translate_map = map_ = opts["schema_translate_map"]
-
- @event.listens_for(engine, "engine_connect")
- def set_schema_translate_map(connection, branch):
- connection._schema_translate_map = map_
-
def set_connection_execution_options(self, connection, opts):
if "isolation_level" in opts:
self._set_connection_isolation(connection, opts["isolation_level"])
- if "schema_translate_map" in opts:
- connection._schema_translate_map = opts["schema_translate_map"]
-
- def set_exec_execution_options(self, connection, opts):
- if "isolation_level" in opts:
- raise exc.InvalidRequestError(
- "The 'isolation_level' execution "
- "option is not supported at the per-statement level"
- )
- self._set_connection_isolation(connection, opts["isolation_level"])
-
- if "schema_translate_map" in opts:
- raise exc.InvalidRequestError(
- "The 'schema_translate_map' execution "
- "option is not supported at the per-statement level"
- )
-
def _set_connection_isolation(self, connection, level):
if connection.in_transaction():
if connection._is_future:
self.unicode_statement = util.text_type(compiled)
if compiled.schema_translate_map:
+ schema_translate_map = self.execution_options.get(
+ "schema_translate_map", {}
+ )
+
rst = compiled.preparer._render_schema_translates
self.unicode_statement = rst(
- self.unicode_statement, connection._schema_translate_map
+ self.unicode_statement, schema_translate_map
)
if not dialect.supports_unicode_statements:
positiontup = self.compiled.positiontup
if compiled.schema_translate_map:
+ schema_translate_map = self.execution_options.get(
+ "schema_translate_map", {}
+ )
rst = compiled.preparer._render_schema_translates
self.unicode_statement = rst(
- self.unicode_statement, connection._schema_translate_map
+ self.unicode_statement, schema_translate_map
)
# final self.unicode_statement is now assigned, encode if needed
def _resolve_for_clause_element(self, element, argname=None, **kw):
original_element = element
- is_clause_element = hasattr(element, "__clause_element__")
+
+ is_clause_element = False
+ while hasattr(element, "__clause_element__"):
+ is_clause_element = True
+ if not getattr(element, "is_clause_element", False):
+ element = element.__clause_element__()
+ else:
+ return element
if is_clause_element:
- while not isinstance(
- element, (elements.ClauseElement, schema.SchemaItem)
- ):
+ return element
+
+ if self._use_inspection:
+ insp = inspection.inspect(element, raiseerr=False)
+ if insp is not None:
try:
- element = element.__clause_element__()
+ return insp.__clause_element__()
except AttributeError:
- break
-
- if not is_clause_element:
- if self._use_inspection:
- insp = inspection.inspect(element, raiseerr=False)
- if insp is not None:
- try:
- return insp.__clause_element__()
- except AttributeError:
- self._raise_for_expected(original_element, argname)
-
- return self._literal_coercion(element, argname=argname, **kw)
- else:
- return element
+ self._raise_for_expected(original_element, argname)
+
+ return self._literal_coercion(element, argname=argname, **kw)
def _implicit_coercions(self, element, resolved, argname=None, **kw):
self._raise_for_expected(element, argname, resolved)
class _Deannotate(object):
+ __slots__ = ()
+
def _post_coercion(self, resolved, **kw):
from .util import _deep_deannotate
class _StringOnly(object):
+ __slots__ = ()
+
def _resolve_for_clause_element(self, element, argname=None, **kw):
return self._literal_coercion(element, **kw)
class _ReturnsStringKey(object):
+ __slots__ = ()
+
def _implicit_coercions(
self, original_element, resolved, argname=None, **kw
):
class _ColumnCoercions(object):
+ __slots__ = ()
+
def _warn_for_scalar_subquery_coercion(self):
util.warn_deprecated(
"coercing SELECT object to scalar subquery in a "
class _NoTextCoercion(object):
+ __slots__ = ()
+
def _literal_coercion(self, element, argname=None, **kw):
if isinstance(element, util.string_types) and issubclass(
elements.TextClause, self._role_class
class _CoerceLiterals(object):
+ __slots__ = ()
_coerce_consts = False
_coerce_star = False
_coerce_numerics = False
class _SelectIsNotFrom(object):
+ __slots__ = ()
+
def _raise_for_expected(self, element, argname=None, resolved=None, **kw):
if isinstance(element, roles.SelectStatementRole) or isinstance(
resolved, roles.SelectStatementRole
)
-class ExpressionElementImpl(
- _ColumnCoercions, RoleImpl, roles.ExpressionElementRole
-):
+class ExpressionElementImpl(_ColumnCoercions, RoleImpl):
+ __slots__ = ()
+
def _literal_coercion(
self, element, name=None, type_=None, argname=None, is_crud=False, **kw
):
self._raise_for_expected(element, err=err)
-class BinaryElementImpl(
- ExpressionElementImpl, RoleImpl, roles.BinaryElementRole
-):
+class BinaryElementImpl(ExpressionElementImpl, RoleImpl):
+
+ __slots__ = ()
+
def _literal_coercion(
self, element, expr, operator, bindparam_type=None, argname=None, **kw
):
return resolved
-class InElementImpl(RoleImpl, roles.InElementRole):
+class InElementImpl(RoleImpl):
+ __slots__ = ()
+
def _implicit_coercions(
self, original_element, resolved, argname=None, **kw
):
return element
-class WhereHavingImpl(
- _CoerceLiterals, _ColumnCoercions, RoleImpl, roles.WhereHavingRole
-):
+class WhereHavingImpl(_CoerceLiterals, _ColumnCoercions, RoleImpl):
+ __slots__ = ()
_coerce_consts = True
return _no_text_coercion(element, argname)
-class StatementOptionImpl(
- _CoerceLiterals, RoleImpl, roles.StatementOptionRole
-):
+class StatementOptionImpl(_CoerceLiterals, RoleImpl):
+ __slots__ = ()
_coerce_consts = True
return elements.TextClause(element)
-class ColumnArgumentImpl(_NoTextCoercion, RoleImpl, roles.ColumnArgumentRole):
- pass
+class ColumnArgumentImpl(_NoTextCoercion, RoleImpl):
+ __slots__ = ()
-class ColumnArgumentOrKeyImpl(
- _ReturnsStringKey, RoleImpl, roles.ColumnArgumentOrKeyRole
-):
- pass
+class ColumnArgumentOrKeyImpl(_ReturnsStringKey, RoleImpl):
+ __slots__ = ()
class ByOfImpl(_CoerceLiterals, _ColumnCoercions, RoleImpl, roles.ByOfRole):
+ __slots__ = ()
+
_coerce_consts = True
def _text_coercion(self, element, argname=None):
return elements._textual_label_reference(element)
-class OrderByImpl(ByOfImpl, RoleImpl, roles.OrderByRole):
+class OrderByImpl(ByOfImpl, RoleImpl):
+ __slots__ = ()
+
def _post_coercion(self, resolved):
if (
isinstance(resolved, self._role_class)
return resolved
-class DMLColumnImpl(_ReturnsStringKey, RoleImpl, roles.DMLColumnRole):
+class DMLColumnImpl(_ReturnsStringKey, RoleImpl):
+ __slots__ = ()
+
def _post_coercion(self, element, as_key=False):
if as_key:
return element.key
return element
-class ConstExprImpl(RoleImpl, roles.ConstExprRole):
+class ConstExprImpl(RoleImpl):
+ __slots__ = ()
+
def _literal_coercion(self, element, argname=None, **kw):
if element is None:
return elements.Null()
self._raise_for_expected(element, argname)
-class TruncatedLabelImpl(_StringOnly, RoleImpl, roles.TruncatedLabelRole):
+class TruncatedLabelImpl(_StringOnly, RoleImpl):
+ __slots__ = ()
+
def _implicit_coercions(
self, original_element, resolved, argname=None, **kw
):
return elements._truncated_label(element)
-class DDLExpressionImpl(
- _Deannotate, _CoerceLiterals, RoleImpl, roles.DDLExpressionRole
-):
+class DDLExpressionImpl(_Deannotate, _CoerceLiterals, RoleImpl):
+
+ __slots__ = ()
_coerce_consts = True
return elements.TextClause(element)
-class DDLConstraintColumnImpl(
- _Deannotate, _ReturnsStringKey, RoleImpl, roles.DDLConstraintColumnRole
-):
- pass
+class DDLConstraintColumnImpl(_Deannotate, _ReturnsStringKey, RoleImpl):
+ __slots__ = ()
class DDLReferredColumnImpl(DDLConstraintColumnImpl):
- pass
+ __slots__ = ()
+
+class LimitOffsetImpl(RoleImpl):
+ __slots__ = ()
-class LimitOffsetImpl(RoleImpl, roles.LimitOffsetRole):
def _implicit_coercions(self, element, resolved, argname=None, **kw):
if resolved is None:
return None
)
-class LabeledColumnExprImpl(
- ExpressionElementImpl, roles.LabeledColumnExprRole
-):
+class LabeledColumnExprImpl(ExpressionElementImpl):
+ __slots__ = ()
+
def _implicit_coercions(
self, original_element, resolved, argname=None, **kw
):
self._raise_for_expected(original_element, argname, resolved)
-class ColumnsClauseImpl(
- _SelectIsNotFrom, _CoerceLiterals, RoleImpl, roles.ColumnsClauseRole
-):
+class ColumnsClauseImpl(_SelectIsNotFrom, _CoerceLiterals, RoleImpl):
+ __slots__ = ()
_coerce_consts = True
_coerce_numerics = True
)
-class ReturnsRowsImpl(RoleImpl, roles.ReturnsRowsRole):
- pass
+class ReturnsRowsImpl(RoleImpl):
+ __slots__ = ()
-class StatementImpl(_NoTextCoercion, RoleImpl, roles.StatementRole):
- pass
+class StatementImpl(_NoTextCoercion, RoleImpl):
+ __slots__ = ()
-class CoerceTextStatementImpl(_CoerceLiterals, RoleImpl, roles.StatementRole):
+class CoerceTextStatementImpl(_CoerceLiterals, RoleImpl):
+ __slots__ = ()
+
def _text_coercion(self, element, argname=None):
+ # TODO: this should emit deprecation warning,
+ # see deprecation warning in engine/base.py execute()
return elements.TextClause(element)
-class SelectStatementImpl(
- _NoTextCoercion, RoleImpl, roles.SelectStatementRole
-):
+class SelectStatementImpl(_NoTextCoercion, RoleImpl):
+ __slots__ = ()
+
def _implicit_coercions(
self, original_element, resolved, argname=None, **kw
):
pass
-class FromClauseImpl(
- _SelectIsNotFrom, _NoTextCoercion, RoleImpl, roles.FromClauseRole
-):
+class FromClauseImpl(_SelectIsNotFrom, _NoTextCoercion, RoleImpl):
+ __slots__ = ()
+
def _implicit_coercions(
self,
original_element,
self._raise_for_expected(original_element, argname, resolved)
-class StrictFromClauseImpl(FromClauseImpl, roles.StrictFromClauseRole):
+class StrictFromClauseImpl(FromClauseImpl):
+ __slots__ = ()
+
def _implicit_coercions(
self,
original_element,
self._raise_for_expected(original_element, argname, resolved)
-class AnonymizedFromClauseImpl(
- StrictFromClauseImpl, roles.AnonymizedFromClauseRole
-):
+class AnonymizedFromClauseImpl(StrictFromClauseImpl):
+ __slots__ = ()
+
def _post_coercion(self, element, flat=False, name=None, **kw):
return element.alias(name=name, flat=flat)
-class DMLSelectImpl(_NoTextCoercion, RoleImpl, roles.DMLSelectRole):
+class DMLSelectImpl(_NoTextCoercion, RoleImpl):
+ __slots__ = ()
+
def _implicit_coercions(
self, original_element, resolved, argname=None, **kw
):
self._raise_for_expected(original_element, argname, resolved)
-class CompoundElementImpl(
- _NoTextCoercion, RoleImpl, roles.CompoundElementRole
-):
+class CompoundElementImpl(_NoTextCoercion, RoleImpl):
+ __slots__ = ()
+
def _raise_for_expected(self, element, argname=None, resolved=None, **kw):
if isinstance(element, roles.FromClauseRole):
if element._is_subquery:
is_false(insp.has_table("t2", schema=config.test_schema))
is_false(insp.has_table("t3", schema=None))
+ @testing.provide_metadata
+ def test_option_on_execute(self):
+ self._fixture()
+
+ map_ = {
+ None: config.test_schema,
+ "foo": config.test_schema,
+ "bar": None,
+ }
+
+ metadata = MetaData()
+ t1 = Table("t1", metadata, Column("x", Integer))
+ t2 = Table("t2", metadata, Column("x", Integer), schema="foo")
+ t3 = Table("t3", metadata, Column("x", Integer), schema="bar")
+
+ with self.sql_execution_asserter(config.db) as asserter:
+ with config.db.connect() as conn:
+
+ execution_options = {"schema_translate_map": map_}
+ conn._execute_20(
+ t1.insert(), {"x": 1}, execution_options=execution_options
+ )
+ conn._execute_20(
+ t2.insert(), {"x": 1}, execution_options=execution_options
+ )
+ conn._execute_20(
+ t3.insert(), {"x": 1}, execution_options=execution_options
+ )
+
+ conn._execute_20(
+ t1.update().values(x=1).where(t1.c.x == 1),
+ execution_options=execution_options,
+ )
+ conn._execute_20(
+ t2.update().values(x=2).where(t2.c.x == 1),
+ execution_options=execution_options,
+ )
+ conn._execute_20(
+ t3.update().values(x=3).where(t3.c.x == 1),
+ execution_options=execution_options,
+ )
+
+ eq_(
+ conn._execute_20(
+ select([t1.c.x]), execution_options=execution_options
+ ).scalar(),
+ 1,
+ )
+ eq_(
+ conn._execute_20(
+ select([t2.c.x]), execution_options=execution_options
+ ).scalar(),
+ 2,
+ )
+ eq_(
+ conn._execute_20(
+ select([t3.c.x]), execution_options=execution_options
+ ).scalar(),
+ 3,
+ )
+
+ conn._execute_20(
+ t1.delete(), execution_options=execution_options
+ )
+ conn._execute_20(
+ t2.delete(), execution_options=execution_options
+ )
+ conn._execute_20(
+ t3.delete(), execution_options=execution_options
+ )
+
+ asserter.assert_(
+ CompiledSQL("INSERT INTO [SCHEMA__none].t1 (x) VALUES (:x)"),
+ CompiledSQL("INSERT INTO [SCHEMA_foo].t2 (x) VALUES (:x)"),
+ CompiledSQL("INSERT INTO [SCHEMA_bar].t3 (x) VALUES (:x)"),
+ CompiledSQL(
+ "UPDATE [SCHEMA__none].t1 SET x=:x WHERE "
+ "[SCHEMA__none].t1.x = :x_1"
+ ),
+ CompiledSQL(
+ "UPDATE [SCHEMA_foo].t2 SET x=:x WHERE "
+ "[SCHEMA_foo].t2.x = :x_1"
+ ),
+ CompiledSQL(
+ "UPDATE [SCHEMA_bar].t3 SET x=:x WHERE "
+ "[SCHEMA_bar].t3.x = :x_1"
+ ),
+ CompiledSQL("SELECT [SCHEMA__none].t1.x FROM [SCHEMA__none].t1"),
+ CompiledSQL("SELECT [SCHEMA_foo].t2.x FROM [SCHEMA_foo].t2"),
+ CompiledSQL("SELECT [SCHEMA_bar].t3.x FROM [SCHEMA_bar].t3"),
+ CompiledSQL("DELETE FROM [SCHEMA__none].t1"),
+ CompiledSQL("DELETE FROM [SCHEMA_foo].t2"),
+ CompiledSQL("DELETE FROM [SCHEMA_bar].t3"),
+ )
+
@testing.provide_metadata
def test_crud(self):
self._fixture()
eq_(opts, [expected])
- def test_execution_opts_invoke_illegal(self, connection):
- assert_raises_message(
- tsa.exc.InvalidRequestError,
- "The 'isolation_level' execution option is not supported "
- "at the per-statement level",
- connection.execute,
- select([1]),
- execution_options={"isolation_level": "AUTOCOMMIT"},
- )
-
- assert_raises_message(
- tsa.exc.InvalidRequestError,
- "The 'schema_translate_map' execution option is not supported "
- "at the per-statement level",
- connection.execute,
- select([1]),
- execution_options={"schema_translate_map": {}},
- )
-
def test_no_branching(self, connection):
assert_raises_message(
NotImplementedError,
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
