From af67d4b19d54ed39242e95d6e9f000bb05a9a74a Mon Sep 17 00:00:00 2001 From: LeXofLeviafan Date: Fri, 16 Dec 2022 02:39:10 +0100 Subject: [PATCH] [jarun#648][jarun#654] improved return values of BukuDb methods --- buku | 114 ++++++++++++++++++++----------------------- bukuserver/api.py | 40 +++++++-------- bukuserver/views.py | 8 +-- tests/test_bukuDb.py | 10 ++-- 4 files changed, 80 insertions(+), 92 deletions(-) diff --git a/buku b/buku index 23eb6fb5..21a52e93 100755 --- a/buku +++ b/buku @@ -44,7 +44,7 @@ import webbrowser from enum import Enum from itertools import chain from subprocess import DEVNULL, PIPE, Popen -from typing import Any, Dict, Iterable, List, Optional, Tuple +from typing import Any, Dict, List, Optional, Tuple import urllib3 from bs4 import BeautifulSoup @@ -512,6 +512,14 @@ class BukuDb: return (conn, cur) + def _fetch(self, query: str, *args) -> List[BookmarkVar]: + self.cur.execute(query, args) + return self.cur.fetchall() + + def _fetch_first(self, query: str, *args) -> Optional[BookmarkVar]: + rows = self._fetch(query + ' LIMIT 1', *args) + return rows[0] if rows else None + def get_rec_all(self): """Get all the bookmarks in the database. @@ -521,8 +529,7 @@ class BukuDb: A list of tuples representing bookmark records. """ - self.cur.execute('SELECT * FROM bookmarks') - return self.cur.fetchall() + return self._fetch('SELECT * FROM bookmarks') def get_rec_by_id(self, index: int) -> Optional[BookmarkVar]: """Get a bookmark from database by its ID. @@ -534,13 +541,11 @@ class BukuDb: Returns ------- - tuple or None + BookmarkVar or None Bookmark data, or None if index is not found. """ - self.cur.execute('SELECT * FROM bookmarks WHERE id = ? LIMIT 1', (index,)) - resultset = self.cur.fetchall() - return resultset[0] if resultset else None + return self._fetch_first('SELECT * FROM bookmarks WHERE id = ?', index) def get_rec_id(self, url): """Check if URL already exists in DB. @@ -553,12 +558,11 @@ class BukuDb: Returns ------- int - DB index, or -1 if URL not found in DB. + DB index, or None if URL not found in DB. """ - self.cur.execute('SELECT id FROM bookmarks WHERE URL = ? LIMIT 1', (url,)) - resultset = self.cur.fetchall() - return resultset[0][0] if resultset else -1 + row = self._fetch_first('SELECT * FROM bookmarks WHERE url = ?', url) + return row and row[0] def get_max_id(self) -> int: """Fetch the ID of the last record. @@ -566,12 +570,11 @@ class BukuDb: Returns ------- int - ID of the record if any record exists, else -1. + ID of the record if any record exists, else None. """ - self.cur.execute('SELECT MAX(id) from bookmarks') - resultset = self.cur.fetchall() - return -1 if resultset[0][0] is None else resultset[0][0] + self.cur.execute('SELECT MAX(id) FROM bookmarks') + return self.cur.fetchall()[0][0] def add_rec( self, @@ -607,19 +610,19 @@ class BukuDb: Returns ------- int - DB index of new bookmark on success, -1 on failure. + DB index of new bookmark on success, None on failure. """ # Return error for empty URL if not url or url == '': LOGERR('Invalid URL') - return -1 + return None # Ensure that the URL does not exist in DB already id = self.get_rec_id(url) - if id != -1: + if id: LOGERR('URL [%s] already exists at index %d', url, id) - return -1 + return None if fetch: # Fetch data @@ -660,7 +663,7 @@ class BukuDb: return self.cur.lastrowid except Exception as e: LOGERR('add_rec(): %s', e) - return -1 + return None def append_tag_at_index(self, index, tags_in, delay_commit=False): """Append tags to bookmark tagset at index. @@ -1137,7 +1140,7 @@ class BukuDb: if index == -1: # Edit the last records index = self.get_max_id() - if index == -1: + if not index: LOGERR('Empty database') return False @@ -1192,11 +1195,10 @@ class BukuDb: q0 += ')' try: - self.cur.execute(q0, []) + return self._fetch(q0) except sqlite3.OperationalError as e: LOGERR(e) - return None - return self.cur.fetchall() + return [] def searchdb( self, @@ -1204,7 +1206,7 @@ class BukuDb: all_keywords: Optional[bool] = False, deep: Optional[bool] = False, regex: Optional[bool] = False - ) -> Optional[Iterable[Any]]: + ) -> List[BookmarkVar]: """Search DB for entries where tags, URL, or title fields match keywords. Parameters @@ -1221,11 +1223,11 @@ class BukuDb: Returns ------- - list or None - List of search results, or None if no matches. + list + List of search results. """ if not keywords: - return None + return [] # Deep query string q1 = ("(tags LIKE ('%' || ? || '%') OR " @@ -1250,7 +1252,7 @@ class BukuDb: qargs += (token, token, token, token,) if not qargs: - return None + return [] q0 = q0[:-3] + ' AS score FROM bookmarks WHERE score > 0 ORDER BY score DESC)' elif all_keywords: @@ -1279,7 +1281,7 @@ class BukuDb: qargs += (token, token, token, token,) if not qargs: - return None + return [] q0 = q0[:-4] q0 += 'ORDER BY id ASC' @@ -1303,24 +1305,22 @@ class BukuDb: qargs += (token, token, token, token,) if not qargs: - return None + return [] q0 = q0[:-3] + ' AS score FROM bookmarks WHERE score > 0 ORDER BY score DESC)' else: LOGERR('Invalid search option') - return None + return [] LOGDBG('query: "%s", args: %s', q0, qargs) try: - self.cur.execute(q0, qargs) + return self._fetch(q0, *qargs) except sqlite3.OperationalError as e: LOGERR(e) - return None - - return self.cur.fetchall() + return [] - def search_by_tag(self, tags: Optional[str]) -> Optional[List[BookmarkVar]]: + def search_by_tag(self, tags: Optional[str]) -> List[BookmarkVar]: """Search bookmarks for entries with given tags. Parameters @@ -1334,18 +1334,18 @@ class BukuDb: Returns ------- - list or None - List of search results, or None if no matches. + list + List of search results. """ LOGDBG(tags) if tags is None or tags == DELIM or tags == '': - return None + return [] tags_, search_operator, excluded_tags = prep_tag_search(tags) if search_operator is None: LOGERR("Cannot use both '+' and ',' in same search") - return None + return [] LOGDBG('tags: %s', tags_) LOGDBG('search_operator: %s', search_operator) @@ -1379,8 +1379,7 @@ class BukuDb: query += ' ORDER BY score DESC)' LOGDBG('query: "%s", args: %s', query, tags_) - self.cur.execute(query, tuple(tags_, )) - return self.cur.fetchall() + return self._fetch(query, *tags_) def search_keywords_and_filter_by_tags( self, @@ -1388,7 +1387,7 @@ class BukuDb: all_keywords: bool, deep: bool, regex: bool, - stag: str) -> Optional[List[BookmarkVar]]: + stag: str) -> List[BookmarkVar]: """Search bookmarks for entries with keywords and specified criteria while filtering out entries with matching tags. @@ -1412,14 +1411,12 @@ class BukuDb: Returns ------- - list or None - List of search results, or None if no matches. + list + List of search results. """ keyword_results = self.searchdb(keywords, all_keywords, deep, regex) - keyword_results = keyword_results if keyword_results is not None else [] stag_results = self.search_by_tag(''.join(stag)) - stag_results = stag_results if stag_results is not None else [] return list(set(keyword_results) & set(stag_results)) def exclude_results_from_search(self, search_results, without, deep): @@ -1436,8 +1433,8 @@ class BukuDb: Returns ------- - list or None - List of search results, or None if no matches. + list + List of search results. """ return list(set(search_results) - set(self.searchdb(without, False, deep))) @@ -1457,7 +1454,7 @@ class BukuDb: # Return if the last index left in DB was just deleted max_id = self.get_max_id() - if max_id == -1: + if not max_id: return query1 = 'SELECT id, URL, metadata, tags, desc, flags FROM bookmarks WHERE id = ? LIMIT 1' @@ -1466,8 +1463,7 @@ class BukuDb: # NOOP if the just deleted index was the last one if max_id > index: - self.cur.execute(query1, (max_id,)) - results = self.cur.fetchall() + results = self._fetch(query1, max_id) for row in results: self.cur.execute(query2, (row[0],)) self.cur.execute(query3, (index, row[1], row[2], row[3], row[4], row[5])) @@ -1784,7 +1780,7 @@ class BukuDb: if not is_range and index < 0: # Show the last n records _id = self.get_max_id() - if _id == -1: + if not _id: LOGERR('Empty database') return False @@ -1813,9 +1809,7 @@ class BukuDb: return False elif index != 0: # Show record at index try: - query = 'SELECT * FROM bookmarks WHERE id = ? LIMIT 1' - self.cur.execute(query, (index,)) - results = self.cur.fetchall() + results = self._fetch('SELECT * FROM bookmarks WHERE id = ? LIMIT 1', index) if not results: LOGERR('No matching index %d', index) return False @@ -2678,7 +2672,7 @@ n: don't add parent folder as tag for item in items: add_rec_res = self.add_rec(*item) - if add_rec_res == -1 and append_tags_resp == 'y': + if not add_rec_res and append_tags_resp == 'y': rec_id = self.get_rec_id(item[0]) self.append_tag_at_index(rec_id, item[2]) @@ -2730,7 +2724,7 @@ n: don't add parent folder as tag def tnyfy_url( self, - index: Optional[int] = 0, + index: Optional[int] = None, url: Optional[str] = None, shorten: Optional[bool] = True) -> Optional[str]: """Shorten a URL using Google URL shortener. @@ -2738,7 +2732,7 @@ n: don't add parent folder as tag Parameters ---------- index : int, optional (if URL is provided) - DB index of the bookmark with the URL to shorten. Default is 0. + DB index of the bookmark with the URL to shorten. Default is None. url : str, optional (if index is provided) URL to shorten. shorten : bool, optional diff --git a/bukuserver/api.py b/bukuserver/api.py index a93b0967..5ab984ab 100644 --- a/bukuserver/api.py +++ b/bukuserver/api.py @@ -144,7 +144,7 @@ def post(self, rec_id: None = None): create_bookmarks_form.tags.data, create_bookmarks_form.description.data ) - return to_response(result_flag != -1) + return to_response(result_flag) def put(self, rec_id: int): bukudb = getattr(flask.g, 'bukudb', get_bukudb()) @@ -171,7 +171,7 @@ class ApiBookmarkRangeView(MethodView): def get(self, starting_id: int, ending_id: int): bukudb = getattr(flask.g, 'bukudb', get_bukudb()) - max_id = bukudb.get_max_id() + max_id = bukudb.get_max_id() or 0 if starting_id > max_id or ending_id > max_id: return response_bad() result = {'bookmarks': {}} # type: ignore @@ -187,7 +187,7 @@ def get(self, starting_id: int, ending_id: int): def put(self, starting_id: int, ending_id: int): bukudb = getattr(flask.g, 'bukudb', get_bukudb()) - max_id = bukudb.get_max_id() + max_id = bukudb.get_max_id() or 0 if starting_id > max_id or ending_id > max_id: return response_bad() for i in range(starting_id, ending_id + 1, 1): @@ -204,7 +204,7 @@ def put(self, starting_id: int, ending_id: int): def delete(self, starting_id: int, ending_id: int): bukudb = getattr(flask.g, 'bukudb', get_bukudb()) - max_id = bukudb.get_max_id() + max_id = bukudb.get_max_id() or 0 if starting_id > max_id or ending_id > max_id: return response_bad() idx = min([starting_id, ending_id]) @@ -233,19 +233,16 @@ def get(self): result = {'bookmarks': []} bukudb = getattr(flask.g, 'bukudb', get_bukudb()) - found_bookmarks = bukudb.searchdb(keywords, all_keywords, deep, regex) - found_bookmarks = [] if found_bookmarks is None else found_bookmarks res = None - if found_bookmarks is not None: - for bookmark in found_bookmarks: - result_bookmark = { - 'id': bookmark[0], - 'url': bookmark[1], - 'title': bookmark[2], - 'tags': list(filter(lambda x: x, bookmark[3].split(','))), - 'description': bookmark[4] - } - result['bookmarks'].append(result_bookmark) + for bookmark in bukudb.searchdb(keywords, all_keywords, deep, regex): + result_bookmark = { + 'id': bookmark[0], + 'url': bookmark[1], + 'title': bookmark[2], + 'tags': list(filter(lambda x: x, bookmark[3].split(','))), + 'description': bookmark[4] + } + result['bookmarks'].append(result_bookmark) current_app.logger.debug('total bookmarks:{}'.format(len(result['bookmarks']))) res = jsonify(result) return res @@ -267,13 +264,10 @@ def delete(self): deep = deep if isinstance(deep, bool) else deep.lower() == 'true' regex = regex if isinstance(regex, bool) else regex.lower() == 'true' bukudb = getattr(flask.g, 'bukudb', get_bukudb()) - found_bookmarks = bukudb.searchdb(keywords, all_keywords, deep, regex) - found_bookmarks = [] if found_bookmarks is None else found_bookmarks res = None - if found_bookmarks is not None: - for bookmark in found_bookmarks: - if not bukudb.delete_rec(bookmark[0]): - res = response_bad() + for bookmark in bukudb.searchdb(keywords, all_keywords, deep, regex): + if not bukudb.delete_rec(bookmark[0]): + res = response_bad() return res or response_ok() @@ -285,6 +279,6 @@ def get(self): bukudb = getattr(flask.g, 'bukudb', get_bukudb()) rec_id = bukudb.get_rec_id(url) - if rec_id >= 0: + if rec_id: return redirect(url_for('bookmark.edit_view', id=rec_id)) return redirect(url_for('bookmark.create_view', link=url, title=title, description=description)) diff --git a/bukuserver/views.py b/bukuserver/views.py index f742bf5f..8ae288f8 100644 --- a/bukuserver/views.py +++ b/bukuserver/views.py @@ -85,11 +85,11 @@ def _filter_arg(flt): """Exposes filter slugify logic; works because BookmarkModelView.named_filter_urls = True""" return BaseModelView.get_filter_arg(BookmarkModelView, None, flt) - def _saved(self, id, url, ok): - if ok: + def _saved(self, id, url, ok=True): + if id and ok: session['saved'] = id else: - raise Exception('Duplicate URL' if self.model.bukudb.get_rec_id(url) not in [-1, id] else + raise Exception('Duplicate URL' if self.model.bukudb.get_rec_id(url) not in [id, None] else 'Rejected by the database') def _apply_filters(self, models, filters): @@ -205,7 +205,7 @@ def create_model(self, form): if item.strip(): kwargs[key] = item vars(model)['id'] = self.model.bukudb.add_rec(**kwargs) - self._saved(model.id, model.url, model.id != -1) + self._saved(model.id, model.url) except Exception as ex: if not self.handle_view_exception(ex): msg = "Failed to create record." diff --git a/tests/test_bukuDb.py b/tests/test_bukuDb.py index bfe93c10..c79f125d 100644 --- a/tests/test_bukuDb.py +++ b/tests/test_bukuDb.py @@ -172,9 +172,9 @@ def test_get_rec_id(self): idx_from_db = self.bdb.get_rec_id(bookmark[0]) self.assertEqual(idx + 1, idx_from_db) - # asserting -1 is returned for nonexistent url + # asserting None is returned for nonexistent url idx_from_db = self.bdb.get_rec_id("http://nonexistent.url") - self.assertEqual(-1, idx_from_db) + self.assertIsNone(idx_from_db) def test_add_rec(self): for bookmark in self.bookmarks: @@ -1076,7 +1076,7 @@ def test_add_rec_add_invalid_url(caplog, url): """test method.""" bdb = BukuDb() res = bdb.add_rec(url=url) - assert res == -1 + assert res is None caplog.records[0].levelname == "ERROR" caplog.records[0].getMessage() == "Invalid URL" @@ -1119,7 +1119,7 @@ def test_add_rec_exec_arg(kwargs, exp_arg): """test func.""" bdb = BukuDb() bdb.cur = mock.Mock() - bdb.get_rec_id = mock.Mock(return_value=-1) + bdb.get_rec_id = mock.Mock(return_value=None) bdb.add_rec(**kwargs) assert bdb.cur.execute.call_args[0][1] == exp_arg @@ -1408,7 +1408,7 @@ def test_exportdb_to_db(): @pytest.mark.parametrize( "urls, exp_res", [ - [[], -1], + [[], None], [["http://example.com"], 1], [["htttp://example.com", "http://google.com"], 2], ],