u3 = await async_session.get(User, 12)
is_(u3, None)
+
+ @async_test
+ async def test_get_one(self, async_session):
+ User = self.classes.User
+
+ u1 = await async_session.get_one(User, 7)
+ u2 = await async_session.get_one(User, 10)
+ u3 = await async_session.get_one(User, 7)
+
+ is_(u1, u3)
+ eq_(u1.name, "jack")
+ eq_(u2.name, "chuck")
+
+ with testing.expect_raises_message(
+ exc.NoResultFound,
+ "No row was found when one was required",
+ ):
+ await async_session.get_one(User, 12)
+
@async_test
async def test_force_a_lazyload(self, async_session):
"""test for #9298"""
toronto = WeatherLocation("North America", "Toronto")
london = WeatherLocation("Europe", "London")
dublin = WeatherLocation("Europe", "Dublin")
- brasilia = WeatherLocation("South America", "Brasila")
+ brasilia = WeatherLocation("South America", "Brasilia")
quito = WeatherLocation("South America", "Quito")
tokyo.reports.append(Report(80.0, id_=1))
newyork.reports.append(Report(75, id_=1))
t2 = sess.get(WeatherLocation, 1)
is_(t2, tokyo)
+ def test_get_one(self):
+ sess = self._fixture_data()
+ brasilia = sess.get_one(WeatherLocation, 6)
+ eq_(brasilia.id, 6)
+ eq_(brasilia.city, "Brasilia")
+
+ toronto = sess.get_one(WeatherLocation, 3)
+ eq_(toronto.id, 3)
+ eq_(toronto.city, "Toronto")
+
def test_get_explicit_shard(self):
sess = self._fixture_data()
tokyo = (
u2 = s.get(User, 7)
assert u is not u2
+ def test_get_one(self):
+ User = self.classes.User
+
+ s = fixture_session()
+ u = s.get_one(User, 7)
+ u2 = s.get_one(User, 7)
+ assert u is u2
+ s.expunge_all()
+ u2 = s.get_one(User, 7)
+ assert u is not u2
+
def test_get_synonym_direct_name(self, decl_base):
"""test #8753"""
is_true(called)
+ def test_get_one_no_result(self):
+ users, User = self.tables.users, self.classes.User
+ self.mapper_registry.map_imperatively(User, users)
+
+ sess = fixture_session()
+ user1 = User(id=1, name="u1")
+
+ sess.add(user1)
+ sess.commit()
+
+ u1 = sess.get_one(User, user1.id)
+ assert user1.name == u1.name
+
+ with expect_raises_message(
+ sa.exc.NoResultFound, "No row was found when one was required"
+ ):
+ sess.get_one(User, 2)
+
class SessionStateTest(_fixtures.FixtureTest):
run_inserts = None
def _public_session_methods(self):
Session = sa.orm.session.Session
- blocklist = {"begin", "query", "bind_mapper", "get", "bind_table"}
+ blocklist = {
+ "begin",
+ "query",
+ "bind_mapper",
+ "get",
+ "get_one",
+ "bind_table",
+ }
specials = {"__iter__", "__contains__"}
ok = set()
for name in dir(Session):