partsdb

electronic parts inventory
git clone https://git.e1e0.net/partsdb.git
Log | Files | Refs | README | LICENSE

database.py (5681B)


      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.pn, 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.pn, 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.pn, 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             pn,
     40             mpn,
     41             manufacturer,
     42             description,
     43             specs,
     44             footprint,
     45             category_id,
     46             storage_id,
     47             quantity,
     48             datasheet,
     49             image,
     50             part_type
     51         )
     52     VALUES (?,?,?,?,?,?,?,?,?,?,?,?)
     53 """
     54 GET_PART_QUERY = """
     55     SELECT p.*, c.name as cat, s.name as storage
     56     FROM parts as p, categories as c, storages as s
     57         INNER JOIN categories
     58             ON p.category_id = c.id
     59         INNER JOIN storages
     60             ON p.storage_id = s.id
     61     WHERE p.id = ?
     62     GROUP BY p.id
     63 """
     64 SEARCH_PARTS_QUERY = """
     65     SELECT p.id, p.pn, p.manufacturer, p.description,
     66            p.footprint, p.quantity, p.part_type, c.name as cname
     67     FROM parts as p, categories as c
     68         INNER JOIN categories
     69             ON p.category_id = c.id
     70     WHERE
     71         pn LIKE '%'||?||'%' OR
     72         mpn LIKE '%'||?||'%' OR
     73         manufacturer LIKE '%'||?||'%' OR
     74         footprint LIKE '%'||?||'%' OR
     75         part_type LIKE '%'||?||'%'
     76     GROUP BY p.id
     77 """
     78 GET_PART_BY_MPN_QUERY = "SELECT id FROM parts WHERE mpn = ?"
     79 GET_HISTORY_QUERY = "SELECT * FROM parts_history WHERE part_id = ?"
     80 INSERT_HISTORY_EVENT = """
     81     INSERT INTO parts_history
     82         (part_id, movement, mcomment)
     83     VALUES (?,?,?)
     84 """
     85 UPDATE_PART_QTY_QUERY = "UPDATE parts SET quantity = ? WHERE id = ?"
     86 GET_PART_QTY = "SELECT quantity FROM parts where id = ?"
     87 DELETE_PART_QUERY = "DELETE FROM parts WHERE id = ?"
     88 DELETE_HISTORY_QUERY = "DELETE FROM parts_history WHERE part_id = ?"
     89 GET_IMAGE_QUERY = "SELECT image FROM parts WHERE id = ?"
     90 GET_DATASHEET_QUERY = "SELECT datasheet FROM parts WHERE id = ?"
     91 
     92 
     93 class PartsDB:
     94     def __init__(self, database):
     95         try:
     96             self.conn = sqlite3.connect(database)
     97             self.conn.row_factory = sqlite3.Row
     98         except Error as e:
     99             print(e)
    100 
    101     def get_categories(self):
    102         c = self.conn.cursor()
    103         c.execute(GET_CATEGORIES_QUERY)
    104         return c.fetchall()
    105 
    106     def get_storages(self):
    107         c = self.conn.cursor()
    108         c.execute(GET_STORAGES_QUERY)
    109         return c.fetchall()
    110 
    111     def list_parts(self):
    112         c = self.conn.cursor()
    113         c.execute(LIST_PARTS_QUERY)
    114         return c.fetchall()
    115 
    116     def list_parts_by_category(self, category):
    117         c = self.conn.cursor()
    118         c.execute(LIST_PARTS_CATEGORY_QUERY, (category, category))
    119         return c.fetchall()
    120 
    121     def list_parts_by_storage(self, storage):
    122         c = self.conn.cursor()
    123         c.execute(LIST_PARTS_STORAGE_QUERY, (storage, storage))
    124         return c.fetchall()
    125 
    126     def new_part(self, part):
    127         if part[9] is not None:
    128             part[9] = sqlite3.Binary(part[9])
    129         p = tuple(part)
    130         with self.conn:
    131             c = self.conn.cursor()
    132             c.execute(INSERT_PART_QUERY, p)
    133             return c.lastrowid
    134 
    135     def get_part(self, part_id):
    136         c = self.conn.cursor()
    137         c.execute(GET_PART_QUERY, (part_id,))
    138         return c.fetchone()
    139 
    140     def get_part_by_mpn(self, mpn):
    141         c = self.conn.cursor()
    142         c.execute(GET_PART_BY_MPN_QUERY, (mpn,))
    143         return c.fetchone()
    144 
    145     def search_parts(self, term):
    146         c = self.conn.cursor()
    147         c.execute(SEARCH_PARTS_QUERY, (term, term, term, term, term))
    148         return c.fetchall()
    149 
    150     def get_part_history(self, part_id):
    151         c = self.conn.cursor()
    152         c.execute(GET_HISTORY_QUERY, (part_id,))
    153         return c.fetchall()
    154 
    155     def new_part_history_event(self, part_id, movement, comment):
    156         with self.conn:
    157             c = self.conn.cursor()
    158             c.execute(INSERT_HISTORY_EVENT, (part_id, movement, comment))
    159 
    160     def update_part_qty(self, part_id, movement):
    161         with self.conn:
    162             c = self.conn.cursor()
    163             c.execute(GET_PART_QTY, (part_id,))
    164             qty = c.fetchone()["quantity"]
    165             qty = qty + movement
    166             c.execute(UPDATE_PART_QTY_QUERY, (qty, part_id))
    167 
    168     def delete_part(self, part_id):
    169         with self.conn:
    170             c = self.conn.cursor()
    171             c.execute(DELETE_HISTORY_QUERY, (part_id,))
    172             c.execute(DELETE_PART_QUERY, (part_id,))
    173 
    174     def get_image(self, part_id):
    175         c = self.conn.cursor()
    176         c.execute(GET_IMAGE_QUERY, (part_id,))
    177         return c.fetchone()
    178 
    179     def get_datasheet(self, part_id):
    180         c = self.conn.cursor()
    181         c.execute(GET_DATASHEET_QUERY, (part_id,))
    182         return c.fetchone()
    183 
    184     def close(self):
    185         self.conn.close()