partsdb

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

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()