43 lines
1.1 KiB
Python
43 lines
1.1 KiB
Python
# -*- coding: utf-8 -*-
|
||
# @Time: 2025/05/19 9:41
|
||
|
||
import asyncio
|
||
from sqlor.dbpools import DBPools
|
||
from appPublic.jsonConfig import getConfig
|
||
|
||
async def tmp(ns={}):
|
||
"""
|
||
:param ns:
|
||
:return:
|
||
"""
|
||
# 增删改查用到的数据，格式dict
|
||
ns = {
|
||
'id': ns.get('id'),
|
||
"clustercode": ns.get('clustercode'),
|
||
}
|
||
db = DBPools()
|
||
async with db.sqlorContext('kboss') as sor:
|
||
try:
|
||
# sor.C sor.R sor.U
|
||
r_resutl = await sor.R('表名', nss)
|
||
|
||
#执行原生sql
|
||
y_sql = """update table set ... %s""" % ns['id']
|
||
result = await sor.sqlExe(y_sql, {})
|
||
|
||
# 返回默认格式
|
||
return {'status': True, 'msg': '操作成功', 'data': r}
|
||
|
||
except Exception as e:
|
||
# TODO 日志等操作
|
||
raise e
|
||
|
||
return {'status': False, 'msg': '操作失败'}
|
||
|
||
if __name__ == '__main__':
|
||
p = '路径：/kboss_code/'
|
||
config = getConfig(p)
|
||
DBPools(config.databases)
|
||
loop = asyncio.get_event_loop()
|
||
print(loop.run_until_complete(tmp()))
|