Source code for sofos.dbf

import sqlite3
import os.path
from . import gr


[docs]def save(dbf, sql): """Safely save (create or update) data to database :param dbf: Database file name (full path) :param sql: select sql :return: True is operation successful, else False """ if not os.path.isfile(dbf): return False, 'File %s does not exist' % dbf if sql.lower()[:6] not in ('insert', 'update'): return False, "sql does not start with insert or update" try: con = sqlite3.connect(dbf) cur = con.cursor() cur.execute(sql) con.commit() except sqlite3.Error as err: con.rollback() cur.close() con.close() return False, str(err) last_inserted_id = cur.lastrowid cur.close() con.close() return True, last_inserted_id
[docs]def read(dbf, sql, returns): """SELECT :param dbf: Database file :param sql: sql :param returns: Return Type :return: one, list of dicts, cols/rows, rows """ if not os.path.isfile(dbf): return False, 'File %s does not exist' % dbf if not sql.lower().startswith('select'): return False, 'sql (%s) does not start with select' % sql if returns not in ('one', 'list_of_dicts', 'cols_rows', 'rows'): msg = "returns must be one of (one, list_of_dicts, cols_rows, rows)" return False, msg with sqlite3.connect(dbf) as con: cur = con.cursor() con.create_function("grup", 1, gr.grup) try: cur.execute(sql) col = tuple([t[0] for t in cur.description]) except sqlite3.OperationalError as err: return False, str(err) if returns == 'one': # {'id': 1, 'key1': val1, 'key2': val2, ...} row = cur.fetchone() if row: return True, dict(zip([c[0] for c in cur.description], row)) else: return False, {} elif returns == 'list_of_dicts': # [{'id': 1, 'k1': v1, ...}, {'id': 2, 'k1': v1, ...}] rows = cur.fetchall() ldicts = [] for row in rows: ldicts.append(dict(zip([c[0] for c in cur.description], row))) return True, ldicts elif returns == 'cols_rows': # {'cols': (c1, c2,..), # 'rows': [(1, v1, ..), (2, v1, ..), ...], # 'rownum': 34, # 'colnum': 4} row = cur.fetchall() return True, {'cols': col, 'rows': row, 'rownum': len(row), 'colnum': len(col)} elif returns == 'rows': # [(a1, a2, ...), (b1, b2, ...), ...] rows = cur.fetchall() return True, rows
[docs]def delete(dbf, sql): """Delete from db :param dbf: Database file :param sql: sql :return: True if success, False else """ if not os.path.isfile(dbf): return False, 'File %s does not exist' % dbf if not sql.lower().startswith('delete'): return False, 'sql (%s) does not start with delete' % sql try: con = sqlite3.connect(dbf) cur = con.cursor() cur.execute(sql) con.commit() except sqlite3.Error as err: con.rollback() cur.close() con.close() return False, str(err) cur.close() con.close() return True, 'Record deleted'
[docs]def script(dbf, sql, create=False): """Run sql script against database dbf :param dbf: Database file name(full path) :param sql: Sql to execute (Normally inside Transaction) :param create: If True creates new database file """ if create: if os.path.isfile(dbf): return False, 'Database file %s already exists' % dbf else: if not os.path.isfile(dbf): return False, 'Database file %s does not exist' % dbf try: with sqlite3.connect(dbf) as con: con.executescript(sql) except sqlite3.Error as err: return False, str(err) return True, 'Script executed succesfuly'
[docs]def backup(dbf, backupfile, overwrite=False, inserts_only=True): """Backup database :param dbf: Database file to backup :param backupfile: backup destination filename :param overwite: If is allowed to overwrite :param inserts_only: If True backup data only, if False backup everything :return: True if backup was successful """ SEL = ('INSERT INTO', 'BEGIN', 'COMMIT') def _insel(lin): """For use in backup_database""" for elm in SEL: if lin.startswith(elm): return True return False if not os.path.isfile(dbf): return False, 'File %s does not exist' % dbf if not overwrite and os.path.isfile(backupfile): return False, 'File %s already exists' % backupfile try: with sqlite3.connect(dbf) as con: if inserts_only: data = '\n'.join([i for i in con.iterdump() if _insel(i)]) else: data = '\n'.join(con.iterdump()) except sqlite3.Error as err: return False, '%s\n' % err except Exception as err: return False, err with open(backupfile, 'w') as fil: fil.write(data) return True, 'Database %s backup saved to %s' % (dbf, backupfile)
[docs]def ref_exists(dbf, table, field, idv): sql = "SELECT COUNT(*) FROM %s WHERE %s='%s'" % (table, field, idv) if not os.path.isfile(dbf): return False, 'File %s does not exist' % dbf try: with sqlite3.connect(dbf) as con: cur = con.cursor() cur.execute(sql) row = cur.fetchone() except sqlite3.Error as err: return False, '%s\n' % err except Exception as err: return False, err if row[0] > 0: return True return False