python基于pymysql连接数据库

古城微笑少年丶 2022-07-13 21:10 259阅读 0赞

\#encoding=utf-8  
  
  
import pymysql as mysql\_module  
import queue  
import threading  
  
  
class sql\_operate():  
    dcur = 0  
    USER = 0  
    PASSWD = 1  
    HOST = 2  
    DB = 3  
    CHAR\_SET = 4  
    g\_conn = 0  
    workqueue = queue.Queue(100)  
  
  
    def \_\_init\_\_(self,sql\_init):  
        if isinstance(sql\_init,list) and (len(sql\_init) >= 4):  
  
  
            try:  
                conn = mysql\_module.connect(user=sql\_init\[self.USER\],passwd=sql\_init\[self.PASSWD\],host=sql\_init\[self.HOST\],db=sql\_init\[self.DB\],charset = 'utf8')  
                print(type(conn))  
                self.g\_conn = conn  
  
  
             \#   print(type(dd))  
            except:  
                print('connect err')  
                self.dcur = 'connect err'  
            \#self.dcur = conn.cursor(mysql\_module.cursors.DictCursor) \#返回字典cursor（K-V存储结构，python中最重要的类型之一）)  
            self.dcur = conn.cursor()  
        else:  
            dcur =  'not a dict'  
    def \_query(self,sqls):  
        try:  
            self.dcur.execute(sqls)  
            data = self.dcur.fetchone()  
            num = self.dcur.rowcount  
            return (data,num)  
        except:  
            return False  
  
  
    def \_insert(self,sqls):  
        try:  
     \#       self.dcur.execute('set ')  
            self.dcur.execute(sqls)  
            self.g\_conn.commit()  
            return self.g\_conn.insert\_id()  
        except:  
            return False  
    def \_update(self,sqls):  
        try:  
            self.dcur.execute(sqls)  
            self.g\_conn.commit()  
            return True  
        except:  
            return False  
  
  
    def \_fetchall(self):  
        return self.dcur.fetchall()  
  
  
    def \_fetchone(self):  
        \#返回一行结果，知道返回为NONE  
        return self.dcur.fetchone()  
  
  
    def \_getrowcount(self):  
        \#返回结果的条数  
        return self.dcur.rowcount  
    def \_rollback(self):  
        return self.g\_conn.rollback()  
    def \_\_del\_\_(self):  
        try:  
            self.dcur.close()  
            self.g\_conn.close()  
        except:  
            pass  
  
  
if \_\_name\_\_ == '\_\_main\_\_':  
    c = \['root','','localhost','test'\]  
    p = sql\_operate(c)  
    sqls = "select \* from mes\_1 where name = '%s'"%('song')  
    print(sqls)  
    data =  p.\_query(sqls)  
    print(p.\_fetchone())  
    print(p.\_fetchone())  
    del p  
\#data = p.query('update mes\_1 set name = \\'hao\\' where name = \\'dwa\\'')  
  
  
\#if data\[1\] != 0:  
\#    print(data\[0\])