[rset] New method: ResultSet.one()
This method will return exactly one entity while enforcing its existence and unicity.
The idea is shamelessly borowed from SQLAlchemy Query.one().
Closes #3352314
[jcr: use len(self) instead of len(self.rows)]
--- a/_exceptions.py Tue Dec 10 12:36:50 2013 +0100
+++ b/_exceptions.py Wed Dec 11 17:52:54 2013 +0100
@@ -134,6 +134,15 @@
a non final entity
"""
+class MultipleResultsError(CubicWebRuntimeError):
+ """raised when ResultSet.one() is called on a resultset with multiple rows
+ of multiple columns.
+ """
+
+class NoResultError(CubicWebRuntimeError):
+ """raised when no result is found but at least one is expected.
+ """
+
class UndoTransactionException(QueryError):
"""Raised when undoing a transaction could not be performed completely.
--- a/rset.py Tue Dec 10 12:36:50 2013 +0100
+++ b/rset.py Wed Dec 11 17:52:54 2013 +0100
@@ -23,7 +23,7 @@
from rql import nodes, stmts
-from cubicweb import NotAnEntity
+from cubicweb import NotAnEntity, NoResultError, MultipleResultsError
class ResultSet(object):
@@ -437,6 +437,25 @@
raise NotAnEntity(etype)
return self._build_entity(row, col)
+ def one(self, col=0):
+ """Retrieve exactly one entity from the query.
+
+ If the result set is empty, raises :exc:`NoResultError`.
+ If the result set has more than one row, raises
+ :exc:`MultipleResultsError`.
+
+ :type col: int
+ :param col: The column localising the entity in the unique row
+
+ :return: the partially initialized `Entity` instance
+ """
+ if len(self) == 1:
+ return self.get_entity(0, col)
+ elif len(self) == 0:
+ raise NoResultError("No row was found for one()")
+ else:
+ raise MultipleResultsError("Multiple rows were found for one()")
+
def _build_entity(self, row, col):
"""internal method to get a single entity, returns a partially
initialized Entity instance.
--- a/test/unittest_rset.py Tue Dec 10 12:36:50 2013 +0100
+++ b/test/unittest_rset.py Wed Dec 11 17:52:54 2013 +0100
@@ -28,6 +28,8 @@
from cubicweb.devtools.testlib import CubicWebTC
from cubicweb.rset import NotAnEntity, ResultSet, attr_desc_iterator
+from cubicweb import NoResultError, MultipleResultsError
+
def pprelcachedict(d):
res = {}
@@ -368,6 +370,39 @@
attr = etype == 'Bookmark' and 'title' or 'name'
self.assertEqual(entity.cw_attr_cache[attr], n)
+ def test_one(self):
+ self.request().create_entity('CWUser', login=u'cdevienne',
+ upassword=u'cdevienne',
+ surname=u'de Vienne',
+ firstname=u'Christophe')
+ e = self.execute('Any X WHERE X login "cdevienne"').one()
+
+ self.assertEqual(e.surname, u'de Vienne')
+
+ e = self.execute(
+ 'Any X, N WHERE X login "cdevienne", X surname N').one()
+ self.assertEqual(e.surname, u'de Vienne')
+
+ e = self.execute(
+ 'Any N, X WHERE X login "cdevienne", X surname N').one(col=1)
+ self.assertEqual(e.surname, u'de Vienne')
+
+ def test_one_no_rows(self):
+ with self.assertRaises(NoResultError):
+ self.execute('Any X WHERE X login "patanok"').one()
+
+ def test_one_multiple_rows(self):
+ self.request().create_entity(
+ 'CWUser', login=u'cdevienne', upassword=u'cdevienne',
+ surname=u'de Vienne', firstname=u'Christophe')
+
+ self.request().create_entity(
+ 'CWUser', login=u'adim', upassword='adim', surname=u'di mascio',
+ firstname=u'adrien')
+
+ with self.assertRaises(MultipleResultsError):
+ self.execute('Any X WHERE X is CWUser').one()
+
def test_related_entity_optional(self):
e = self.request().create_entity('Bookmark', title=u'aaaa', path=u'path')
rset = self.execute('Any B,U,L WHERE B bookmarked_by U?, U login L')