database.py (5175B)
1 #!/usr/bin/env python3 2 # -*- coding: utf-8 -*- 3 # vim:fenc=utf-8 4 5 import sqlite3 6 from sqlite3 import Error 7 8 9 GET_CATEGORIES_QUERY = "SELECT id, name FROM categories" 10 GET_STORAGES_QUERY = "SELECT id, name FROM storages" 11 LIST_PARTS_QUERY = """ 12 SELECT p.id, p.pn, p.manufacturer, p.description, 13 p.footprint, p.quantity, p.part_type, c.name as cname 14 FROM parts as p, categories as c 15 INNER JOIN categories 16 ON p.category_id = c.id 17 GROUP BY p.id 18 """ 19 LIST_PARTS_CATEGORY_QUERY = """ 20 SELECT p.id, p.pn, p.manufacturer, p.description, 21 p.footprint, p.quantity, p.part_type, c.name as cname 22 FROM parts as p, categories as c 23 INNER JOIN categories 24 ON p.category_id = c.id 25 WHERE c.name = ? 26 GROUP BY p.id 27 """ 28 INSERT_PART_QUERY = """ 29 INSERT INTO parts 30 ( 31 pn, 32 mpn, 33 manufacturer, 34 description, 35 specs, 36 footprint, 37 category_id, 38 storage_id, 39 quantity, 40 datasheet, 41 image, 42 part_type 43 ) 44 VALUES (?,?,?,?,?,?,?,?,?,?,?,?) 45 """ 46 GET_PART_QUERY = """ 47 SELECT p.*, c.name as cat, s.name as storage 48 FROM parts as p, categories as c, storages as s 49 INNER JOIN categories 50 ON p.category_id = c.id 51 INNER JOIN storages 52 ON p.storage_id = s.id 53 WHERE p.id = ? 54 GROUP BY p.id 55 """ 56 SEARCH_PARTS_QUERY = """ 57 SELECT p.id, p.pn, p.manufacturer, p.description, 58 p.footprint, p.quantity, p.part_type, c.name as cname 59 FROM parts as p, categories as c 60 INNER JOIN categories 61 ON p.category_id = c.id 62 WHERE 63 pn LIKE '%'||?||'%' OR 64 mpn LIKE '%'||?||'%' OR 65 manufacturer LIKE '%'||?||'%' OR 66 footprint LIKE '%'||?||'%' OR 67 part_type LIKE '%'||?||'%' 68 GROUP BY p.id 69 """ 70 GET_PART_BY_MPN_QUERY = "SELECT id FROM parts WHERE mpn = ?" 71 GET_HISTORY_QUERY = "SELECT * FROM parts_history WHERE part_id = ?" 72 INSERT_HISTORY_EVENT = """ 73 INSERT INTO parts_history 74 (part_id, movement, mcomment) 75 VALUES (?,?,?) 76 """ 77 UPDATE_PART_QTY_QUERY = "UPDATE parts SET quantity = ? WHERE id = ?" 78 GET_PART_QTY = "SELECT quantity FROM parts where id = ?" 79 DELETE_PART_QUERY = "DELETE FROM parts WHERE id = ?" 80 DELETE_HISTORY_QUERY = "DELETE FROM parts_history WHERE part_id = ?" 81 GET_IMAGE_QUERY = "SELECT image FROM parts WHERE id = ?" 82 GET_DATASHEET_QUERY = "SELECT datasheet FROM parts WHERE id = ?" 83 84 85 class PartsDB(): 86 87 def __init__(self, database): 88 try: 89 self.conn = sqlite3.connect(database) 90 self.conn.row_factory = sqlite3.Row 91 except Error as e: 92 print(e) 93 94 def get_categories(self): 95 c = self.conn.cursor() 96 c.execute(GET_CATEGORIES_QUERY) 97 return c.fetchall() 98 99 def get_storages(self): 100 c = self.conn.cursor() 101 c.execute(GET_STORAGES_QUERY) 102 return c.fetchall() 103 104 def list_parts(self): 105 c = self.conn.cursor() 106 c.execute(LIST_PARTS_QUERY) 107 return c.fetchall() 108 109 def list_parts_by_category(self, category): 110 c = self.conn.cursor() 111 c.execute(LIST_PARTS_CATEGORY_QUERY, (category,)) 112 return c.fetchall() 113 114 def new_part(self, part): 115 if part[9] is not None: 116 part[9] = sqlite3.Binary(part[9]) 117 p = tuple(part) 118 with self.conn: 119 c = self.conn.cursor() 120 c.execute(INSERT_PART_QUERY, p) 121 return c.lastrowid 122 123 def get_part(self, part_id): 124 c = self.conn.cursor() 125 c.execute(GET_PART_QUERY, (part_id,)) 126 return c.fetchone() 127 128 def get_part_by_mpn(self, mpn): 129 c = self.conn.cursor() 130 c.execute(GET_PART_BY_MPN_QUERY, (mpn,)) 131 return c.fetchone() 132 133 def search_parts(self, term): 134 c = self.conn.cursor() 135 c.execute(SEARCH_PARTS_QUERY, (term, term, term, term, term)) 136 return c.fetchall() 137 138 def get_part_history(self, part_id): 139 c = self.conn.cursor() 140 c.execute(GET_HISTORY_QUERY, (part_id,)) 141 return c.fetchall() 142 143 def new_part_history_event(self, part_id, movement, comment): 144 with self.conn: 145 c = self.conn.cursor() 146 c.execute(INSERT_HISTORY_EVENT, (part_id, movement, comment)) 147 148 def update_part_qty(self, part_id, movement): 149 with self.conn: 150 c = self.conn.cursor() 151 c.execute(GET_PART_QTY, (part_id,)) 152 qty = c.fetchone()['quantity'] 153 qty = qty + movement 154 c.execute(UPDATE_PART_QTY_QUERY, (qty, part_id)) 155 156 def delete_part(self, part_id): 157 with self.conn: 158 c = self.conn.cursor() 159 c.execute(DELETE_HISTORY_QUERY, (part_id,)) 160 c.execute(DELETE_PART_QUERY, (part_id,)) 161 162 def get_image(self, part_id): 163 c = self.conn.cursor() 164 c.execute(GET_IMAGE_QUERY, (part_id,)) 165 return c.fetchone() 166 167 def get_datasheet(self, part_id): 168 c = self.conn.cursor() 169 c.execute(GET_DATASHEET_QUERY, (part_id,)) 170 return c.fetchone() 171 172 def close(self): 173 self.conn.close()