注册 登录  
 加关注
   显示下一条  |  关闭
温馨提示!由于新浪微博认证机制调整,您的新浪微博帐号绑定已过期,请重新绑定!立即重新绑定新浪微博》  |  关闭

且行且记录

点滴记录,行的更远!

 
 
 

日志

 
 

SQLITE练习脚本  

2016-08-15 09:42:25|  分类: 默认分类 |  标签: |举报 |字号 订阅

  下载LOFTER 我的照片书  |
做了一个用于编辑sqlite数据库的py脚本。
简陋,用于练习数据库,方便操作。
#!/usr/bin/env python
# -*- coding:gb2312 -*-
# hcper@2016.8.15 v0.4
import sys
import sqlite3

class Cmd:
def __init__(self):
self.path = ''
self.con = None
self.cur = None
self.cmd = []
self.cmdset = {
'help':self.onhelp, 'exit':self.onexit, 'open':self.onopen, 'close':self.onclose,
'newtbl':self.onnewtbl, 'deltbl':self.ondeltbl, 'insrow':self.oninsrow, 'delrow':self.ondelrow,
'renew':self.onrenew, 'query':self.onquery, 'alltbl':self.onalltbl, 'dosql':self.ondosql }
def GetCmd(self):
print('[ 帮助 help; 退出 exit ]')
s = input('输入命令: ')
self.cmd = s.lower().split()
def DoCmd(self):
if len(self.cmd) == 0: return
if self.cmd[0] in self.cmdset:
self.cmdset[ self.cmd[0] ]()
else:
print('[-] 未知命令!')
def CmdLoop(self):
while True:
self.GetCmd()
try:
self.DoCmd()
except Exception as err:
print('[-] 发生异常!', err)
continue;
def onhelp(self):
h = """
===Sqlite3数据库操作测试===
打开数据库 open db_path
关闭数据库 close
新建表 newtbl table_name
删除表 deltbl table_name
插入数据 insrow
删除数据 delrow
更新数据 renew
查询数据 query
查所有表 alltbl
执行SQL语句 dosql
"""
print(h)
def onexit(self):
sys.exit(0)
def onopen(self):
if len(self.cmd) != 2:
print('help: open db_path')
return
try:
if self.con is None:
self.con = sqlite3.connect(self.cmd[1])
self.con.isolation_level = None
self.cur = self.con.cursor()
self.path = self.cmd[1]
print('[+] 数据库已经打开: ', self.path)
else:
print('[*] 数据库已经打开过了!', self.path)
return
except sqlite3.OperationalError as err:
print('[-] exception!', err)
return
def onclose(self):
if self.con is not None:
self.cur.close()
self.cur = None
self.con.close()
print('[+] 数据库已经关闭!', self.path)
self.con = None
self.path = ''
def onnewtbl(self):
if len(self.cmd) != 2:
print('help: newtbl table_name')
return
if self.con is not None:
print('输入各列的名称,以竖线分隔。')
s = input('输入>>')
col_names = s.split('|')
cn = len(col_names)
if cn == 0 : return
print('输入各列的类型,以竖线分隔,如: NULL|INTEGER|REAL|TEXT|BLOB')
s = input('输入>>')
col_types = s.split('|')
ct = len(col_types)
if ct == 0 : return
if cn != ct:
print('[-] 名称和类型数量不等!')
return
s = ''
for i in range(cn):
s += col_names[i] + ' ' + col_types[i] + ', '
s = s[0:-2]
sql = "create table {0} ({1})".format(self.cmd[1], s)
self.con.execute(sql)
print('[+] 表创建成功!', self.cmd[1])
else:
print('[-] 请先打开数据库!')
def ondeltbl(self):
if len(self.cmd) != 2:
print('help: deltbl table_name')
return
if self.con is not None:
sql = "drop table " + self.cmd[1]
self.con.execute(sql)
print('[+] 表删除成功!', self.cmd[1])
else:
print('[-] 请先打开数据库!')
def oninsrow(self):
if self.con is not None:
sql = "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name"
self.cur.execute(sql)
rs = self.cur.fetchall()
ts = [x[0] for x in rs]
tn = len(ts)
if tn == 0 : return
for i in range(tn):
print(repr(i) + ': ' + ts[i])
j = int(input('选择要插入到的表: '))
if j not in range(tn): return
tbl = ts[j]

sql = "PRAGMA table_info(%s)" % tbl
self.cur.execute(sql)
rs = self.cur.fetchall()
cols = [x[1] for x in rs]
typs = [x[2] for x in rs]
print('索引', '列名', '类型')
for i in range(len(cols)):
print(i, cols[i], typs[i])
sc = ""
sv = ""
for i in range(len(cols)):
s = input('请输入' + cols[i] + '的值: ')
if typs[i] == 'text':
s = "'" + s + "'"
sc += cols[i] + ', '
sv += s + ', '
sc = sc[0:-2]
sv = sv[0:-2]
sql = "insert into {0}({1}) values({2})".format(tbl,sc,sv)
self.con.execute(sql)
print('[+] 数据插入成功!')
else:
print('[-] 请先打开数据库!')
def ondelrow(self):
if self.con is not None:
sql = "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name"
self.cur.execute(sql)
rs = self.cur.fetchall()
ts = [x[0] for x in rs]
tn = len(ts)
if tn == 0 : return
for i in range(tn):
print(repr(i) + ': ' + ts[i])
j = int(input('选择要更新的表: '))
if j not in range(tn): return
tbl = ts[j]

sql = "PRAGMA table_info(%s)" % tbl
self.cur.execute(sql)
rs = self.cur.fetchall()
cols = [x[1] for x in rs]
typs = [x[2] for x in rs]
print('索引', '列名', '类型')
for i in range(len(cols)):
print(i, cols[i], typs[i])

s = input('请输入条件: ')
if s == '':
s = '1=1'

sql = "delete from {0} where {1}".format(tbl, s)
self.con.execute(sql)
print('[+] 数据删除成功!')
else:
print('[-] 请先打开数据库!')
def onrenew(self):
if self.con is not None:
sql = "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name"
self.cur.execute(sql)
rs = self.cur.fetchall()
ts = [x[0] for x in rs]
tn = len(ts)
if tn == 0 : return
for i in range(tn):
print(repr(i) + ': ' + ts[i])
j = int(input('选择要更新的表: '))
if j not in range(tn): return
tbl = ts[j]

sql = "PRAGMA table_info(%s)" % tbl
self.cur.execute(sql)
rs = self.cur.fetchall()
cols = [x[1] for x in rs]
typs = [x[2] for x in rs]
print('索引', '列名', '类型')
for i in range(len(cols)):
print(i, cols[i], typs[i])

cv = ""
for i in range(len(cols)):
print('输入回车跳过!')
s = input('请输入更新' + cols[i] + '的值: ')
if s == '': continue
if typs[i] == 'text':
s = "'" + s + "'"
cv += cols[i] + '=' + s + ', '
cv = cv[0:-2]

s = input('请输入条件: ')
if s == '':
s = '1=1'

sql = "update {0} set {1} where {2}".format(tbl, cv, s)
self.con.execute(sql)
print('[+] 数据更新成功!')
else:
print('[-] 请先打开数据库!')
def onquery(self):
if self.cur is not None:
sql = "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name"
self.cur.execute(sql)
rs = self.cur.fetchall()
ts = [x[0] for x in rs]
tn = len(ts)
if tn == 0 : return
for i in range(tn):
print(repr(i) + ': ' + ts[i])
j = int(input('选择要查询的表: '))
if j not in range(tn): return
tbl = ts[j]

sql = "select * from {0}".format(tbl)
self.cur.execute(sql)
rs = self.cur.fetchall()
print(rs)
print('[+] 数据查询成功!')
def onalltbl(self):
# 查询所有表 SELECT name FROM sqlite_master WHERE type='table' ORDER BY name
# 查询表信息 PRAGMA table_info(catalog)
if self.cur is not None:
sql = "select * from sqlite_master WHERE type = 'table'"
self.cur.execute(sql)
rs = self.cur.fetchall()
print(rs)
print('[+] 查询所有表成功!')
def ondosql(self):
if self.cur is not None:
sql = input('输入SQL语句>> ')
self.cur.execute(sql)
rs = self.cur.fetchall()
print(rs)
print('[+] SQL执行成功!')

#######################
cmd = Cmd()
cmd.CmdLoop()
  评论这张
 
阅读(7)| 评论(0)
推荐 转载

历史上的今天

在LOFTER的更多文章

评论

<#--最新日志,群博日志--> <#--推荐日志--> <#--引用记录--> <#--博主推荐--> <#--随机阅读--> <#--首页推荐--> <#--历史上的今天--> <#--被推荐日志--> <#--上一篇,下一篇--> <#-- 热度 --> <#-- 网易新闻广告 --> <#--右边模块结构--> <#--评论模块结构--> <#--引用模块结构--> <#--博主发起的投票-->
 
 
 
 
 
 
 
 
 
 
 
 
 
 

页脚

网易公司版权所有 ©1997-2017