zoukankan      html  css  js  c++  java
  • 封装操作mysql、redis

    封装操作mysql:

    import pymysql
    
    class MyDb:
        def __init__(self,host,password,user,db,port=3306,charset='utf8',autocommit=True):
            try:
                self.coon = pymysql.connect(host=host,password=password,
                                            user=user,db=db,port=port,
                                            charset=charset,autocommit=autocommit)
            except Exception as e:
                print('数据库连接失败!错误信息是%s'%e)
                raise Exception('数据库连接失败!错误信息是%s'%e)
                #主动抛出一个异常
            else:
                self.cur = self.coon.cursor(pymysql.cursors.DictCursor)
    
        def select_all(self,sql):
            try:
                self.cur.execute(sql)
            except Exception as e:
                res = 'sql语句错误,%s
    ,错误的sql是【%s】'%(e,sql)
                print(res)
            else:
                res = self.cur.fetchall()
            return res
        def select_one(self,sql):
            try:
                self.cur.execute(sql)
            except Exception as e:
                res = 'sql语句错误,%s
    ,错误的sql是【%s】'%(e,sql)
                print(res)
            else:
                res = self.cur.fetchone()
            return res
        def other_sql(self,sql):
            try:
                self.cur.execute(sql)
            except Exception as e:
                res = 'sql语句错误,%s
    ,错误的sql是【%s】'%(e,sql)
                print(res)
                return res
        def close(self):
            print('拜拜啦')
            self.cur.close()
            self.coon.close()
    
    
    
    my = MyDb('ip','fsdf','sdfsd','sdfsdf')
    my.select_one('select * from xx;')
    my.select_one('select * from xx;')
    my.select_one('select * from xx;')
    my.close()
    #1、先测能不能执行sql
    #2、数据连接异常的

    封装操作redis:

    import redis
    
    class MyRedis:
        def __init__(self,host,password,port=6379,db=0):
            self.r = redis.Redis(host=host,password=password,port=port,db=db)
        def op_str(self,k,v=None,ex=-1):
            if v:
                self.r.set(k,v,ex)#set
            else:
                res = self.r.get(k)
                if res:
                    return res.decode()
        def op_hash(self,name,k=None):
            if k:#判断是取单个值
                res = self.r.hget(name,k)
                if res:#判断是否有返回值
                    return res.decode()
            else:#取key里面所有的值
                res = {}
                for k,v in self.r.hgetall(name).items():
                    res[k.decode()] = v.decode()
                return res
        def my_type(self,k):
            return self.r.type(k).decode()
  • 相关阅读:
    Java 快速入门-06-JDK 目录文件说明
    Java快速入门-05-数组循环条件 实例《延禧攻略》
    腾讯云服务器 选购+远程控制 图文教程
    无法获得锁 /var/lib/dpkg/lock
    Ubuntu 安装 PhpMyAdmin 图文教程
    基于Redis的BloomFilter算法去重
    CAP理论
    Linux常用命令回顾
    基于Solr实现HBase的二级索引
    Solr搜索服务架构图
  • 原文地址:https://www.cnblogs.com/Noul/p/9392074.html
Copyright © 2011-2022 走看看