安装

shell

pip install pymysql

使用

建立数据库，用Navicat进行测试 ，确认能访问，然后用pymysql 执行相同操作

代码

from mysql_cli import *
class MysqlConnector():
    def __init__(self, config):
        self.config = config
        self.database = get_mysql_db(self.config)

    def create_table(self, sql):
        cur = self.database.cursor()
        cur.execute(sql)

    def insert(self,table,data):
        field_list = []
        value_list = []
        for field,value in data.items():
            field_list.append(field)
            value_list.append(value)
        fields = ','.join(field_list)
        values  = ','.join([u if not isinstance(u,str) else """'{}'""".format(u) for u in value_list])
        sql = f'INSERT INTO {table} ({fields}) values ({values})'
        cur =self.database.cursor()
        print('insert',sql)
        cur.execute(sql)
        self.database.commit()

    def update(self, table, change, condition):
        update_record(self.database, table, change, condition)

    def delelte(self, table, condition):
        delete_record(self.database, table, condition)

    def check_record(self, tb):
        check_record(self.database, tb)

    def _find(self,table,query):
        cur = self.database.cursor()
        sql = ''' select * from {}  WHERE  {} '''.format(table,get_condition_sql(query))
        print('find -- >', sql)
        cur.execute(sql)
        return cur

    def find(self,table,query):
        cur = self._find(table,query)
        return cur.fetchall()

    def findone(self,table,query):
        cur = self._find(table,query)
        return cur.fetchone()

config = {

        'host': '192.168.1.90',
        'password': 'root',
        'user': 'root',
        'database': 'tableau'

    }

tb_name = 'new_table5'
connector = MysqlConnector(config=config)
res  = connector.findone(tb_name,{'id':1})
connector.insert(tb_name,{'name':'inserted_element2'})
connector.database.close()
print(res)

mongodb

mongo 把处理和查询的数据放在 dict 中， 我这里也仿照这样的模式进行封装