diff --git a/README.md b/README.md index 1ffb502..b5cd6c7 100644 --- a/README.md +++ b/README.md @@ -11,7 +11,23 @@ using it is extremely simple, say that you have a dataclass definition, just add the decorator `@datalite(db_name="db.db")` to the top of the definition, and the dataclass will now be bound to the file `db.db` -For example: +## Download and Install + +You can install `datalite` simply by + +```shell script +pip install datalite +``` + +Or you can clone the repository and run + +```shell script +python setup.py +``` + +Datalite has no dependencies! As it is built on Python 3.7+ standard library. Albeit, its tests require `unittest` library. + +## Datalite in Action ```python from dataclasses import dataclass @@ -30,7 +46,9 @@ table name `student` and rows `student_id`, `student_name` with datatypes integer and text, respectively. The default value for `student_name` is `John Smith`. -## Entry manipulation +##Basic Usage + +### Entry manipulation After creating an object traditionally, given that you used the `datalite` decorator, the object has three new methods: `.create_entry()`, `.update_entry()` @@ -81,4 +99,10 @@ The last two helper methods, `fetch_if(class_, condition)` fetches all the records of type `class_` that fit a certain condition. Here conditions must be written is SQL syntax. For easier, only one conditional checks, there is `fetch_equals(class_, field, value)` that checks the value of only one `field` -and returns the object whose `field` equals the provided `value`. \ No newline at end of file +and returns the object whose `field` equals the provided `value`. + +#### Pagination + +`datalite` also supports pagination on `fetch_if`, `fetch_all` and `fetch_where`, +you can specify `page` number and `element_count` for each page (default 10), for +these functions in order to get a subgroup of records. \ No newline at end of file diff --git a/datalite/fetch.py b/datalite/fetch.py index 3f945d6..12cf4b4 100644 --- a/datalite/fetch.py +++ b/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() diff --git a/setup.py b/setup.py index 88a746e..093f091 100644 --- a/setup.py +++ b/setup.py @@ -5,7 +5,7 @@ with open("README.md", "r") as fh: setuptools.setup( name="datalite", # Replace with your own username - version="0.5.1", + version="0.5.2", author="Ege Ozkan", author_email="egeemirozkan24@gmail.com", description="A small package that binds dataclasses to an sqlite database", diff --git a/test/main_tests.py b/test/main_tests.py index e1fbe0c..54378a8 100644 --- a/test/main_tests.py +++ b/test/main_tests.py @@ -3,6 +3,7 @@ from datalite import datalite from datalite.fetch import fetch_if, fetch_all, fetch_range, fetch_from, fetch_equals, fetch_where from sqlite3 import connect from dataclasses import dataclass, asdict +from math import floor from os import remove @@ -73,7 +74,6 @@ class DatabaseMain(unittest.TestCase): self.assertEqual(len(objects), init_len) - class DatabaseFetchCalls(unittest.TestCase): def setUp(self) -> None: self.objs = [FetchClass(1, 'a'), FetchClass(2, 'b'), FetchClass(3, 'b')] @@ -107,5 +107,25 @@ class DatabaseFetchCalls(unittest.TestCase): [obj.remove_entry() for obj in self.objs] +class DatabaseFetchPaginationCalls(unittest.TestCase): + def setUp(self) -> None: + self.objs = [FetchClass(i, f'{floor(i/10)}') for i in range(30)] + [obj.create_entry() for obj in self.objs] + + def testFetchAllPagination(self): + t_objs = fetch_all(FetchClass, 1, 10) + self.assertEqual(tuple(self.objs[:10]), t_objs) + + def testFetchWherePagination(self): + t_objs = fetch_where(FetchClass, 'str_', '0', 2, 5) + self.assertEqual(tuple(self.objs[5:10]), t_objs) + + def testFetchIfPagination(self): + t_objs = fetch_if(FetchClass, 'str_ = "0"', 1, 5) + self.assertEqual(tuple(self.objs[:5]), t_objs) + + def tearDown(self) -> None: + [obj.remove_entry() for obj in self.objs] + if __name__ == '__main__': unittest.main()