1234567891011121314151617181920212223242526272829 |
- from ZlwlAlgosCommon.utils.ProUtils import *
- from ZlwlAlgosCommon.service.iotp.IotpAlgoService import IotpAlgoService
- cur_env = 'dev' # 设置运行环境
- app_path = "/home/wangliming/project/zlwl-algos" # 设置相对路径
- app_name = "schedule" # 应用名, 建议与topic的后缀相同
- sysUtils = SysUtils(cur_env, app_path)
- mysqlUtils = MysqlUtils()
- mysql_params = sysUtils.get_cf_param('mysql-algo')
- mysql_engine, mysql_Session= mysqlUtils.get_mysql_engine(mysql_params)
- mysql_conn = mysql_engine.connect()
- # with mysql_Session() as session:
- # session.add(Test2(id=12, field1=10))
- # session.commit()
- # 删除
- # with mysql_Session() as session:
- # session.execute(delete(Test2).where(or_((Test2.id == 10), (Test2.id == 12))))
- # session.commit()
- # 修改
- # with mysql_Session() as session:
- # session.execute(update(Test2).where(and_((Test2.id == 12), (Test2.field1 == 10))).values(id=15))
- # session.commit()
- # # upsert
- # insert_data = insert(Test2).valuess()
- # with mysql_Session() as session:
- # session.execute(update(Test2).where(and_((Test2.id == 12), (Test2.field1 == 10))).values(id=15))
- # session.commit()
|