database.py (8947B)
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 GET_CATEGORIES_QUERY = "SELECT id, name FROM categories" 9 GET_STORAGES_QUERY = "SELECT id, name FROM storages" 10 LIST_PARTS_QUERY = """ 11 SELECT p.id, p.mpn, p.manufacturer, p.description, 12 p.footprint, p.quantity, p.part_type, c.name as cname 13 FROM parts as p, categories as c 14 INNER JOIN categories 15 ON p.category_id = c.id 16 GROUP BY p.id 17 """ 18 LIST_PARTS_CATEGORY_QUERY = """ 19 SELECT p.id, p.mpn, p.manufacturer, p.description, 20 p.footprint, p.quantity, p.part_type, c.name as cname 21 FROM parts as p, categories as c 22 INNER JOIN categories 23 ON p.category_id = c.id 24 WHERE c.name = ? OR c.id = ? 25 GROUP BY p.id 26 """ 27 LIST_PARTS_STORAGE_QUERY = """ 28 SELECT p.id, p.mpn, p.manufacturer, p.description, 29 p.footprint, p.quantity, p.part_type, s.name as sname 30 FROM parts as p, storages as s 31 INNER JOIN categories 32 ON p.storage_id = s.id 33 WHERE s.name = ? OR s.id = ? 34 GROUP BY p.id 35 """ 36 INSERT_PART_QUERY = """ 37 INSERT INTO parts 38 ( 39 mpn, 40 manufacturer, 41 description, 42 specs, 43 footprint, 44 category_id, 45 storage_id, 46 quantity, 47 datasheet, 48 image, 49 part_type 50 ) 51 VALUES (?,?,?,?,?,?,?,?,?,?,?) 52 """ 53 INSERT_PART_QUERY_FROM_TMPL = """ 54 INSERT INTO parts 55 ( 56 mpn, 57 manufacturer, 58 description, 59 specs, 60 footprint, 61 category_id, 62 storage_id, 63 quantity, 64 part_type 65 ) 66 VALUES (?,?,?,?,?,?,?,?,?) 67 """ 68 UPDATE_PART_QUERY = """ 69 UPDATE parts 70 SET 71 mpn = ?, 72 manufacturer = ?, 73 description = ?, 74 specs = ?, 75 footprint = ?, 76 category_id = ?, 77 storage_id = ?, 78 quantity = ?, 79 part_type = ?, 80 update_date = datetime('now','localtime') 81 WHERE 82 id = ? 83 """ 84 UPDATE_DATASHEET_QUERY = """ 85 UPDATE parts 86 SET 87 datasheet = ?, 88 update_date = datetime('now','localtime') 89 WHERE 90 id = ? 91 """ 92 UPDATE_IMAGE_QUERY = """ 93 UPDATE parts 94 SET 95 image = ?, 96 update_date = datetime('now','localtime') 97 WHERE 98 id = ? 99 """ 100 GET_PART_QUERY = """ 101 SELECT p.*, c.name as cat, s.name as storage 102 FROM parts as p, categories as c, storages as s 103 INNER JOIN categories 104 ON p.category_id = c.id 105 INNER JOIN storages 106 ON p.storage_id = s.id 107 WHERE p.id = ? 108 GROUP BY p.id 109 """ 110 SEARCH_PARTS_QUERY = """ 111 SELECT p.id, p.mpn, p.manufacturer, p.description, 112 p.footprint, p.quantity, p.part_type, c.name as cname 113 FROM parts as p, categories as c 114 INNER JOIN categories 115 ON p.category_id = c.id 116 WHERE 117 mpn LIKE '%'||?||'%' OR 118 manufacturer LIKE '%'||?||'%' OR 119 footprint LIKE '%'||?||'%' OR 120 part_type LIKE '%'||?||'%' 121 GROUP BY p.id 122 """ 123 GET_PART_BY_MPN_QUERY = "SELECT id FROM parts WHERE mpn = ?" 124 GET_HISTORY_QUERY = "SELECT * FROM parts_history WHERE part_id = ?" 125 INSERT_HISTORY_EVENT = """ 126 INSERT INTO parts_history 127 (part_id, movement, mcomment) 128 VALUES (?,?,?) 129 """ 130 UPDATE_PART_QTY_QUERY = "UPDATE parts SET quantity = ? WHERE id = ?" 131 GET_PART_QTY = "SELECT quantity FROM parts where id = ?" 132 DELETE_PART_QUERY = "DELETE FROM parts WHERE id = ?" 133 DELETE_HISTORY_QUERY = "DELETE FROM parts_history WHERE part_id = ?" 134 GET_IMAGE_QUERY = "SELECT image FROM parts WHERE id = ?" 135 GET_DATASHEET_QUERY = "SELECT datasheet FROM parts WHERE id = ?" 136 137 GET_DB_MAJOR_VERSION_METADATA = "SELECT value FROM metadata WHERE name = 'major_version'" 138 139 140 class PartsDB: 141 def __init__(self, database): 142 try: 143 self.conn = sqlite3.connect(database) 144 self.conn.row_factory = sqlite3.Row 145 except Error as e: 146 print(e) 147 148 def get_categories(self): 149 c = self.conn.cursor() 150 c.execute(GET_CATEGORIES_QUERY) 151 return c.fetchall() 152 153 def get_category_by_name(self, cat_name): 154 c = self.conn.cursor() 155 c.execute(f"{GET_CATEGORIES_QUERY} WHERE name = ?", (cat_name,)) 156 return c.fetchone() 157 158 def get_storages(self): 159 c = self.conn.cursor() 160 c.execute(GET_STORAGES_QUERY) 161 return c.fetchall() 162 163 def get_storage_by_name(self, sto_name): 164 c = self.conn.cursor() 165 c.execute(f"{GET_STORAGES_QUERY} WHERE name = ?", (sto_name,)) 166 return c.fetchone() 167 168 def list_parts(self): 169 c = self.conn.cursor() 170 c.execute(LIST_PARTS_QUERY) 171 return c.fetchall() 172 173 def list_parts_by_category(self, category): 174 c = self.conn.cursor() 175 c.execute(LIST_PARTS_CATEGORY_QUERY, (category, category)) 176 return c.fetchall() 177 178 def list_parts_by_storage(self, storage): 179 c = self.conn.cursor() 180 c.execute(LIST_PARTS_STORAGE_QUERY, (storage, storage)) 181 return c.fetchall() 182 183 def new_part(self, part): 184 if part[9] is not None: 185 part[9] = sqlite3.Binary(part[9]) 186 p = tuple(part) 187 with self.conn: 188 c = self.conn.cursor() 189 c.execute(INSERT_PART_QUERY, p) 190 return c.lastrowid 191 192 def new_part_from_template(self, part): 193 with self.conn: 194 cat_id = self.get_category_by_name(part["cat"])["id"] 195 sto_id = self.get_storage_by_name(part["storage"])["id"] 196 c = self.conn.cursor() 197 update_tuple = ( 198 part["mpn"], 199 part["manufacturer"], 200 part["description"], 201 part["specs"], 202 part["footprint"], 203 cat_id, 204 sto_id, 205 part["quantity"], 206 part["part_type"], 207 ) 208 c.execute(INSERT_PART_QUERY_FROM_TMPL, update_tuple) 209 return c.lastrowid 210 211 def get_part(self, part_id): 212 c = self.conn.cursor() 213 c.execute(GET_PART_QUERY, (part_id,)) 214 return c.fetchone() 215 216 def get_part_by_mpn(self, mpn): 217 c = self.conn.cursor() 218 c.execute(GET_PART_BY_MPN_QUERY, (mpn,)) 219 return c.fetchone() 220 221 def search_parts(self, term): 222 c = self.conn.cursor() 223 c.execute(SEARCH_PARTS_QUERY, (term, term, term, term)) 224 return c.fetchall() 225 226 def get_part_history(self, part_id): 227 c = self.conn.cursor() 228 c.execute(GET_HISTORY_QUERY, (part_id,)) 229 return c.fetchall() 230 231 def new_part_history_event(self, part_id, movement, comment): 232 with self.conn: 233 c = self.conn.cursor() 234 c.execute(INSERT_HISTORY_EVENT, (part_id, movement, comment)) 235 236 def update_part_qty(self, part_id, movement): 237 with self.conn: 238 c = self.conn.cursor() 239 c.execute(GET_PART_QTY, (part_id,)) 240 qty = c.fetchone()["quantity"] 241 qty = qty + movement 242 c.execute(UPDATE_PART_QTY_QUERY, (qty, part_id)) 243 244 def delete_part(self, part_id): 245 with self.conn: 246 c = self.conn.cursor() 247 c.execute(DELETE_HISTORY_QUERY, (part_id,)) 248 c.execute(DELETE_PART_QUERY, (part_id,)) 249 250 def update_part(self, part): 251 with self.conn: 252 cat_id = self.get_category_by_name(part["cat"])["id"] 253 sto_id = self.get_storage_by_name(part["storage"])["id"] 254 c = self.conn.cursor() 255 update_tuple = ( 256 part["mpn"], 257 part["manufacturer"], 258 part["description"], 259 part["specs"], 260 part["footprint"], 261 cat_id, 262 sto_id, 263 part["quantity"], 264 part["part_type"], 265 part["id"], 266 ) 267 c.execute(UPDATE_PART_QUERY, update_tuple) 268 269 def update_datasheet(self, part_id, datasheet): 270 with self.conn: 271 c = self.conn.cursor() 272 c.execute(UPDATE_DATASHEET_QUERY, (sqlite3.Binary(datasheet), part_id)) 273 274 def update_image(self, part_id, image): 275 with self.conn: 276 c = self.conn.cursor() 277 c.execute(UPDATE_IMAGE_QUERY, (sqlite3.Binary(image), part_id)) 278 279 def get_image(self, part_id): 280 c = self.conn.cursor() 281 c.execute(GET_IMAGE_QUERY, (part_id,)) 282 return c.fetchone() 283 284 def get_datasheet(self, part_id): 285 c = self.conn.cursor() 286 c.execute(GET_DATASHEET_QUERY, (part_id,)) 287 return c.fetchone() 288 289 def validate_major_version(self, version: int) -> bool: 290 c = self.conn.cursor() 291 c.execute(GET_DB_MAJOR_VERSION_METADATA) 292 db_major_version_record = c.fetchone() 293 294 return int(db_major_version_record["value"]) == version 295 296 def close(self): 297 self.conn.close()