概要
Pythonの標準ライブラリである「SQLite3」のラッパーライブラリを書いてみた。
テーマは、SQL文を書くことなく、SQL操作を行うことで、
SQLがわからない場合でもある程度使うことはできると思う。
作成するきっかけ
最近流行っているノーコードに触れる機会があり、
実際にはGUI上からの簡単な操作で、データ変更や削除、抽出といったデータベース操作を行うことができ、
それに近い形で、Pythonでの操作はできないかと思い、作成しようと思った。
実際に、Python言語から、ライブラリを呼び出す形で使用するため、
残念ながら”ノーコード”ではない。
実際のコード
sqlite_easy.py
import sqlite3
import os.path
class sqlite_easy():
def __init__(self):
self.con = ''
self.tableName = ''
self.selectList = []
self.updateList = []
self.whereList = []
def connect(self, databaseName):
"""
After confirming the existence of the database, connect if it exists.
Parameters
----------
databaseName : str
The name of the database that exists.
"""
if not os.path.exists(databaseName):
raise FileNotFoundError(databaseName)
self.con = sqlite3.connect(databaseName)
def createConnect(self, databaseName):
"""
If the database is not found, create and connect.
Parameters
----------
databaseName : str
The name of the database.
"""
self.con = sqlite3.connect(databaseName)
def createTable(self, tableName, *args):
"""
Create table.
Parameters
----------
args : tuple
Many columns in the table to create.
"""
if not self.con:
raise Exception('Not connect database.')
createquery = "CREATE TABLE {0} {1}".format(tableName, args)
cur = self.con.cursor()
cur.execute(createquery)
self.con.commit()
def setTable(self, tableName):
"""
Set table name.
Parameters
----------
tableName : str
The name of the table
"""
self.tableName = tableName
def setUpdate(self, columnName, value):
"""
Set update value.
Parameters
----------
columnName : str
The name of the columnname.
value : str or int
Update value.
"""
self.updateList.append('{0} = {1}'.format(columnName, value))
def setSelectColumn(self, columnName):
"""
Set select column.
Parameters
----------
columnName : str
The name of the columnname.
"""
self.selectList.append(columnName)
def setWhere(self, columnName, operator, value):
"""
Set the where clause for selection, update, and deletion.
Parameters
----------
columnName : str
Where clause column name.
operator : str
The available values are dependent on the data type of the value parameter
column = value : column data and value equal.
column <> value : column data and value not equal.
column > value : column data is greater than the value.
column >= value : column data is equal to or greater than the value.
column < value : value is greater than the column data.
column <= value : value is equal to or greater than the column data.
"""
if not operator in ('=', '<>', '>', '<', '>=', '<='):
raise Exception('For operator, please set the following. =, <>, >, >=, <, <=')
if not len(self.whereList):
self.whereList.append("WHERE {0} {1} {2}".format(columnName, operator, value))
else:
self.whereList.append("AND {0} {1} {2}".format(columnName, operator, value))
def initializeWhereList(self):
"""
Initialize where clause list.
"""
self.whereList = []
def initializeSelectList(self):
"""
Initialize select list.
"""
self.selectList = []
def initializeUpdateList(self):
"""
Initialize update list.
"""
self.updateList = []
def initializeAll(self):
"""
Initialize all setList,setTable and connection.
"""
if self.con:
self.con.close()
self.selectList = []
self.tableName = ''
self.whereList = []
self.updateList = []
self.con = ''
print('Complete Reset.')
def showTable(self):
"""
View all table list.
"""
if not self.con:
raise Exception('Not connect database.')
selectquery = "SELECT name FROM sqlite_master WHERE type='table'"
cur = self.con.cursor()
cur.execute(selectquery)
datas = cur.fetchall()
if not datas:
print('Not found:(')
else:
for data in datas:
print(data)
def showColumn(self):
"""
View all table columns.
"""
if not self.con:
raise Exception('Not connect database.')
if not self.tableName:
raise Exception('Execute setTable() and try again.')
selectquery = 'PRAGMA table_info({0})'.format(self.tableName)
cur = self.con.cursor()
cur.execute(selectquery)
datas = cur.fetchall()
if not datas:
print('Not Found:(')
else:
for data in datas:
print(data)
def selectAll(self):
"""
View all data in the table.
Parameters
----------
tableName : str
The name of the table.
"""
if not self.con:
raise Exception('Not connect database.')
if not self.tableName:
raise Exception('Execute setTable() and try again.')
selectquery = "SELECT * FROM {}".format(self.tableName)
cur = self.con.cursor()
cur.execute(selectquery)
datas = cur.fetchall()
if not datas:
print('No data.')
else:
for data in datas:
print(data)
def select(self, resetFlg = 0):
"""
View select data in the table.
Parameters
----------
resetFlg : int
default 0.
"""
if not self.con:
raise Exception('Not connect database.')
if not self.tableName:
raise Exception('Execute setTable() and try again.')
if not resetFlg in (0,1):
raise Exception('If you want to perform a reset, set up 1.')
selectquery = "SELECT "
if not self.selectList:
self.selectList.append('*')
for column in self.selectList:
if self.selectList[-1] == column:
selectquery = selectquery + column + ' '
else:
selectquery = selectquery + column + ","
selectquery += 'FROM {0}'.format(self.tableName)
for where in self.whereList:
selectquery += ' ' + where
cur = self.con.cursor()
cur.execute(selectquery)
datas = cur.fetchall()
if not datas:
print('No data.')
else:
for data in datas:
print(data)
if resetFlg:
self.initializeAll()
def insert(self, *args):
"""
Insert data in the table.
Parameters
----------
tableName : str
The name of the table.
args : tuple
The number of values that were in the columns of the table.
"""
if not self.con:
raise Exception('Not connect database.')
if not self.tableName:
raise Exception('Execute setTable() and try again.')
if not args:
raise Exception('Not insert value has been set.')
insertquery = "INSERT INTO {0} VALUES {1}".format(self.tableName, args)
cur = self.con.cursor()
cur.execute(insertquery)
self.con.commit()
def updateAll(self):
"""
Update data in the table.
"""
if not self.con:
raise Exception('Not connect database.')
if not self.tableName:
raise Exception('Execute setTable() and try again.')
if not self.updateList:
raise Exception('The column and value to update are not set.')
updatequery = "UPDATE {0} SET ".format(self.tableName)
for query in self.updateList:
updatequery = updatequery + query + ","
updatequery = updatequery.rstrip(',')
cur = self.con.cursor()
cur.execute(updatequery)
self.con.commit()
def update(self):
"""
Update data in the table with condition.
"""
if not self.con:
raise Exception('Not connect database.')
if not self.tableName:
raise Exception('Execute setTable() and try again.')
if not self.whereList:
raise Exception('Execute setWhere() and try again.')
if not self.updateList:
raise Exception('Execute setUpdate() and try again.')
updatequery = "UPDATE {0} SET ".format(self.tableName)
for query in self.updateList:
updatequery = updatequery + query + ","
updatequery = updatequery.rstrip(',')
updatequery = updatequery + ' '
for where in self.whereList:
if self.whereList[-1] == where:
updatequery = updatequery + where
else:
updatequery = updatequery + where
cur = self.con.cursor()
cur.execute(updatequery)
self.con.commit()
def deleteAll(self):
"""
Delete data in the table.
"""
if not self.con:
raise Exception('Not connect database.')
if not self.tableName:
raise Exception('Execute setTable() and try again.')
executeFlg = input('If you run it, all the records will be deleted, but will you really run it? y/n:')
if executeFlg == 'n':
print("Cancelld the execution.")
else:
deletequery = "DELETE FROM {0}".format(self.tableName)
cur = self.con.cursor()
cur.execute(deletequery)
self.con.commit()
def delete(self):
"""
Delete data in the table with condition.
"""
if not self.con:
raise Exception('Not connect database.')
if not self.tableName:
raise Exception('Execute setTable() and try again.')
if not self.whereList:
raise Exception('Execute setWhere() and try again.')
deletequery = "DELETE FROM {0} ".format(self.tableName)
for where in self.whereList:
deletequery += where
cur = self.con.cursor()
cur.execute(deletequery)
self.con.commit()
使い方
Pythonインタプリタを起動した後
全てのCRUDに共通して必要な処理
import sqlite_easy
sql = sqlite_easy.sqlite_easy()
sql.connect('データベース名')
テーブルのレコードを全件取得
sql.setTable('テーブル名')
sql.selectAll()
テーブルのレコードを条件に従い取得
※レコードの一部カラムを取得したい場合は、setSelectColumn()を使用して、取得するカラムを設定することができる。
sql.setTable('テーブル名')
sql.setWhere('カラム名', '比較演算子', '条件の値')
sql.select()
テーブルにレコードを挿入
sql.setTable('テーブル名')
sql.insert('値1', '値2', ...)
テーブルのレコード全件の上書き
sql.setTable('テーブル名')
sql.setUpdate('カラム名', '値')
sql.updateAll()
テーブルのレコード全件の削除
※途中で、入力を求められ、nを入力すると、実行を中止します。
sql.setTable('テーブル名')
sql.deleteAll()
If you run it, all the records will be deleted, but will you really run it? y/n:y
テーブルの新規作成
sql.createTable('テーブル名', 'カラム1', 'カラム2', ...)
テーブル一覧取得
sql.showTable()
テーブルのカラム一覧取得
sql.setTable('テーブル名')
sql.showColumn()
編集後記
今回作成している途中で、一度作り直していて、作成テストをしている段階で、2週間くらいかかってしまいました。
ちゃんと設計してから作らないとだめだなぁとしみじみしたのですが、
やっぱり作っているとこうした方が良かったなぁとか思うことはありました。
完璧に作るって難しいですね。
例えば、毎回どの関数でも同じようなバリデーションを行なっていますが、
その部分をcheck()みたいな関数を作成して、check()関数で弾かれた場合はエラーを出力するような作りの方が、全体的なコードの量は少なくなったと思いますし、
各CRUD関数内の実行部分も全て同じなので、
実行関数を用意して呼び出すという作りにしておけば、毎回書く必要もなかったと。
まだまだ、改善の余地があるsqlite_easyですが、ぜひ使ってみて、見てみて、
ここバグあんぞや、こうした方がいいんじゃないか？等ありましたら、是非コメントいただければと思います。