mzinfo/V2/app/utils.py

#!/usr/bin/python
# -*- coding=utf-8 -*-
# @Create Time:		2024-01-11 14:57:50
# @Last Modified time: 2024-02-26 14:44:28
import dmPython
from flask_redis import FlaskRedis


class RedisUtils(object):
    def __init__(self, app):
        self.redis_client = FlaskRedis()
        self.redis_client.init_app(app)

    def set(self, key, value):
        self.redis_client.set(key, value)

    def get(self, key):
        value = self.redis_client.get(key)
        if value:
            return value.decode()
        else:
            return None

    def keys(self) -> list:
        keys = self.redis_client.keys()
        if keys:
            return [key.decode() for key in keys]
        else:
            return None

    def delete(self, key):
        result = self.redis_client.delete(key)
        return result


# class DM8Utils(object):
#     def __init__(self):
#         self.conn = dmPython.connect(user='SYSDBA', password='SYSDBA001', server='192.168.1.202', port=30236)

#     def query(self, sql_cmd: str, args=None):
#         '''以数据框形式返回查询据结果'''
#         try:
#             self.cursor = self.conn.cursor()
#             if args:
#                 self.cursor.execute(sql_cmd, args)
#             else:
#                 self.cursor.execute(sql_cmd)
#             data = self.cursor.fetchall()  # 以元组形式返回查询数据
#             header = [t[0] for t in self.cursor.description]
#             df = pd.DataFrame(list(data), columns=header)  # pd.DataFrame 对列表具有更好的兼容性
#         except Exception as e:
#             print(e)
#         finally:
#             self.cursor.close()
#         return df

#     def execute(self, sql_cmd: str, args):
#         try:
#             self.cursor = self.conn.cursor()
#             # 执行sql语句
#             self.cursor.execute(sql_cmd, args)
#             self.conn.commit()
#             status = True
#         except Exception as e:
#             # 发生错误时回滚
#             self.conn.rollback()
#             print(e)
#             status = False
#         finally:
#             self.cursor.close()
#         return status

#     def __del__(self):
#         self.conn.close()