Commit ed54d135 authored by François Ferry's avatar François Ferry
Browse files

feat(rqlio): add get_entities route to mimic cubicweb.rset.ResultSet.entities

parent cc34b20da66f
Pipeline #87391 passed with stages
in 3 minutes and 16 seconds
......@@ -227,9 +227,57 @@ class AnonMultipartRqlIOController(BaseRqlIOController):
require_csrf = False
class GetEntitiesRqlIOController(JsonRqlIOController):
"""GetEntitiesRqlIOController allow to get entities attributes
from a RQL. Only attributes present in the RQL will be sent.
__regid__ = "rqlio_entities"
__select__ = match_http_method("POST")
def publish(self, rset=None):
self._cw.ajax_request = True
args = self.json()
rql_args, col = args
result = self.get_entities(col, *rql_args)
except (RemoteCallFailed, DirectResponse):
except Exception as exc:
raise RemoteCallFailed(exc_message(exc, self._cw.encoding))
if result is None:
return b""
return json_dumps(result).encode(self._cw.encoding)
def get_entities(self, col, *rql_args):
return self._get_entities(col, rql_args)
except Exception:
def _get_entities(self, col, rql_args):
output = []
for rql, args in rql_args:
if args is None:
args = {}
rewrite_args(args, output, self._cw.form)
entity_dict_list = []
for entity in self._cw.execute(rql, args).entities(col=col):
entity_dict = {"eid": entity.eid}
for attribute, value in entity.cw_attr_cache.items():
if attribute in rql:
entity_dict[attribute] = value
return output
class RQLIORewriter(SchemaBasedRewriter):
rules = [
(re.compile("/rqlio/schema"), rgx_action(controller="rqlio_schema")),
(re.compile(r"/rqlio/get_entities"), rgx_action(controller="rqlio_entities")),
rgx_action(controller="rqlio", formgroups=("version",)),
......@@ -169,6 +169,23 @@ class RqlIOTC(PyramidCWTest):
self.assertEqual(rset.variables, result["variables"])
self.assertEqual(rset.rows, result["rows"])
def test_variables_in_rqlio_get_entities(self):
ueid = self.admin_access._user.eid
queries = [
("Any X, L WHERE X is CWUser, X eid %(x)s, X login L", {"x": ueid}),
self.webapp.login(user="admin", password=self.password)
res =
params=json.dumps([queries, 0]),
headers={"Content-Type": "application/json"},
output = res.json
with self.admin_access.client_cnx() as cnx:
rset = cnx.execute(queries[0][0], queries[0][1])
self.assertEqual(, output[0][0].get("login"))
if __name__ == "__main__":
import unittest
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment