)
result = self.session.execute(
- statement, params, execution_options=execution_options, future=True
+ statement, params, execution_options=execution_options
)
if result._attributes.get("is_single_entity", False):
result = result.scalars()
% (orm_util.instance_str(instance), self.attr.key)
)
- result = sess.execute(self._statement, future=True)
+ result = sess.execute(self._statement)
result = result.scalars()
if result._attributes.get("filtered", False):
params=params,
execution_options=execution_options,
bind_arguments=bind_arguments,
- future=True,
)
.unique()
.scalars()
execution_options,
bind_arguments,
_add_event=skip_for_full_returning,
- future=True,
)
matched_rows = result.fetchall()
statement,
params,
execution_options={"_sa_orm_load_options": self.load_options},
- future=True,
)
# legacy: automatically set scalars, unique
delete_,
self._params,
execution_options={"synchronize_session": synchronize_session},
- future=True,
)
bulk_del.result = result
self.session.dispatch.after_bulk_delete(bulk_del)
upd,
self._params,
execution_options={"synchronize_session": synchronize_session},
- future=True,
)
bulk_ud.result = result
self.session.dispatch.after_bulk_update(bulk_ud)
_execution_options,
_bind_arguments,
_parent_execute_state=self,
- future=self._future,
)
@property
params=None,
execution_options=util.EMPTY_DICT,
bind_arguments=None,
- future=False,
_parent_execute_state=None,
_add_event=None,
**kw
Contents of this dictionary are passed to the
:meth:`.Session.get_bind` method.
- :param future:
- Use future style execution for this statement. This is
- the same effect as the :paramref:`_orm.Session.future` flag,
- except at the level of this single statement execution. See
- that flag for details.
-
- .. versionadded:: 1.4
-
:param mapper:
deprecated; use the bind_arguments dictionary
"""
statement = coercions.expect(roles.CoerceTextStatementRole, statement)
- future = future or self.future
-
if not bind_arguments:
bind_arguments = kw
elif kw:
)
else:
bind_arguments.setdefault("clause", statement)
- if future:
- # not sure if immutabledict is working w/ this syntax
- # execution_options =
- # execution_options.union(future_result=True)
- execution_options = execution_options.union(
- {"future_result": True}
- )
+ execution_options = execution_options.union(
+ {"future_result": True}
+ )
if _parent_execute_state:
events_todo = _parent_execute_state._remaining_events()
if result:
return result
- # TODO: coverage for this pattern
statement = orm_exec_state.statement
execution_options = orm_exec_state.local_execution_options
)
result = session.execute(
- stmt, params, future=True, execution_options=execution_options
+ stmt, params, execution_options=execution_options
)
result = result.unique().scalars().all()
for key in chunk
]
},
- future=True,
).unique()
}
data = collections.defaultdict(list)
for k, v in itertools.groupby(
context.session.execute(
- q, params={"primary_keys": primary_keys}, future=True
+ q, params={"primary_keys": primary_keys}
).unique(),
lambda x: x[0],
):
q = sess.query(User).filter(User.id == 7).set_cache_key("user7")
eq_(
- sess.execute(q, future=True).all(),
- [(User(id=7, addresses=[Address(id=1)]),)],
+ sess.execute(q).all(), [(User(id=7, addresses=[Address(id=1)]),)],
)
eq_(list(q.cache), ["user7"])
eq_(
- sess.execute(q, future=True).all(),
- [(User(id=7, addresses=[Address(id=1)]),)],
+ sess.execute(q).all(), [(User(id=7, addresses=[Address(id=1)]),)],
)
def test_use_w_baked(self):
mapper(User, users)
- s = Session(testing.db, future=True)
+ s = Session(testing.db)
q = testing.resolve_lambda(test_case, **locals())
row = s.execute(q.order_by(User.id)).first()
- # old style row
- assert "jack" not in row
- assert "jack" in tuple(row)
-
- row = s.execute(q.order_by(User.id), future=True).first()
-
- # new style row - not sure what to do here w/ future yet
+ # s.execute() is now new style row
assert "jack" in row
- assert "jack" in tuple(row)
def test_entity_mapping_access(self):
User, users = self.classes.User, self.tables.users
def test_populate_existing_future(self):
User, Address = self.classes.User, self.classes.Address
- s = Session(testing.db, future=True, autoflush=False)
+ s = Session(testing.db, autoflush=False)
userlist = s.query(User).all()
stmt = select(User).execution_options(
populate_existing=True, autoflush=False, yield_per=10
)
- s = Session(testing.db, future=True)
+ s = Session(testing.db)
m1 = mock.Mock()
def test_select_star_future(self):
User = self.classes.User
- sess = Session(testing.db, future=True)
+ sess = Session(testing.db)
eq_(
sess.execute(
select(User).from_statement(
# ordering doesn't matter
User = self.classes.User
- s = create_session(testing.db, future=True)
+ s = create_session(testing.db)
q = select(User).from_statement(
text(
"select name, 27 as foo, id as users_id from users order by id"
User = self.classes.User
Address = self.classes.Address
- s = create_session(testing.db, future=True)
+ s = create_session(testing.db)
q = select(User, Address).from_statement(
text(
"select users.name AS users_name, users.id AS users_id, "
User = self.classes.User
Address = self.classes.Address
- s = create_session(testing.db, future=True)
+ s = create_session(testing.db)
q = (
select(User)
.from_statement(
User = self.classes.User
Address = self.classes.Address
- s = create_session(testing.db, future=True)
+ s = create_session(testing.db)
q = (
select(User)
.from_statement(
def test_whereclause_future(self):
User = self.classes.User
- s = create_session(testing.db, future=True)
+ s = create_session(testing.db)
eq_(
s.execute(select(User).filter(text("id in (8, 9)")))
.scalars()
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
