From 98718c546aa17f4cc88134bed21c032c5a9a8599 Mon Sep 17 00:00:00 2001 From: NiceActior <1452492851@qq.com> Date: Sun, 25 Aug 2024 22:00:03 +0800 Subject: [PATCH] wheel-sharing 03, python, redis-opt and mysql-opt shareable --- python/mysql_opt.py | 111 ++++++++++++++++++++++++++++++++++++++++++++ python/redis_opt.py | 79 +++++++++++++++++++++++++++++++ 2 files changed, 190 insertions(+) create mode 100644 python/mysql_opt.py create mode 100644 python/redis_opt.py diff --git a/python/mysql_opt.py b/python/mysql_opt.py new file mode 100644 index 0000000..479b868 --- /dev/null +++ b/python/mysql_opt.py @@ -0,0 +1,111 @@ +""" +@author donglm +@datetime 2024/8/24 15:18 +""" + +import pymysql +from pymysql import ProgrammingError + +global conn +envArr = {"int": ["mpython", "mpython", "127.0.0.1", "db_mpython", 3306, "utf8"], + "uat": ["mpython", "mpython", "127.0.0.1", "db_mpython", 3306, "utf8"]} +cmdArr = ["INSERT", "DELETE", "UPDATE"] + + +def conn_test(): + curs = conn.cursor() + curs.execute("SELECT version();") + rest = curs.fetchone() + print("Database version: %s" % rest) + conn.close() + pass + + +def opt_exec(): + print("opt_exec") + curs = conn.cursor() + path = "mpython_dml.sql" + with open(path, "r", encoding="utf-8") as file: + cont = file.read() + print(cont) + try: + eft = curs.execute(cont) + print("opt_exec effect row", eft) + print("opt_exec success") + conn.commit() + except ProgrammingError as err: + print("opt_exec failure") + print(err) + conn.rollback() + finally: + print("opt_exec finally") + pass + + +def opt_exec_many(): + print("opt_exec_many") + curs = conn.cursor() + path = "mpython_dml.sql" + with open(path, "r", encoding="utf-8") as file: + cont = file.read() + # print(cont) + try: + eft = curs.executemany(cont, []) + print("opt_exec_many effect row", eft) + print("opt_exec_many success") + conn.commit() + except ProgrammingError as err: + print("opt_exec_many failure") + print(err) + conn.rollback() + finally: + print("opt_exec_many finally") + pass + + +def opt_exec_part(): + print("opt_exec_part") + semicolon = ";" + curs = conn.cursor() + path = "mpython_dml.sql" + with open(path, "r", encoding="utf-8") as file: + cont = file.read() + # print(cont) + split_arr = cont.split(";") + print(len(split_arr)) + for part in split_arr: + part = part.lstrip() + # 跳过处理, 空行, 注释 + if (len(part) < 1) or (part.find("/**") >= 0): + # print(part) + continue + else: + # print(part + semicolon) + exec_sql(curs, part + semicolon) + pass + + +def exec_sql(curs, cont): + try: + eft = curs.execute(cont) + print("exec_sql effect row", eft) + print("exec_sql success") + conn.commit() + except ProgrammingError as err: + print("exec_sql failure") + print(err) + conn.rollback() + finally: + print("exec_sql finally") + + +if __name__ == '__main__': + # print("mysql_opt") + env = envArr["int"] + # global conn + conn = pymysql.connect(user=env[0], password=env[1], host=env[2], database=env[3], port=env[4], charset="utf8") + # conn_test() + # opt_exec() + # opt_exec_many() + opt_exec_part() + pass diff --git a/python/redis_opt.py b/python/redis_opt.py new file mode 100644 index 0000000..b4e092b --- /dev/null +++ b/python/redis_opt.py @@ -0,0 +1,79 @@ +""" +@author donglm +@datetime 2024/8/24 10:32 +""" + +import argparse +import redis + +envArr = {"int": ["127.0.0.1", 6379, 0, "password"], "uat": ["127.0.0.1", 6379, 1, "password"]} +optArr = ["get", "set", "del", "flushdb"] + + +# check args +def check(args): + opt = args.opt + sue = args.sue + key = args.key + if opt not in optArr: + raise ValueError("raise a error to you, opt not support", opt) + if opt == 'flushdb' and sue != 1: + raise ValueError("raise a error to you, opt must confirm", opt) + if opt == 'set': + raise ValueError("just for the sake of writing and playing, opt", opt) + if opt == 'del' and key is None: + raise ValueError("raise a error to you, opt must specify key", opt) + pass + + +def opt_get(args, conn): + get_key = conn.get(args.key) + print(get_key) + pass + + +def opt_del(args, conn): + del_key = conn.delete(args.key) + print(del_key) + pass + + +def opt_flushdb(args, conn): + flushdb_true = conn.flushdb(True) + print(flushdb_true) + pass + + +# convert specific instruction execution +def switch(args, conn): + opt = args.opt + if opt == 'get': + opt_get(args, conn) + elif opt == 'del': + opt_del(args, conn) + elif opt == 'flushdb': + opt_flushdb(args, conn) + else: + print("no reachable") + pass + + +if __name__ == '__main__': + # print("redis_opt") + parser = argparse.ArgumentParser(usage="python opt redis, support get set del flushdb", add_help=True) + parser.description = "Authorized by Donglm, det: determine" + parser.add_argument('-env', type=str, required=False, choices=['int', 'uat'], default='int', help='specify environ') + parser.add_argument('-opt', type=str, required=False, metavar="redis command", help='specify operate') + parser.add_argument('-key', type=str, required=False, metavar="cache key", help='specify key') + parser.add_argument('-val', type=str, required=False, metavar="cache val", help='specify val') + parser.add_argument('-bar', type=str, dest="what", nargs=1, help='specify bar') + parser.add_argument('-sue', type=int, required=False, metavar="0 not 1 det", choices=[0, 1], help='specify confirm') + args = parser.parse_args() + print(args) + check(args) + env = envArr[args.env] + # print(env) + # See the parameters specifically, host, port, db, encoding="utf-8", decode_responses=False + conn = redis.Redis(env[0], env[1], env[2], decode_responses=True) + # print(conn.ping()) + switch(args, conn) -- Gitee