py.lib
py.lib/db/sqlite.py
* С разбегу влетел в type hinting, пришлось обкладывать костылями. И этот процесс судя по всему длительный, и место больное будет долго...
| awgur@13 | 1 # coding: utf-8 |
| awgur@13 | 2 import sqlite3 |
| awgur@14 | 3 from sqlite3 import Error, IntegrityError |
| awgur@13 | 4 |
| awgur@13 | 5 class DB: |
| awgur@13 | 6 def __init__(self, db_file): |
| awgur@19 | 7 self._conn = sqlite3.connect(db_file) |
| awgur@19 | 8 self._ex = self._conn.execute |
| awgur@19 | 9 self.commit = self._conn.commit |
| awgur@19 | 10 self.rollback = self._conn.rollback |
| awgur@13 | 11 |
| awgur@13 | 12 # DB PREP |
| awgur@13 | 13 self._ex("PRAGMA journal=WAL") |
| awgur@15 | 14 self._ex("PRAGMA foreign_keys=ON") |
| awgur@13 | 15 self.commit() |
| awgur@13 | 16 |
| awgur@13 | 17 def __del__(self): |
| awgur@13 | 18 try: |
| awgur@13 | 19 self.rollback() |
| awgur@19 | 20 self._conn.close() |
| awgur@13 | 21 except: |
| awgur@13 | 22 pass |
| awgur@13 | 23 |
| awgur@13 | 24 def __call__(self, *a, **kwa): |
| awgur@19 | 25 cur = self._conn.cursor() |
| awgur@13 | 26 cur.execute(*a, **kwa) |
| awgur@13 | 27 return cur |
| awgur@13 | 28 |
| awgur@13 | 29 def cq(self, *a, **wa): |
| awgur@13 | 30 try: |
| awgur@13 | 31 res = self(*a, **wa) |
| awgur@13 | 32 self.commit() |
| awgur@13 | 33 return res |
| awgur@14 | 34 except Error as e: |
| awgur@13 | 35 self.rollback() |
| awgur@13 | 36 raise e |