:param ident: A scalar or tuple value representing
the primary key. For a composite primary key,
- the order of identifiers corresponds in most cases
- to that of the mapped :class:`.Table` object's
- primary key columns. For a :func:`.mapper` that
- was given the ``primary key`` argument during
- construction, the order of identifiers corresponds
- to the elements present in this collection.
+ a dictionary with keys being primary key column names.
:return: The object instance, or ``None``.
if hasattr(primary_key_identity, "__composite_values__"):
primary_key_identity = primary_key_identity.__composite_values__()
- primary_key_identity = util.to_list(primary_key_identity)
-
mapper = self._only_full_mapper_zero("get")
+ is_dict = isinstance(primary_key_identity, dict)
+ if not is_dict:
+ primary_key_identity = util.to_list(primary_key_identity)
+
if len(primary_key_identity) != len(mapper.primary_key):
raise sa_exc.InvalidRequestError(
"Incorrect number of values in identifier to formulate "
% ",".join("'%s'" % c for c in mapper.primary_key)
)
+ if is_dict:
+ try:
+ primary_key_identity = list(
+ primary_key_identity[prop.key]
+ for prop in mapper._identity_key_props
+ )
+ except KeyError:
+ raise sa_exc.InvalidRequestError(
+ "Incorrect names of values in identifier to formulate "
+ "primary key for query.get(); primary key attribute names are %s"
+ % ",".join("'%s'" % prop.key for prop in mapper._identity_key_props)
+ )
+
if (
not self._populate_existing
and not mapper.always_refresh
class GetTest(QueryTest):
+ def test_get_composite_pk_keyword_based_no_result(self):
+ CompositePk = self.classes.CompositePk
+
+ s = Session()
+ is_(s.query(CompositePk).get({"i": 100, "j": 100}), None)
+
+ def test_get_composite_pk_keyword_based_result(self):
+ CompositePk = self.classes.CompositePk
+
+ s = Session()
+ one_two = s.query(CompositePk).get({"i": 1, "j": 2})
+ eq_(one_two.i, 1)
+ eq_(one_two.j, 2)
+ eq_(one_two.k, 3)
+
+ def test_get_composite_pk_keyword_based_wrong_keys(self):
+ CompositePk = self.classes.CompositePk
+
+ s = Session()
+ q = s.query(CompositePk)
+ assert_raises(sa_exc.InvalidRequestError, q.get, {"i": 1, "k": 2})
+
+ def test_get_composite_pk_keyword_based_too_few_keys(self):
+ CompositePk = self.classes.CompositePk
+
+ s = Session()
+ q = s.query(CompositePk)
+ assert_raises(sa_exc.InvalidRequestError, q.get, {"i": 1})
+
+ def test_get_composite_pk_keyword_based_too_many_keys(self):
+ CompositePk = self.classes.CompositePk
+
+ s = Session()
+ q = s.query(CompositePk)
+ assert_raises(sa_exc.InvalidRequestError, q.get, {"i": 1,"j": '2',"k": 3})
+
def test_get(self):
User = self.classes.User