2024年10月17日 09:45 by wst
web开发由于项目之前已经存在，连接MySQL的时候采用了imysql这个包，虽然在编码中连接了多次数据库，速度看起来还可以。直到后来出现了数据库切换的场景，才发现切换后不生效。
随着项目的扩展，我们需要支持多商家的访问，但是每个商家都有自己的菜品，不可能混合用一个表，如果有用多个表，需要改写的代码太多。后来经过讨论，决定采用多数据库的策略。
每个请求带上数据库ID参数，服务端根据这个ID找到对应的数据库参数并连接，然后进行各种逻辑操作。
当所有代码都写好后，开始调试，发现不用的用户（数据库ID不同）看到的内容竟然是一样的。于是分析问题原因是什么？
1.Python代码编译读取被载入内存，当切换数据库连接参数没有重新链接？
答：于是每次切换重启下服务，发现生效了。但是这不是长久之计。
2.难道是imysql这个库有问题，它保持了连接没有切换？
答：把imysql换成pymysql后，发现切换数据库生效了。
采用pymysql这个包去连接数据库。
这里有一些代码供大家参考：
from asyncio.log import logger
import pymysql
import pymysql.cursors
import datetime
from .tools.usercenter import uc_db_get
class opmysql:
def __init__(self, store_id):
res = uc_db_get("/rel/store/list", {'id': store_id})
db_info = res['data']['dbinfo']
# print("db_info:", db_info)
self.conn = pymysql.connect(**db_info, cursorclass=pymysql.cursors.DictCursor)
def _tcond_(self, condition):
"转字典为条件"
cond = '1'
if isinstance(condition, int):
return cond
for key in condition:
nkey = "`%s`.`%s`"%tuple(key.split(".")) if "." in key else "`%s`"%key
if isinstance(condition[key], str):
cond += f" and {nkey}='{condition[key]}'"
elif isinstance(condition[key], list) or isinstance(condition[key], tuple):
if condition[key][0]=='in' and len(condition[key][1])>0:
tmp = ",".join([str(x) for x in condition[key][1]])
cond += f" and {nkey} {condition[key][0]} ({tmp})"
elif condition[key][0]=='in' and len(condition[key][1])==0:
continue
elif condition[key][0]=='between' and len(condition[key][1])>0:
tmp = " and ".join(["'%s'"%str(x) for x in condition[key][1]])
cond += f" and {nkey} {condition[key][0]} {tmp}"
elif condition[key][0]=='like':
cond += f" and {nkey} {condition[key][0]} '{condition[key][1]}'"
else:
cond += f" and {nkey} {condition[key][0]} {condition[key][1]}"
else:
cond += f" and {nkey}={condition[key]}"
return cond
def _updata_(self, up_data):
"转换为更新语句"
up_lis = []
for key in up_data:
up_lis.append((f"`{key}`='{up_data[key]}'" if isinstance(up_data[key], str) else f"`{key}`={up_data[key]}"))
return ",".join(up_lis)
def _insdata_(self, table_name, ins_data):
"转换字典为插入数据"
keys = []
values = []
for key in ins_data:
keys.append(f'`{key}`')
if isinstance(ins_data[key], str):
values.append(f"'{ins_data[key]}'")
elif isinstance(ins_data[key], datetime.datetime):
values.append("'%s'"%ins_data[key].strftime("%Y-%m-%d %X"))
else:
values.append(f"{ins_data[key]}")
sql = f"insert into {table_name}({','.join(keys)}) values({','.join(values)})"
return sql
def get_data_count(self, tb_name):
try:
cursor = self.conn.cursor()
cursor.execute(f"select count(*) as count from {tb_name}")
res = cursor.fetchone()
count = res['count']
cursor.close()
return count
except Exception as e:
logger.error(f'Failed to get_data_count into {tb_name}: {e}')
def get_data_count_by_condition(self, tb_name, condition):
try:
cursor = self.conn.cursor()
sql = f"select count(*) as count from {tb_name} where {self._tcond_(condition)};"
print("page-count:", sql)
cursor.execute(sql)
res = cursor.fetchone()
count = res['count']
cursor.close()
return count
except Exception as e:
logger.error(f'Failed to get_data_count into {tb_name}: {e}')
def insert_one_Data(self, tb_name, ins_data):
# try:
cursor = self.conn.cursor()
sql = self._insdata_(tb_name, ins_data)
# print("insert_one_Data-sql:", sql)
cursor.execute(sql)
ins_id = self.conn.insert_id()
print("insert_id:", ins_id)
self.conn.commit()
cursor.close()
return ins_id
# except Exception as e:
# logger.error(f'Failed to insert_on_data into {tb_name}: {e}')
# return False
def find_all_Data(self, tb_name):
try:
cursor = self.conn.cursor()
cursor.execute(f"select * from {tb_name}")
res = cursor.fetchall()
cursor.close()
return res
except Exception as e:
logger.error(f'Failed to find_all_Data into {tb_name}: {e}')
def find_Data_byid(self, tb_name,id):
try:
cursor = self.conn.cursor()
cursor.execute(f"select * from {tb_name} where id={id}")
results = cursor.fetchall()
cursor.close()
return results
except Exception as e:
logger.error(f'Failed to find_Data_byid into {tb_name}: {e}')
def find_Data_By_Condition(self, tb_name,condition):
try:
cursor = self.conn.cursor()
sql = f"select * from {tb_name} where {self._tcond_(condition)};"
# print("find_Data_By_Condition-sql:", sql)
cursor.execute(sql)
results = cursor.fetchall()
cursor.close()
return results
except Exception as e:
logger.error(f'Failed to find_Data_By_Condition into {tb_name}: {e}')
def find_Data_By_Condition_order(self, tb_name,order,condition):
try:
cursor = self.conn.cursor()
sql = f"select * from {tb_name} where {self._tcond_(condition)} order by {order};"
cursor.execute(sql)
results = cursor.fetchall()
cursor.close()
return results
except Exception as e:
logger.error(f'Failed to find_Data_By_Condition into {tb_name}: {e}')
def find_Data_By_Condition_one(self, tb_name,condition):
try:
cursor = self.conn.cursor()
cursor.execute(f"select * from {tb_name} where {self._tcond_(condition)};")
results = cursor.fetchone()
cursor.close()
return results
except Exception as e:
logger.error(f'Failed to find_Data_By_Condition_one into {tb_name}: {e}')
def find_Query_Page_Data(self, tb_name, order, start, end, condif=1): # 分页查询获取数据
try:
cursor = self.conn.cursor()
sql = f"select * from {tb_name} where {self._tcond_(condif)} order by {order} limit {start},{end};"
print("sql:", sql)
cursor.execute(sql)
results = cursor.fetchall()
cursor.close()
return results
except Exception as e:
logger.error(f'Failed to find_Query_Page_Data into {tb_name}: {e}')
def update_Data_byid(self, tb_name, id, data):
try:
cursor = self.conn.cursor()
sql = f"update {tb_name} set {self._updata_(data)} where id={id};"
count = cursor.execute(sql)
self.conn.commit()
cursor.close()
return count
except Exception as e:
logger.error(f'Failed to update_Data_byid into {tb_name}: {e}')
return False
def update_Data_by_condf(self, tb_name, condf,data):
try:
cursor = self.conn.cursor()
count = cursor.execute(f"update {tb_name} set {self._updata_(data)} where {self._tcond_(condf)};")
self.conn.commit()
cursor.close()
return count
except Exception as e:
logger.error(f'Failed to update_Data_by_condf into {tb_name}: {e}')
def del_Data(self, tb_name, id):
try:
cursor = self.conn.cursor()
cursor.execute(f"delete from {tb_name} where id={id};")
self.conn.commit()
cursor.close()
except Exception as e:
logger.error(f'Failed to del_Data into {tb_name}: {e}')
def del_Data_condf(self, tb_name, condf):
try:
cursor = self.conn.cursor()
cursor.execute(f"delete from {tb_name} where {self._tcond_(condf)};")
self.conn.commit()
cursor.close()
return True
except Exception as e:
logger.error(f'Failed to del_Data_condf into {tb_name}: {e}')
return False
def select_Data_like(self, tb_name, key,value):
try:
cursor = self.conn.cursor()
cursor.execute(f"select * from {tb_name} where {key} {'%'+str(value)+'%'}")
results = cursor.fetchall()
cursor.close()
return results
except Exception as e:
logger.error(f'Failed to select_Data_like into {tb_name}: {e}')
def add_Data(self, tb_name, data_info):
try:
cursor = self.conn.cursor()
cursor.execute(self._insdata_(tb_name, data_info))
cursor.execute("SELECT LAST_INSERT_ID() as last_id;")
idres = cursor.fetchone()
self.conn.commit()
cursor.close()
return idres['last_id']
except Exception as e:
logger.error(f'Failed to add_Data into {tb_name}: {e}')
def get_last_id(self,tb_name):
try:
cursor = self.conn.cursor()
sql = "SELECT MAX(id) as last_id FROM %s;" % (tb_name)
cursor.execute(sql)
res = cursor.fetchone()
cursor.close()
return res['last_id']
except Exception as e:
logger.error(f'Failed to get_last_id into {tb_name}: {e}')
def __del__(self):
try:
self.conn.close()
except Exception as e:
logger.error(f"close mysql error:{e}")
说明：这个是在imysql的影响下封装的类。细心的读者可以发现，有个转换查询条件的成员函数（_tcond_）比较复杂。因为在imysql中查询条件是以字典的形式传递的，为了适配这个才这么写的。
如果您有什么高见，欢迎评论区留言。