""" Test connection to MySQL using mysql-client conn = MySQLdb.connect(host,port,user,passwd,db,charset) cursor = conn.cursor() cursor.execute(sql, (arg1,arg2,arg3...)) conn.commit() conn.rollback() cursor.description # （col1,...），（col2,....），（col3,...），（col4,...） cursor.rowcount columns = map(lambda e:e[0], cursor.description) columns = tuple(columns) # (col1,col2,col3...) cur_record = cursor.fetchone() # (key1,key2,key3...) cur_result = dict(zip(columns,cur_record)) # {col1:key1, col2:key2, col3:key3....} cursor.close() # no more operation is valid if cursor closed conn.close() # close connection to MySQL database """ import MySQLdb import datetime class MySQLHandler(): def __init__(self): self.conn = None self.cursor = None self.cur_record = () self.host = '' self.port = 0 self.user = '' self.passwd = '' self.db = '' self.charset = 'utf8' def connect(self, host, port, user, passwd, db, charset): self.host = host self.port = port self.user = user self.passwd = passwd self.db = db self.charset = charset try: self.conn = MySQLdb.connect( host = self.host, port = self.port, user = self.user, passwd = self.passwd, db = self.db, charset = self.charset ) self.cursor = self.conn.cursor() except MySQLdb.Error as e: print("Conn MySQL Error, %d: %s" %(e.args[0], e.args[1]) ) def execute_one(self, sql, args): self.cursor.execute(sql, args) def insert_one(self,sql,args): try: # 如果多条插入语句都没异常，则直接通过commit()将数据写入db # 如果任意一条语句产生异常，则进入except处理（还没运行try中的commit) # except中可以全部不提交，或者提交没触发异常的sql语句 # 全部不提交, 则except中写入：self.conn.commit() # 部分提交（没触发异常的语句结果，则except中写入：self.conn.commit() self.cursor.execute(sql,args) # write changes to db, if no exception self.conn.commit() except : # drop data, if any exception happens, no data saved to db self.conn.rollback() def delete_one(self,sql,args): try: # 如果多条delete语句都没异常，则直接通过commit()将数据写入db # 如果任意一条语句产生异常，则进入except处理（还没运行try中的commit) # except中可以全部不提交，或者提交没触发异常的sql语句 # 全部不提交, 则except中写入：self.conn.commit() # 部分提交（没触发异常的语句结果，则except中写入：self.conn.commit() self.cursor.execute(sql,args) # write changes to db, if no exception self.conn.commit() except : # drop data, if any exception happens, no data saved to db self.conn.rollback() def show(self): # show row count print("Row Count: " + str(self.cursor.rowcount)) # get table column name columns = map(lambda e: e[0], self.cursor.description) columns = tuple(columns) # show all results self.cur_record = self.cursor.fetchone() while self.cur_record != None: print( dict(zip(columns, self.cur_record)) ) self.cur_record = self.cursor.fetchone() def close(self): # close connection to mysql self.cursor.close() self.conn.close() # reset attributes self.conn = None self.cursor = None self.cur_record = () self.host = '' self.port = 0 self.user = '' self.passwd = '' self.db = '' self.charset = 'utf8' # Test MySQLHandler, create connection mysql_handler = MySQLHandler() mysql_handler.connect('localhost', 3306, 'root', 'weilan0415', 'news', 'utf8') # prepare sql to execute # sql = 'select * from employees where salary > %s limit %s, %s' # args = (2500, 0, 10) # run 1 sql operation # mysql_handler.execute_one(sql, args) # insert one record into mysql sql = 'insert into news (title, image, content, types, create_at, is_valid) ' 'value (%s, %s, %s, %s, %s, %s);' cur_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") mysql_handler.insert_one(sql, ("new222", "D:\Durian.jpg", "新闻内容222", "美食", cur_time, 1) ) # delete sql = 'delete from news where id between 6 and 10' mysql_handler.delete_one(sql, ()) # query sql = 'select * from news' mysql_handler.execute_one(sql, ()) # show result mysql_handler.show() # close connection mysql_handler.close()