当前位置：首页 > ai >正文

python连接sqllite数据库工具类

ai 2025/5/10 17:30:50

背景

在数据集成业务中, 有很多token是有短效的到期时间的. 需要在调用多个接口的时候统一获取token ,因为我们集成平台的执行客户端是分步式的,集中保存平台大量客户的token在服务端有性能瓶颈.所以一般在客户端本地通过sqllite存储,故写了一个调用sqllite的工具类,在后续分享的集成代码中也会使用到.

直接上代码

#-*- coding:utf-8 -*-
#Author：Evan
#Mail：evan#chengwenit.com# sqllite数据库工具类import sqlite3
import time
import cjcconfigclass sqllitToolSql():"""sqllite数据库工具类方便其他类继承此类复用方法"""def __init__(self,filename):"""初始化数据库,定义文件名："""self.filename = filename + ".db"self.db = sqlite3.connect(self.filename)self.c = self.db.cursor()def close(self):"""关闭数据库"""self.c.close()self.db.close()def execute(self,sql,param=None):"""执行数据库的增、删、改sql：执行的SQL语句param：list或tuple类型参数，可为Noneretutn：成功返回True，失败返回False"""try:if param is None:self.c.execute(sql)else:if type(param) is list:self.c.executemany(sql,param)else :self.c.execute(sql,param)count = self.db.total_changesself.db.commit()except Exception as e:print(e)return False,eif count > 0 :return Trueelse :return Falsedef query(self,sql,param=None):"""查询语句sql：执行的SQL语句param：参数,可为Noneretutn：成功返回True"""if param is None:self.c.execute(sql)else:self.c.execute(sql,param)return self.c.fetchall()if __name__ == "__main__":"""初始化代码，第一次初始化建表用，初始化后注释"""sql = sqllitToolSql(cjcconfig.sqldbname)## 创建APP表#f = sql.execute("""CREATE TABLE IF NOT EXISTS chw_qw_app(#     appsecret TEXT PRIMARY KEY,#     apptoken TEXT,#     synctime TEXT,#     memo TEXT);""")## 创建客户表#f = sql.execute("""CREATE TABLE IF NOT EXISTS chw_qw_cust(#     custuserid TEXT,#     username TEXT,#     corpname TEXT,#     gender TEXT,#     position TEXT,#     userid TEXT,#     type TEXT,#     createtime TEXT,#     tag_id	 TEXT,#     mobiles	 TEXT,#     add_way	 TEXT,#     record_id	 TEXT,#     PRIMARY KEY (custuserid, userid));""") ## 写入需要初始化的数据#applist = cjcconfig.applist#for i in range(len(applist)):#    sql.execute("INSERT INTO chw_qw_app (appsecret,apptoken,synctime,memo) values(?,?,?,?);",(applist[i]["appsecret"],'',str(time.time()),applist[i]["appname"]))## 带参数插入和查询示例#sql.execute("insert into chw_qw_cust (custuserid,userid,record_id) values (?,?,?);",('a001','evan','x999'))#sql.execute("replace into chw_qw_cust (custuserid,userid,type) values (?,?,?);",('a001','evan','101'))##res = sql.query("select * from chw_qw_app where id=?;",())### 删除表示例#f = sql.execute("""DROP TABLE chw_qw_cust;""")## 查询初始化数据res = sql.query("select count(*) from chw_qw_cust where record_id is null limit 10;")#res = sql.query("select count(*) from chw_qw_cust where custuserid=? and userid=?;",('231','123'))print(res)# 关闭连接sql.close()