DBTool
DBTool.py
from operator import truediv
import sqlite3
class DBTool(object):
def __init__(self, dbname):
"""
init func, create databse connection.
"""
self.conn = sqlite3.connect('%s.db' % dbname)
self.c = self.conn.cursor()
# query
def executeQuery(self, sql, obj):
return self.c.execute(sql, obj)
# insert
def executeInsert(self, sql, obj):
try:
self.c.execute(sql, obj)
i = self.conn.total_changes
except Exception as e:
print('executeInsert error:', e)
self.conn.rollback()
return False
finally:
self.conn.commit()
if i > 0:
return True
else:
return False
# update
def executeUpdate(self, sql, obj):
try:
self.c.execute(sql, obj)
i = self.conn.total_changes
except Exception as e:
print('executeQuery error:', e)
self.conn.rollback()
return False
finally:
self.conn.commit()
if i > 0:
return True
else:
return False
# delete
def executeDelete(self, sql, obj):
try:
self.c.execute(sql, obj)
i = self.conn.total_changes
except Exception as e:
print('executeDelete error: ', e)
self.conn.rollback()
return False
finally:
self.conn.commit()
if i > 0:
return True
else:
return False
# close
def close(self):
self.c.close()
self.conn.close()
fastapi
main.py
from ast import Try
from fastapi import FastAPI
import uvicorn
import sqlite3
from DBTool import DBTool
app = FastAPI()
@app.get("/")
def index():
return "hello fastapi~"
@app.get("/get_data")
def add(id, name):
return get_users(id, name)
def get_users(id, name):
# try:
# conn = sqlite3.connect('test.db')
# cursor = conn.cursor()
# res = cursor.execute('select * from user')
# arr = []
# for item in res:
# obj = {}
# obj["id"] = item[0]
# obj["name"] = item[1]
# arr.append(obj)
# cursor.close()
# conn.close()
# return arr
# except Exception as e:
# print(e)
# return []
# return 'ok'
db = DBTool('test')
obj = (id, name)
sql = "insert into user (id,name) values(?,?)"
res = db.executeInsert(sql,obj)
if res:
print('insert successfully')
else:
print('insert failure')
db.close()
return res
if __name__ == "__main__":
uvicorn.run(app=app)
