Cutlery/calibre
1
0
Fork 0
mirror of https://github.com/kovidgoyal/calibre.git synced 2025-07-09 03:04:10 -04:00
Code Issues Packages Projects Releases Wiki Activity

Fix dump and restore on memory limited systems

Browse Source
This commit is contained in:
Kovid Goyal 2013-09-07 12:55:11 +05:30
parent 00bd5a658e
commit 18e3f9464b
2 changed files with 24 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
src/calibre/db/backend.py
View File

@ -982,26 +982,28 @@ class DB(object):
self.conn
def dump_and_restore(self, callback=None, sql=None):
from io import StringIO
import codecs
from calibre.utils.apsw_shell import Shell
from contextlib import closing
if callback is None:
callback = lambda x: x
uv = int(self.user_version)
with TemporaryFile(suffix='.sql') as fname:
if sql is None:
from calibre.utils.apsw_shell import Shell
callback(_('Dumping database to SQL') + '...')
buf = StringIO()
with codecs.open(fname, 'wb', encoding='utf-8') as buf:
shell = Shell(db=self.conn, stdout=buf)
shell.process_command('.dump')
sql = buf.getvalue()
del shell
del buf
else:
with open(fname, 'wb') as buf:
buf.write(sql if isinstance(sql, bytes) else sql.encode('utf-8'))
with TemporaryFile(suffix='_tmpdb.db', dir=os.path.dirname(self.dbpath)) as tmpdb:
callback(_('Restoring database from SQL') + '...')
with closing(Connection(tmpdb)) as conn:
conn.execute(sql)
shell = Shell(db=conn, encoding='utf-8')
shell.process_command('.read ' + fname)
conn.execute('PRAGMA user_version=%d;'%uv)
self.close()

2
src/calibre/db/tests/writing.py
View File

@ -606,11 +606,13 @@ class WritingTest(BaseTest):
def test_dump_and_restore(self): # {{{
' Test roundtripping the db through SQL '
cache = self.init_cache()
uv = int(cache.backend.user_version)
all_ids = cache.all_book_ids()
cache.dump_and_restore()
self.assertEqual(cache.set_field('title', {1:'nt'}), set([1]), 'database connection broken')
cache = self.init_cache()
self.assertEqual(cache.all_book_ids(), all_ids, 'dump and restore broke database')
self.assertEqual(int(cache.backend.user_version), uv)
# }}}
def test_set_author_data(self): # {{{