library/library/database.py

450 lines
14 KiB
Python

import sqlite3
from library import permissions
from library.exceptions import *
# because of the way that SQLite works we need to have these two
# files be different, because the Office staff needs read/write
# permission to the directory that contains the checkout.db file.
# (sqlite needs to create temporary files in that directory)
_catalogue_db_file = '/users/libcom/catalogue.db'
_book_table = 'books'
_book_category_table='book_categories'
_category_table = 'categories'
_checkout_db_file = '/users/libcom/checkout/checkout.db'
_checkout_table = 'checked_out'
_return_table = 'returned'
_checkout_table_creation = '''
CREATE TABLE IF NOT EXISTS checked_out
(id INTEGER UNIQUE, uwid STRING, date_out DATETIME DEFAULT current_timestamp);
CREATE TABLE IF NOT EXISTS returned
(id INTEGER, uwid STRING, date_out DATETIME, date_in DATETIME DEFAULT current_timestamp);
'''
_book_table_creation = '''
CREATE TABLE IF NOT EXISTS books
(id INTEGER PRIMARY KEY AUTOINCREMENT,
isbn, lccn, title, subtitle, authors, edition,
publisher, publish_year, publish_month, publish_location,
pages, pagination, weight,
last_updated DATETIME DEFAULT current_timestamp,
deleted BOOLEAN DEFAULT 0);
CREATE TABLE IF NOT EXISTS categories
(cat_id INTEGER PRIMARY KEY, category STRING UNIQUE ON CONFLICT IGNORE);
CREATE TABLE IF NOT EXISTS book_categories
(id INTEGER, cat_id INTEGER);
'''
columns = ['id', 'isbn', 'lccn',
'title', 'subtitle', 'authors', 'edition',
'publisher', 'publish year', 'publish month', 'publish location',
'pages', 'pagination', 'weight', 'last updated', 'deleted']
_book_trigger_creation = '''
CREATE TRIGGER IF NOT EXISTS update_books_time AFTER UPDATE ON books
BEGIN
UPDATE books SET last_updated = DATETIME('NOW') WHERE rowid = new.rowid;
END;
CREATE TRIGGER IF NOT EXISTS delete_book AFTER DELETE ON books
BEGIN
DELETE FROM book_categories WHERE id = old.rowid;
END;
CREATE TRIGGER IF NOT EXISTS delete_category AFTER DELETE ON categories
BEGIN
DELETE FROM book_categories WHERE cat_id = old.cat_id;
END;
CREATE TRIGGER IF NOT EXISTS insert_book_category_time AFTER INSERT
ON book_categories
BEGIN
UPDATE books SET last_updated = DATETIME('NOW') WHERE id = new.id;
END;
'''
#################################
# character escaping, etc for sql queries
#################################
def _colify(s):
return s.replace(" ","_").lower()
def _stringify(v):
return '"' + str(v).strip().replace('"','""') + '"'
###################################
# book functions
##################################
@permissions.check_permissions(permissions.PERMISSION_LIBCOM)
def addBook(book):
conn = sqlite3.connect(_catalogue_db_file)
c = conn.cursor()
cols = []
vals = []
for k,v in book.items():
if v!="":
cols.append(_colify(k))
vals.append(_stringify(v))
query = ("INSERT INTO "+_book_table+" ("+", ".join(cols)+") VALUES ("+
", ".join(vals)+");")
c.execute(query)
conn.commit()
c.close()
@permissions.check_permissions(permissions.PERMISSION_LIBCOM)
def updateBook(book, bookID):
'''
Takes book attribute dictionary and a string representating the book ID
number, and returns updates the book accordingly
'''
conn = sqlite3.connect(_catalogue_db_file)
c = conn.cursor()
updates=[]
for k,v in book.items():
updates.append(_colify(k)+"="+_stringify(v))
query = ("UPDATE "+_book_table+" SET " + ", ".join(updates)+" WHERE id = " +
str(bookID)+";")
c.execute(query)
conn.commit()
c.close()
def get_books():
conn = sqlite3.connect(_catalogue_db_file)
c = conn.cursor()
query = "SELECT * FROM "+_book_table+" WHERE deleted=0;"
c.execute(query)
books = [_query_to_book(b) for b in c]
c.close()
return books
def getBooksByCategory(cat):
'''
Takes a string representating the category ID number, and returns
non-deleted books in that category
'''
conn = sqlite3.connect(_catalogue_db_file)
c = conn.cursor()
query = ("SELECT "+",".join(map(_colify,columns))+" FROM "+_book_table+
" JOIN "+_book_category_table+
" USING (id) WHERE cat_id = :id AND deleted=0;")
c.execute(query,cat)
books = [_query_to_book(b) for b in c]
c.close()
return books
def getRemovedBooks():
conn = sqlite3.connect(_catalogue_db_file)
c = conn.cursor()
query = "SELECT * FROM "+_book_table+" WHERE DELETED=1;"
c.execute(query)
books = [_query_to_book(b) for b in c]
c.close()
return books
def getUncategorizedBooks():
conn = sqlite3.connect(_catalogue_db_file)
c = conn.cursor()
query = ("SELECT "+",".join(map(_colify,columns))+" FROM "+_book_table+
" WHERE id NOT IN (SELECT id FROM "+_book_category_table+")"+
" AND deleted=0;")
c.execute(query)
books = [_query_to_book(b) for b in c]
c.close()
return books
def get_book(bookid):
conn = sqlite3.connect(_catalogue_db_file)
c = conn.cursor()
query = "SELECT * FROM "+_book_table+" WHERE id = "+str(bookid)+";"
c.execute(query)
book = _query_to_book(c.fetchone())
c.close()
return book
# removes book from catalogue
@permissions.check_permissions(permissions.PERMISSION_LIBCOM)
def removeBook(bookid):
conn = sqlite3.connect(_catalogue_db_file)
c = conn.cursor()
query = "UPDATE " +_book_table+ " SET deleted=1 WHERE id = "+str(bookid)+";"
c.execute(query)
conn.commit()
c.close()
@permissions.check_permissions(permissions.PERMISSION_LIBCOM)
def removeBooks(books):
conn = sqlite3.connect(_catalogue_db_file)
c = conn.cursor()
query1 = "UPDATE " +_book_table+ " SET deleted=1 WHERE id = :id;"
for book in books:
c.execute(query1, book)
conn.commit()
c.close()
# restores trashed books
@permissions.check_permissions(permissions.PERMISSION_LIBCOM)
def restoreBooks(books):
conn = sqlite3.connect(_catalogue_db_file)
c = conn.cursor()
query1 = "UPDATE " +_book_table+ " SET deleted=0 WHERE id = :id;"
for book in books:
c.execute(query1,book)
conn.commit()
c.close()
# fully deletes book from books table
@permissions.check_permissions(permissions.PERMISSION_LIBCOM)
def deleteBook(bookid):
conn = sqlite3.connect(_catalogue_db_file)
c = conn.cursor()
query = "DELETE FROM " +_book_table+ " WHERE id = "+str(bookid)+";"
c.execute(query)
conn.commit()
c.close()
@permissions.check_permissions(permissions.PERMISSION_LIBCOM)
def deleteBooks(books):
conn = sqlite3.connect(_catalogue_db_file)
c = conn.cursor()
query = "DELETE FROM " +_book_table+ " WHERE id = :id;"
for book in books:
c.execute(query, book)
conn.commit()
c.close()
def _query_to_book(book_query):
# Make a dict out of column name and query results.
# Empty entries return None, which are removed from the dict.
return dict(filter(lambda t:t[1], zip(columns,book_query)))
def _query_to_book_checkout(book_query):
# Make a dict out of column name and query results.
# Empty entries return None, which are removed from the dict.
b = _query_to_book(book_query)
b['uwid'] = book_query[-2]
b['date'] = book_query[-1]
return b
#########################################
# Category related functions
########################################
def getBookCategories(book):
conn = sqlite3.connect(_catalogue_db_file)
c = conn.cursor()
query = ("SELECT id,cat_id,category FROM "+_book_category_table+" JOIN "+
_category_table+" USING (cat_id) WHERE id = :id ;")
c.execute(query,book)
cats = []
for book_id,cat_id,cat_name in c:
cats.append({'id':book_id, 'cat_id':cat_id, 'category':cat_name})
c.close()
return cats
def isStickered(book):
cats = getBookCategories(book)
for c in cats:
if c['category'] == "Stickered":
return True
return False
@permissions.check_permissions(permissions.PERMISSION_LIBCOM)
def categorizeBook(book, cats):
conn = sqlite3.connect(_catalogue_db_file)
c = conn.cursor()
query = ("INSERT OR IGNORE INTO "+_book_category_table+
" (id,cat_id) VALUES (?, ?);")
for cat in cats:
args = (book['id'],cat['id'])
c.execute(query,args)
conn.commit()
c.close()
@permissions.check_permissions(permissions.PERMISSION_LIBCOM)
def categorizeBooks(cat, books):
conn = sqlite3.connect(_catalogue_db_file)
c = conn.cursor()
query = ("INSERT OR IGNORE INTO "+_book_category_table+
" (id,cat_id) VALUES (?, ?);")
exists_query = "SELECT * FROM "+_book_category_table+" WHERE (id = ? AND cat_id = ?);"
for book in books:
args = (book['id'],cat['id'])
c.execute(exists_query, args)
if len(c.fetchall()) == 0:
c.execute(query,args)
conn.commit()
c.close()
@permissions.check_permissions(permissions.PERMISSION_LIBCOM)
def uncategorizeBook(book, cats):
conn = sqlite3.connect(_catalogue_db_file)
c = conn.cursor()
query = "DELETE FROM "+_book_category_table+" WHERE (id = ? AND cat_id = ?);"
for cat in cats:
args = (book['id'],cat['id'])
c.execute(query,args)
conn.commit()
c.close()
@permissions.check_permissions(permissions.PERMISSION_LIBCOM)
def uncategorizeBooks(books, cat):
conn = sqlite3.connect(_catalogue_db_file)
c = conn.cursor()
query = "DELETE FROM "+_book_category_table+" WHERE (id = ? AND cat_id = ?);"
for book in books:
args = (book['id'],cat['id'])
c.execute(query,args)
conn.commit()
c.close()
def getCategories():
conn = sqlite3.connect(_catalogue_db_file)
c = conn.cursor()
query = "SELECT cat_id, category FROM "+_category_table+";"
c.execute(query)
cats = []
for cat_id,cat in c:
cats.append({'id':cat_id, 'category':cat})
c.close()
return cats
@permissions.check_permissions(permissions.PERMISSION_LIBCOM)
def addCategory(cat):
conn = sqlite3.connect(_catalogue_db_file)
c = conn.cursor()
query = ("INSERT OR IGNORE INTO "+_category_table+" (category) VALUES ("
+_stringify(cat)+");")
c.execute(query)
conn.commit()
c.close()
@permissions.check_permissions(permissions.PERMISSION_LIBCOM)
def deleteCategories(cats):
conn = sqlite3.connect(_catalogue_db_file)
c = conn.cursor()
query1 = "DELETE FROM " +_category_table+ " WHERE cat_id = :id;"
for cat in cats:
c.execute(query1, cat)
conn.commit()
c.close()
#########################################
# Book Checkout functions
#########################################
@permissions.check_permissions(permissions.PERMISSION_OFFICE)
def checkout_book(book_id, uwid):
book = get_book(book_id)
if isStickered(book):
raise StickeredError()
conn = sqlite3.connect(_checkout_db_file)
c = conn.cursor()
try:
query = "INSERT INTO " + _checkout_table + " (id, uwid) VALUES (?, ?);"
c.execute(query, (book_id, uwid))
except sqlite3.IntegrityError:
raise CheckoutError()
finally:
conn.commit()
c.close()
def get_checkout_status(book_id):
conn = sqlite3.connect(_checkout_db_file)
c = conn.cursor()
query = "SELECT uwid,date_out FROM "+ _checkout_table + " WHERE id = :id ;"
c.execute(query, {"id": book_id})
data = c.fetchone()
c.close()
if data: return {"uwid":data[0], "date_out":data[1]}
return None
@permissions.check_permissions(permissions.PERMISSION_OFFICE)
def return_book(book_id):
conn = sqlite3.connect(_checkout_db_file)
c = conn.cursor()
query = "SELECT uwid,date_out FROM "+ _checkout_table + " WHERE id = :id ;"
c.execute(query, {"id": book_id})
tmp = c.fetchone()
uwid = tmp[0]
date_out = tmp[1]
query = "INSERT INTO " + _return_table + " (id, uwid, date_out) VALUES (?, ?, ?);"
query2 = "DELETE FROM " + _checkout_table + " WHERE id= :id ;"
c.execute(query, (book_id, uwid, date_out))
c.execute(query2, {"id": book_id});
conn.commit()
c.close()
def get_checkedout_books():
'''
retrieves checked out books. The returned books also have the fields
uwid: ID of person who signed out the book, and
date: date when the book was checked out
'''
conn = sqlite3.connect(_catalogue_db_file)
c = conn.cursor()
query = 'ATTACH "' + _checkout_db_file + '" AS co'
c.execute(query)
query = ("SELECT "+",".join(map(_colify,columns))+",uwid,date_out FROM "+_book_table+
" JOIN co."+_checkout_table+
" USING (id) ;")
c.execute(query)
books = [_query_to_book_checkout(b) for b in c]
c.close()
return books
def get_onshelf_books():
'''
retrieves checked out books. The returned books also have the fields
uwid: ID of person who signed out the book, and
date: date when the book was checked out
'''
conn = sqlite3.connect(_catalogue_db_file)
c = conn.cursor()
query = 'ATTACH "' + _checkout_db_file + '" AS co'
c.execute(query)
query = ("SELECT "+",".join(map(_colify,columns))+" FROM "+_book_table+
" LEFT JOIN co."+_checkout_table+
" USING (id) WHERE uwid ISNULL;")
c.execute(query)
books = [_query_to_book(b) for b in c]
c.close()
return books
#########################################
# Database initialization
#########################################
def _createBooksTable():
conn = sqlite3.connect(_catalogue_db_file)
c = conn.cursor()
c.executescript(_book_table_creation)
conn.commit()
c.close()
def _createTriggers():
conn = sqlite3.connect(_catalogue_db_file)
c = conn.cursor()
c.executescript(_book_trigger_creation)
conn.commit()
c.close()
def _create_checkout_table():
conn = sqlite3.connect(_checkout_db_file)
c = conn.cursor()
c.executescript(_checkout_table_creation)
conn.commit()
c.close()
def initializeDatabase():
_createBooksTable()
_createTriggers()
_create_checkout_table()