Add pagination support

datalite/fetch.py

@@ -3,6 +3,19 @@ from typing import List, Tuple, Any
 from .commons import _convert_sql_format
 
 
+def insert_pagination(query: str, page: int, element_count: int) -> str:
+    """
+    Insert the pagination arguments if page number is given.
+    :param query: Query to insert to
+    :param page: Page to get.
+    :param element_count: Element count in each page.
+    :return: The modified (or not) query.
+    """
+    if page:
+        query += f" ORDER BY obj_id LIMIT {element_count} OFFSET {(page - 1) * element_count}"
+    return query + ";"
+
+
 def is_fetchable(class_: type, obj_id: int) -> bool:
     """
     Check if a record is fetchable given its obj_id and
@@ -83,25 +96,27 @@ def _convert_record_to_object(class_: type, record: Tuple[Any], field_names: Lis
     return obj
 
 
-def fetch_if(class_: type, condition: str) -> tuple:
+def fetch_if(class_: type, condition: str, page: int = 0, element_count: int = 10) -> tuple:
     """
     Fetch all class_ type variables from the bound db,
     provided they fit the given condition
     :param class_: Class type to fetch.
     :param condition: Condition to check for.
+    :param page: Which page to retrieve, default all. (0 means closed).
+    :param element_count: Element count in each page.
     :return: A tuple of records that fit the given condition
     of given type class_.
     """
     table_name = class_.__name__.lower()
     with sql.connect(getattr(class_, 'db_path')) as con:
         cur: sql.Cursor = con.cursor()
-        cur.execute(f"SELECT * FROM {table_name} WHERE {condition};")
+        cur.execute(insert_pagination(f"SELECT * FROM {table_name} WHERE {condition}", page, element_count))
         records: list = cur.fetchall()
         field_names: List[str] = _get_table_cols(cur, table_name)
     return tuple(_convert_record_to_object(class_, record, field_names) for record in records)
 
 
-def fetch_where(class_: type, field: str, value: Any) -> tuple:
+def fetch_where(class_: type, field: str, value: Any, page: int = 0, element_count: int = 10) -> tuple:
     """
     Fetch all class_ type variables from the bound db,
     provided that the field of the records fit the
@@ -109,9 +124,11 @@ def fetch_where(class_: type, field: str, value: Any) -> tuple:
     :param class_: Class of the records.
     :param field: Field to check.
     :param value: Value to check for.
+    :param page: Which page to retrieve, default all. (0 means closed).
+    :param element_count: Element count in each page.
     :return: A tuple of the records.
     """
-    return fetch_if(class_, f"{field} = {_convert_sql_format(value)}")
+    return fetch_if(class_, f"{field} = {_convert_sql_format(value)}", page, element_count)
 
 
 def fetch_range(class_: type, range_: range) -> tuple:
@@ -125,10 +142,12 @@ def fetch_range(class_: type, range_: range) -> tuple:
     return tuple(fetch_from(class_, obj_id) for obj_id in range_ if is_fetchable(class_, obj_id))
 
 
-def fetch_all(class_: type) -> tuple:
+def fetch_all(class_: type, page: int = 0, element_count: int = 10) -> tuple:
     """
     Fetchall the records in the bound database.
     :param class_: Class of the records.
+    :param page: Which page to retrieve, default all. (0 means closed).
+    :param element_count: Element count in each page.
     :return: All the records of type class_ in
     the bound database as a tuple.
     """
@@ -139,7 +158,7 @@ def fetch_all(class_: type) -> tuple:
     with sql.connect(db_path) as con:
         cur: sql.Cursor = con.cursor()
         try:
-            cur.execute(f"SELECT * FROM {class_.__name__.lower()}")
+            cur.execute(insert_pagination(f"SELECT * FROM {class_.__name__.lower()}", page, element_count))
         except sql.OperationalError:
             raise TypeError(f"No record of type {class_.__name__.lower()}")
         records = cur.fetchall()