GitHub/Anki
1
0
Fork 0
mirror of https://github.com/ankitects/anki.git synced 2025-09-19 06:22:22 -04:00
Code Issues Projects Releases Packages 1 Wiki Activity Actions

tweak layout of db methods

Browse source
This commit is contained in:
Damien Elmes 2020-03-03 08:51:29 +10:00
parent 5778459d7c
commit 0b1d96fce0
1 changed files with 43 additions and 31 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

74
pylib/anki/dbproxy.py
View file

@ -4,48 +4,40 @@
# fixme: lossy utf8 handling # fixme: lossy utf8 handling
# fixme: progress # fixme: progress
import time
from sqlite3 import Cursor from sqlite3 import Cursor
from sqlite3 import dbapi2 as sqlite from sqlite3 import dbapi2 as sqlite
from typing import Any, List from typing import Any, Iterable, List
class DBProxy: class DBProxy:
# Lifecycle
###############
def __init__(self, path: str) -> None: def __init__(self, path: str) -> None:
self._db = sqlite.connect(path, timeout=0) self._db = sqlite.connect(path, timeout=0)
self._path = path self._path = path
self.mod = False self.mod = False
def execute(self, sql: str, *args) -> Cursor: def close(self) -> None:
s = sql.strip().lower() self._db.close()
# mark modified?
for stmt in "insert", "update", "delete":
if s.startswith(stmt):
self.mod = True
res = self._db.execute(sql, args)
return res
def executemany(self, sql: str, l: Any) -> None: # Transactions
self.mod = True ###############
t = time.time()
self._db.executemany(sql, l)
def commit(self) -> None: def commit(self) -> None:
t = time.time()
self._db.commit() self._db.commit()
def executescript(self, sql: str) -> None:
self.mod = True
self._db.executescript(sql)
def rollback(self) -> None: def rollback(self) -> None:
self._db.rollback() self._db.rollback()
def scalar(self, sql: str, *args) -> Any: def setAutocommit(self, autocommit: bool) -> None:
res = self.execute(sql, *args).fetchone() if autocommit:
if res: self._db.isolation_level = None
return res[0] else:
return None self._db.isolation_level = ""
# Querying
################
def all(self, sql: str, *args) -> List: def all(self, sql: str, *args) -> List:
return self.execute(sql, *args).fetchall() return self.execute(sql, *args).fetchall()
@ -59,11 +51,31 @@ class DBProxy:
def list(self, sql: str, *args) -> List: def list(self, sql: str, *args) -> List:
return [x[0] for x in self.execute(sql, *args)] return [x[0] for x in self.execute(sql, *args)]
def close(self) -> None: def scalar(self, sql: str, *args) -> Any:
self._db.close() res = self.execute(sql, *args).fetchone()
if res:
return res[0]
return None
def setAutocommit(self, autocommit: bool) -> None: # Updates
if autocommit: ################
self._db.isolation_level = None
else: def executemany(self, sql: str, args: Iterable) -> None:
self._db.isolation_level = "" self.mod = True
self._db.executemany(sql, args)
def executescript(self, sql: str) -> None:
self.mod = True
self._db.executescript(sql)
# Cursor API
###############
def execute(self, sql: str, *args) -> Cursor:
s = sql.strip().lower()
# mark modified?
for stmt in "insert", "update", "delete":
if s.startswith(stmt):
self.mod = True
res = self._db.execute(sql, args)
return res