使用方法：

在Vscode中，会有代码提示，很容易看懂

  1. import mysqldb
  3. db = mysqldb.connect_db(mysqldb.conn_confing)
  5. # 执行sql，会有提示
  6. db.query("update table1 set col='' ")
  8. # 获取多个结果，返回 dict
  9. rows = db.fetch_rows(table='table')

源代码 mysqldb.py

  1. from pymysql import (connect, cursors, err, escape_sequence)
  3. conn_confing = {
  4.     'host': 'localhost',
  5.     'port': '3306',
  6.     'user': 'root',
  7.     'password': 'root',
  8.     'db': 'your_db',
  9.     'charset': 'utf-8'
  10. }
  13. def connect_db(conn_confing):
  14.     # msyql dababase connection info
  15.     dbconn = MYSQL(
  16.         dbhost=conn_confing.get('host'),
  17.         dbport=conn_confing.get('port'),
  18.         dbuser=conn_confing.get('user'),
  19.         dbpwd=conn_confing.get('password'),
  20.         dbname=conn_confing.get('db'),
  21.         dbcharset=conn_confing.get('charset')
  22.     )
  23.     return dbconn
  26. def connect_ssdc(conn_confing):
  27.     """Connect to the database return SSDictCursor dbsession"""
  28.     connection = connect(
  29.         host=conn_confing.get('host'),
  30.         port=int(conn_confing.get('port')) or 3306,
  31.         user=conn_confing.get('user'),
  32.         password=conn_confing.get('password'),
  33.         db=conn_confing.get('db'),
  34.         charset=conn_confing.get('charset'),
  35.         cursorclass=cursors.SSDictCursor)
  36.     return connection
  39. class MYSQL:
  40.     """A Friendly pymysql Class, Provide CRUD functionality"""
  42.     def __init__(self, dbhost, dbuser, dbpwd, dbname, dbcharset='utf-8', dbport=3306,):
  43.         self.dbhost = dbhost
  44.         self.dbport = int(dbport)
  45.         self.dbuser = dbuser
  46.         self.dbpwd = dbpwd
  47.         self.dbname = dbname
  48.         self.dbcharset = dbcharset
  49.         self.connection = self.session()
  51.     def session(self):
  52.         """Connect to the database return dbsession"""
  53.         connection = connect(
  54.             host=self.dbhost,
  55.             port=self.dbport,
  56.             user=self.dbuser,
  57.             password=self.dbpwd,
  58.             db=self.dbname,
  59.             # charset=self.dbcharset,
  60.             cursorclass=cursors.DictCursor)
  61.         return connection
  63.     def insert(self, table, data):
  64.         """mysql insert() function"""
  66.         with self.connection.cursor() as cursor:
  68.             params = self.join_field_value(data)
  70.             sql = "INSERT IGNORE INTO {table} SET {params}".format(
  71.                 table=table, params=params)
  72.             # print(sql)
  73.             try:
  74.                 cursor.execute(sql, tuple(data.values()))
  75.                 last_id = cursor.lastrowid
  77.                 self.connection.commit()
  78.             except Exception as err:
  79.                 print(err)                
  81.             return last_id
  83.     def bulk_insert(self, table, data):
  85.         assert isinstance(data, list) and data != [], "data_format_error"
  87.         with self.connection.cursor() as cursor:
  89.             params = []
  90.             for param in data:
  91.                 params.append(escape_sequence(param.values(), 'utf-8'))
  93.             values = ', '.join(params)
  94.             fields = ', '.join('`{}`'.format(x) for x in param.keys())
  96.             sql = u"INSERT IGNORE INTO {table} ({fields}) VALUES {values}".format(
  97.                 fields=fields, table=table, values=values)
  99.             cursor.execute(sql)
  100.             last_id = cursor.lastrowid
  102.             self.connection.commit()
  103.             return last_id
  105.     def delete(self, table, condition=None, limit=None):
  106.         """
  107.         mysql delete() function
  108.         sql.PreparedStatement method
  109.         """
  110.         with self.connection.cursor() as cursor:
  112.             prepared = []
  114.             if not condition:
  115.                 where = '1'
  116.             elif isinstance(condition, dict):
  117.                 where = self.join_field_value(condition, ' AND ')
  118.                 prepared.extend(condition.values())
  119.             else:
  120.                 where = condition
  122.             limits = "LIMIT {limit}".format(limit=limit) if limit else ""
  124.             sql = "DELETE FROM {table} WHERE {where} {limits}".format(
  125.                 table=table, where=where, limits=limits)
  127.             if not prepared:
  128.                 result = cursor.execute(sql)
  129.             else:
  130.                 result = cursor.execute(sql, tuple(prepared))
  132.             self.connection.commit()
  133.             return result
  135.     def update(self, table, data, condition=None):
  136.         """
  137.         mysql update() function
  138.         Use sql.PreparedStatement method
  139.         """
  140.         with self.connection.cursor() as cursor:
  142.             prepared = []
  143.             params = self.join_field_value(data)
  144.             prepared.extend(data.values())
  146.             if not condition:
  147.                 where = '1'
  148.             elif isinstance(condition, dict):
  149.                 where = self.join_field_value(condition, ' AND ')
  150.                 prepared.extend(condition.values())
  151.             else:
  152.                 where = condition
  154.             sql = "UPDATE IGNORE {table} SET {params} WHERE {where}".format(
  155.                 table=table, params=params, where=where)
  157.             # check PreparedStatement
  158.             if not prepared:
  159.                 result = cursor.execute(sql)
  160.             else:
  161.                 result = cursor.execute(sql, tuple(prepared))
  163.             self.connection.commit()
  164.             return result
  166.     def count(self, table, condition=None):
  167.         """
  168.         count database record
  169.         Use sql.PreparedStatement method
  170.         """
  171.         with self.connection.cursor() as cursor:
  173.             prepared = []
  175.             if not condition:
  176.                 where = '1'
  177.             elif isinstance(condition, dict):
  178.                 where = self.join_field_value(condition, ' AND ')
  179.                 prepared.extend(condition.values())
  180.             else:
  181.                 where = condition
  183.             sql = "SELECT COUNT(*) as cnt FROM {table} WHERE {where}".format(
  184.                 table=table, where=where)
  186.             if not prepared:
  187.                 cursor.execute(sql)
  188.             else:
  189.                 cursor.execute(sql, tuple(prepared))
  191.             self.connection.commit()
  192.             return cursor.fetchone().get('cnt')
  194.     def fetch_rows(self, table, fields=None, condition=None, order=None, limit=None, fetchone=False):
  195.         """
  196.         mysql select() function
  197.         Use sql.PreparedStatement method
  198.         """
  199.         with self.connection.cursor() as cursor:
  201.             prepared = []
  203.             if not fields:
  204.                 fields = '*'
  205.             elif isinstance(fields, tuple) or isinstance(fields, list):
  206.                 fields = '`{0}`'.format('`, `'.join(fields))
  207.             else:
  208.                 fields = fields
  210.             if not condition:
  211.                 where = '1'
  212.             elif isinstance(condition, dict):
  213.                 where = self.join_field_value(condition, ' AND ')
  214.                 prepared.extend(condition.values())
  215.             else:
  216.                 where = condition
  218.             if not order:
  219.                 orderby = ''
  220.             else:
  221.                 orderby = 'ORDER BY {order}'.format(order=order)
  223.             limits = "LIMIT {limit}".format(limit=limit) if limit else ""
  225.             sql = "SELECT {fields} FROM {table} WHERE {where} {orderby} {limits}".format(
  226.                 fields=fields, table=table, where=where, orderby=orderby, limits=limits)
  228.             if not prepared:
  229.                 cursor.execute(sql)
  230.             else:
  231.                 cursor.execute(sql, tuple(prepared))
  233.             self.connection.commit()
  234.             return cursor.fetchone() if fetchone else cursor.fetchall()
  236.     def query(self, sql, fetchone=False, execute=False):
  237.         """execute custom sql query"""
  238.         with self.connection.cursor() as cursor:
  240.             cursor.execute(sql)
  241.             self.connection.commit()
  243.             if execute:
  244.                 return
  246.             return cursor.fetchone() if fetchone else cursor.fetchall()
  248.     def join_field_value(self, data, glue=', '):
  249.         sql = comma = ''
  250.         for key in data.keys():
  251.             sql += "{}`{}` = %s".format(comma, key)
  252.             comma = glue
  253.         return sql
  255.     def close(self):
  256.         if getattr(self, 'connection', 0):
  257.             return self.connection.close()
  259.     def __del__(self):
  260.         """close mysql database connection"""
  261.         self.close()