Coverage for /home/runner/work/viur-core/viur-core/viur/src/viur/core/pagination.py: 0%

1from hashlib import sha256 

2import typing as t 

3 

4from viur.core import db, utils 

5 

6 

7class Pagination: 

8 """ 

9 This module provides efficient pagination for a small specified set of queries. 

10 The datastore does not provide an efficient method for skipping N number of entities. This prevents 

11 the usual navigation over multiple pages (in most cases - like a google search - the user expects a list 

12 of pages (e.g. 1-10) on the bottom of each page with direct access to these pages). With the datastore and it's 

13 cursors, the most we can provide is a next-page & previous-page link using cursors. This module provides an 

14 efficient method to provide these direct-access page links under the condition that only a few, known-in-advance 

15 queries will be run. This is typically the case for forums, where there is only one query per thread (it's posts 

16 ordered by creation date) and one for the threadlist (it's threads, ordered by changedate). 

17 

18 To use this module, create an instance of this index-manager on class-level (setting page_size & max_pages). 

19 Then call :meth:get_pages with the query you want to retrieve the cursors for the individual pages for. This 

20 will return one start-cursor per available page that can then be used to create urls that point to the specific 

21 page. When the entities returned by the query change (eg a new post is added), call :meth:refresh_index for 

22 each affected query. 

23 

24 .. Note:: 

25 

26 The refreshAll Method is missing - intentionally. Whenever data changes you have to call 

27 refresh_index for each affected Index. As long as you can name them, their number is 

28 limited and this module can be efficiently used. 

29 """ 

30 

31 _db_type = "viur_pagination" 

32 

33 def __init__(self, page_size: int = 10, max_pages: int = 100): 

34 """ 

35 :param page_size: How many entities shall fit on one page 

36 :param max_pages: How many pages are build. 

37 Items become unreachable if the amount of items exceeds 

38 page_size*max_pages (i.e. if a forum-thread has more than 

39 page_size*max_pages Posts, Posts after that barrier won't show up). 

40 """ 

41 self.page_size = page_size 

42 self.max_pages = max_pages 

43 

44 def key_from_query(self, query: db.Query) -> str: 

45 """ 

46 Derives a unique Database-Key from a given query. 

47 This Key is stable regardless in which order the filter have been applied 

48 

49 :param query: Query to derive key from 

50 :returns: The unique key derived 

51 """ 

52 if not isinstance(query, db.Query): 

53 raise TypeError( 

54 f"Expected a query. Got {query!r} of type {type(query)!r}") 

55 if isinstance(query.queries, list): 

56 raise NotImplementedError # TODO: Can we handle this case? How? 

57 elif query.queries is None: 

58 raise ValueError("The query has no queries!") 

59 orig_filter = [(x, y) for x, y in query.queries.filters.items()] 

60 for field, sort_order in query.queries.orders: 

61 orig_filter.append((f"{field} =", sort_order)) 

62 if query.queries.limit: 

63 orig_filter.append(("__pagesize =", self.page_size)) 

64 orig_filter.sort(key=lambda sx: sx[0]) 

65 filter_key = "".join(f"{x}{y}" for x, y in orig_filter) 

66 return sha256(filter_key.encode()).hexdigest() 

67 

68 def get_or_build_index(self, orig_query: db.Query) -> list[str]: 

69 """ 

70 Builds a specific index based on origQuery 

71 AND local variables (self.page_size and self.max_pages) 

72 Returns a list of starting-cursors for each page. 

73 You probably shouldn't call this directly. Use cursor_for_query. 

74 

75 :param orig_query: Query to build the index for 

76 """ 

77 key = self.key_from_query(orig_query) 

78 

79 # We don't have it cached - try to load it from DB 

80 index = db.Get(db.Key(self._db_type, key)) 

81 if index is not None: 

82 return index["data"] 

83 

84 # We don't have this index yet... Build it 

85 query = orig_query.clone() 

86 cursors = [None] 

87 while len(cursors) < self.max_pages: 

88 query_res = query.run(limit=self.page_size) 

89 if not query_res: 

90 # This cursor returns no data, remove it 

91 cursors.pop() 

92 break 

93 if query.getCursor() is None: 

94 # We reached the end of our data 

95 break 

96 cursors.append(query.getCursor()) 

97 query.setCursor(query.getCursor()) 

98 

99 entry = db.Entity(db.Key(self._db_type, key)) 

100 entry["data"] = cursors 

101 entry["creationdate"] = utils.utcNow() 

102 db.Put(entry) 

103 return cursors 

104 

105 def cursor_for_query(self, query: db.Query, page: int) -> t.Optional[str]: 

106 """ 

107 Returns the starting-cursor for the given query and page using an index. 

108 

109 .. WARNING: 

110 

111 Make sure the maximum count of different queries are limited! 

112 If an attacker can choose the query freely, he can consume a lot 

113 datastore quota per request! 

114 

115 :param query: Query to get the cursor for 

116 :param page: Page the user wants to retrieve 

117 :returns: Cursor or None if no cursor is applicable 

118 """ 

119 page = int(page) 

120 pages = self.get_or_build_index(query) 

121 if 0 <= page < len(pages): 

122 return pages[page] 

123 else: 

124 return None 

125 

126 def get_pages(self, query: db.Query) -> list[str]: 

127 """ 

128 Returns a list of all starting-cursors for this query. 

129 The first element is always None as the first page doesn't 

130 have any start-cursor 

131 """ 

132 return self.get_or_build_index(query) 

133 

134 def refresh_index(self, query: db.Query) -> None: 

135 """ 

136 Refreshes the Index for the given query 

137 (Actually it removes it from the db, so it gets rebuild on next use) 

138 

139 :param query: Query for which the index should be refreshed 

140 """ 

141 key = self.key_from_query(query) 

142 db.Delete(db.Key(self._db_type, key))