API for querying last read positions from db

src/calibre/db/cache.py

@@ -12,6 +12,7 @@ from io import BytesIO
 from collections import defaultdict, Set, MutableSet
 from functools import wraps, partial
 from future_builtins import zip
+from time import time
 
 from calibre import isbytestring, as_unicode
 from calibre.constants import iswindows, preferred_encoding
@@ -2131,6 +2132,30 @@ class Cache(object):
             if report_progress is not None:
                 report_progress(i+1, len(book_ids), mi)
 
+    @read_api
+    def get_last_read_positions(self, book_id, fmt, user):
+        fmt = fmt.upper()
+        ans = []
+        for device, cfi, epoch in self.backend.execute(
+                'SELECT device,cfi,epoch FROM last_read_positions WHERE book=? AND format=? AND user=?',
+                (book_id, fmt, user)):
+            ans.append({'device':device, 'cfi': cfi, 'epoch':epoch})
+        return ans
+
+    @write_api
+    def set_last_read_position(self, book_id, fmt, user='_', device='_', cfi=None, epoch=None):
+        fmt = fmt.upper()
+        device = device or '_'
+        user = user or '_'
+        if not cfi:
+            self.backend.execute(
+                'DELETE FROM last_read_positions WHERE book=? AND format=? AND user=? AND device=?',
+                (book_id, fmt, user, device))
+        else:
+            self.backend.execute(
+                'INSERT OR REPLACE INTO last_read_positions(book,format,user,device,cfi,epoch) VALUES (?,?,?,?,?,?)',
+                (book_id, fmt, user, device, cfi, epoch or time()))
+
     @read_api
     def export_library(self, library_key, exporter, progress=None, abort=None):
         from binascii import hexlify

src/calibre/db/tests/reading.py

@@ -9,6 +9,7 @@ __docformat__ = 'restructuredtext en'
 
 import datetime
 from io import BytesIO
+from time import time
 
 from calibre.utils.date import utc_tz
 from calibre.db.tests.base import BaseTest
@@ -668,3 +669,15 @@ class ReadingTest(BaseTest):
             self.assertEqual(books, cache.find_identical_books(mi))
             self.assertEqual(books, find_identical_books(mi, data))
     # }}}
+
+    def test_last_read_positions(self):  # {{{
+        cache = self.init_cache(self.library_path)
+        self.assertFalse(cache.get_last_read_positions(1, 'x', 'u'))
+        self.assertRaises(Exception, cache.set_last_read_position, 12, 'x', cfi='c')
+        epoch = time()
+        cache.set_last_read_position(1, 'EPUB', 'user', 'device', 'cFi', epoch)
+        self.assertFalse(cache.get_last_read_positions(1, 'x', 'u'))
+        self.assertEqual(cache.get_last_read_positions(1, 'ePuB', 'user'), [{'epoch':epoch, 'device':'device', 'cfi':'cFi'}])
+        cache.set_last_read_position(1, 'EPUB', 'user', 'device')
+        self.assertFalse(cache.get_last_read_positions(1, 'ePuB', 'user'))
+    # }}}