diff --git a/libervia/backend/core/exceptions.py b/libervia/backend/core/exceptions.py index 9d817182..27126069 100644 --- a/libervia/backend/core/exceptions.py +++ b/libervia/backend/core/exceptions.py @@ -78,6 +78,10 @@ class DataError(Exception): pass +class BadRequest(DataError): + """Request contains invalid parameters""" + + class ExternalRequestError(Exception): """Request to third party server failed""" diff --git a/libervia/backend/memory/sqla.py b/libervia/backend/memory/sqla.py index 08293355..c2132f6b 100644 --- a/libervia/backend/memory/sqla.py +++ b/libervia/backend/memory/sqla.py @@ -1436,15 +1436,11 @@ class Storage: ) @aio - async def update_pubsub_node( - self, node: PubsubNode, values: dict - ) -> None: + async def update_pubsub_node(self, node: PubsubNode, values: dict) -> None: async with self.session() as session: async with session.begin(): await session.execute( - update(PubsubNode) - .filter_by(id=node.id) - .values(**values) + update(PubsubNode).filter_by(id=node.id).values(**values) ) @aio @@ -1589,42 +1585,170 @@ class Storage: await session.execute(stmt) await session.commit() + async def _get_items_rsm( + self, + node: PubsubNode, + max_items: int, + before: str | None, + after: str | None, + from_index: int | None, + order: list, + metadata: dict, + ) -> tuple[list[PubsubItem], dict]: + """Get Pubsub Items using RSM + + @param node: retrieve items from this node (must be synchronised) + @param max_items: maximum number of items to retrieve + @param before: get items which are before the item with this name + @param after: get items which are after the item with this name + @param from_index: get items starting from this index (None if before/after used) + @param order: SQL order clauses + @param metadata: metadata dict to populate with RSM information + @return: tuple of (items list, metadata dict) + @raise exceptions.NotFound: if before or after references a non-existent item. + """ + # CTE to have result row numbers + row_num_q = select( + PubsubItem.id, + PubsubItem.name, + # row_number starts from 1, but RSM index must start from 0 + (func.row_number().over(order_by=order) - 1).label("item_index"), + ).filter_by(node_id=node.id) + + row_num_cte = row_num_q.cte() + + if max_items != 0: + # As we can't simply use PubsubItem.id when we order by modification, we need + # to use row number. + + stmt = ( + select(row_num_cte.c.item_index, PubsubItem) + .join(row_num_cte, PubsubItem.id == row_num_cte.c.id) + .limit(max_items) + ) + + if before == "": + # We look for the last page, we need to use descending order for the + # query, and reverse later. + stmt = stmt.order_by(row_num_cte.c.item_index.desc()) + elif before or after: + # We use an item name as limit, we find which row number correspond to it. + item_name = before or after + row_num_limit_q = ( + select(row_num_cte.c.item_index).where( + row_num_cte.c.name == item_name + ) + ).scalar_subquery() + + # Check if the reference item exists by executing the subquery + async with self.session() as session: + ref_index_result = await session.execute(row_num_limit_q) + ref_index = ref_index_result.scalar_one_or_none() + + if ref_index is None: + raise exceptions.NotFound( + f"Item {item_name!r} not found in result set" + ) + + if before: + # We order by descending for before, we'll reverse after the + # retrieved items. + stmt = stmt.where(row_num_cte.c.item_index < ref_index).order_by( + row_num_cte.c.item_index.desc() + ) + elif after: + stmt = stmt.where(row_num_cte.c.item_index > ref_index).order_by( + row_num_cte.c.item_index.asc() + ) + else: + # If we reach here, from_index must be set. + assert from_index is not None + stmt = stmt.where(row_num_cte.c.item_index >= from_index).order_by( + row_num_cte.c.item_index.asc() + ) + + async with self.session() as session: + if max_items == 0: + items = result = [] + else: + result = await session.execute(stmt) + result = result.all() + if before is not None: + result.reverse() + items = [row[-1] for row in result] + rows_count = ( + await session.execute(row_num_q.with_only_columns(count())) + ).scalar_one() + + try: + index = result[0][0] + except IndexError: + index = None + + try: + first = result[0][1].name + except IndexError: + first = None + last = None + else: + last = result[-1][1].name + + metadata["rsm"] = { + k: v + for k, v in { + "index": index, + "count": rows_count, + "first": first, + "last": last, + }.items() + if v is not None + } + metadata["complete"] = (index or 0) + len(result) == rows_count + + return items, metadata + @aio async def get_items( self, node: PubsubNode, - max_items: Optional[int] = None, - item_ids: Optional[list[str]] = None, - before: Optional[str] = None, - after: Optional[str] = None, - from_index: Optional[int] = None, - order_by: Optional[List[str]] = None, + max_items: int | None = None, + item_ids: list[str] | None = None, + before: str | None = None, + after: str | None = None, + from_index: int | None = None, + order_by: list[str] | None = None, desc: bool = True, force_rsm: bool = False, - ) -> Tuple[List[PubsubItem], dict]: + ) -> tuple[list[PubsubItem], dict]: """Get Pubsub Items from cache @param node: retrieve items from this node (must be synchronised) @param max_items: maximum number of items to retrieve - @param before: get items which are before the item with this name in given order - empty string is not managed here, use desc order to reproduce RSM - behaviour. + @param item_ids: list of specific item IDs to retrieve. + Cannot be used with RSM parameters (before, after, from_index). + @param before: get items which are before the item with this name in given order. + Use empty string (before="") to request the last page. @param after: get items which are after the item with this name in given order @param from_index: get items with item index (as defined in RSM spec) starting from this number @param order_by: sorting order of items (one of C.ORDER_BY_*) - @param desc: direction or ordering + @param desc: direction of ordering. + If True, ordering is descending. @param force_rsm: if True, force the use of RSM worklow. - RSM workflow is automatically used if any of before, after or - from_index is used, but if only RSM max_items is used, it won't be - used by default. This parameter let's use RSM workflow in this - case. Note that in addition to RSM metadata, the result will not be - the same (max_items without RSM will returns most recent items, - i.e. last items in modification order, while max_items with RSM - will return the oldest ones (i.e. first items in modification - order). - to be used when max_items is used from RSM + RSM workflow is automatically used if any of before, after or from_index is + used, but if only RSM max_items is used, it won't be used by default. This + parameter let use RSM workflow in this case. Note that in addition to RSM + metadata, the result will not be the same (max_items without RSM will returns + most recent items, i.e. last items in modification order, while max_items with + RSM will return the oldest ones (i.e. first items in modification order). + To be used when max_items is used from RSM + @raise exceptions.BadRequest: Something is wrong with the request. + @raise exceptions.NotFound: if before or after references a non-existent item. """ + if max_items is not None and max_items < 0: + raise exceptions.BadRequest("max_items must be non-negative") + if from_index is not None and from_index < 0: + raise exceptions.BadRequest("from_index must be non-negative") metadata = { "service": node.service, @@ -1638,17 +1762,7 @@ class Storage: if max_items is None: max_items = 20 - use_rsm = any((before, after, from_index is not None)) - if force_rsm and not use_rsm: - # - use_rsm = True - from_index = 0 - - stmt = select(PubsubItem).filter_by(node_id=node.id).limit(max_items) - - if item_ids is not None: - stmt = stmt.where(PubsubItem.name.in_(item_ids)) - + # ordering if not order_by: order_by = [C.ORDER_BY_MODIFICATION] @@ -1667,94 +1781,33 @@ class Storage: else: raise exceptions.InternalError(f"Unknown order type {order_type!r}") - stmt = stmt.order_by(*order) + use_rsm = any((before is not None, after, from_index is not None)) + if force_rsm and not use_rsm: + use_rsm = True + from_index = 0 + + # item_ids with RSM doesn't make sense + if item_ids is not None and use_rsm: + raise exceptions.BadRequest( + "item_ids cannot be used with RSM parameters" + ) if use_rsm: - # CTE to have result row numbers - row_num_q = select( - PubsubItem.id, - PubsubItem.name, - # row_number starts from 1, but RSM index must start from 0 - (func.row_number().over(order_by=order) - 1).label("item_index"), - ).filter_by(node_id=node.id) + return await self._get_items_rsm( + node, max_items, before, after, from_index, order, metadata + ) - row_num_cte = row_num_q.cte() + # Query without RSM + stmt = select(PubsubItem).filter_by(node_id=node.id).limit(max_items) + stmt = stmt.order_by(*order) - if max_items > 0: - # as we can't simply use PubsubItem.id when we order by modification, - # we need to use row number - item_name = before or after - row_num_limit_q = ( - select(row_num_cte.c.item_index).where( - row_num_cte.c.name == item_name - ) - ).scalar_subquery() - - stmt = ( - select(row_num_cte.c.item_index, PubsubItem) - .join(row_num_cte, PubsubItem.id == row_num_cte.c.id) - .limit(max_items) - ) - if before: - stmt = stmt.where( - row_num_cte.c.item_index < row_num_limit_q - ).order_by(row_num_cte.c.item_index.desc()) - elif after: - stmt = stmt.where( - row_num_cte.c.item_index > row_num_limit_q - ).order_by(row_num_cte.c.item_index.asc()) - else: - stmt = stmt.where(row_num_cte.c.item_index >= from_index).order_by( - row_num_cte.c.item_index.asc() - ) - # from_index is used - - async with self.session() as session: - if max_items == 0: - items = result = [] - else: - result = await session.execute(stmt) - result = result.all() - if before: - result.reverse() - items = [row[-1] for row in result] - rows_count = ( - await session.execute(row_num_q.with_only_columns(count())) - ).scalar_one() - - try: - index = result[0][0] - except IndexError: - index = None - - try: - first = result[0][1].name - except IndexError: - first = None - last = None - else: - last = result[-1][1].name - - metadata["rsm"] = { - k: v - for k, v in { - "index": index, - "count": rows_count, - "first": first, - "last": last, - }.items() - if v is not None - } - metadata["complete"] = (index or 0) + len(result) == rows_count - - return items, metadata + if item_ids is not None: + stmt = stmt.where(PubsubItem.name.in_(item_ids)) async with self.session() as session: result = await session.execute(stmt) result = result.scalars().all() - if desc: - result.reverse() return result, metadata def _get_sqlite_path(self, path: List[Union[str, int]]) -> str: diff --git a/libervia/backend/plugins/plugin_pubsub_cache.py b/libervia/backend/plugins/plugin_pubsub_cache.py index b58e1558..e85b8b97 100644 --- a/libervia/backend/plugins/plugin_pubsub_cache.py +++ b/libervia/backend/plugins/plugin_pubsub_cache.py @@ -505,20 +505,15 @@ class PubsubCache: node, max_items=max_items, order_by=extra.get(C.KEY_ORDER_BY) ) else: - desc = False - if rsm_request.before == "": - before = None - desc = True - else: - before = rsm_request.before + # RSM request pubsub_items, metadata = await self.host.memory.storage.get_items( node, max_items=rsm_request.max, - before=before, + before=rsm_request.before, after=rsm_request.after, from_index=rsm_request.index, order_by=extra.get(C.KEY_ORDER_BY), - desc=desc, + desc=False, force_rsm=True, )