安装

Python3：

pip3 install pymysql

Conda：

conda install pymysql

使用

导入模块

import pymysql

建立连接

def connect(host, port, user, password, charset):print("host = ", host)print("port = ", port)print("user = ", user)print("password = ", password)print("charset = ", charset)con = pymysql.connect(host = host, port = port,user = user, password = password, charset = charset)cursor = con.cursor()return con, cursor

关闭连接

def close(con, cursor):if cursor: cursor.close()if con : con.close()

创建数据库

def create_database(con, database):if (con or database) is None: returncursor = con.cursor()try:sql = "CREATE DATABASE IF NOT EXISTS {}".format(database)cursor.execute(sql)returnexcept pymysql.Error as err:print(err)

创建数据表

def create_table(con, database, table):if (con or database or table) is None: returncursor = con.cursor()    try:sql = "use " + databasecursor.execute(sql)# OHLC: Open, High, Low, Closesql = ''' CREATE TABLE IF NOT EXISTS {}(DateTime datetime, Open float, High float, Low float, Close float, TradeVolume BIGINT,HoldVOlume BIGINT, Average float, PRIMARY (DateTime))ENGINE=innodb DEFAULT CHARSET=UTF8MB4;'''.format(table)print(sql)cursor.execute(sql)except pymysql.Error as err:print(err)

删除数据库

def drop_database(con, database):if (con or database) is None: returncursor = con.cursor()sql = "DROP DATABASE IF EXISTS " + databasecursor.execute(sql)

删除数据表

def drop_database(con, database, table):if (con or database or table) is None: returncursor = con.cursor()sql = "USE " + databasecursor.execute(sql)    sql = "DROP TABLE IF EXISTS " + tablecursor.execute(sql)

完整代码

#!/usr/local/bin/python3# @Time    : 2021/1/22 14:00
# @Author  : Paul Ding
# @File    : mysql.py
# @Software: PyCharmimport pymysqlclass mysql:def __init__(self, **kwargs):self.host = "localhost"self.port = 3306self.user = ""self.password = ""self.charset = "UTF8MB4"self.database = ""self.con = Noneself.cursor = Nonefor key in kwargs:if key == 'host': self.host = kwargs[key]elif key == 'port': self.port = kwargs[key]elif key == 'user': self.user = kwargs[key]elif key == 'password': self.password = kwargs[key]elif key == 'charset': self.charset = kwargs[key]elif key == 'database': self.database = kwargs[key]else: print("Not used arguments: {} = {}".format(key, kwargs[key]))self.connect()def connect(self):host = self.hostport = self.portuser = self.userpassword = self.passwordcharset = self.charsetprint("host = ", host)print("port = ", port)print("user = ", user)print("password = ", password)print("charset = ", charset)con = pymysql.connect(host = host, port = port,user = user, password = password, charset = charset)cursor = con.cursor()self.con = conself.cursor = cursordef close(self):con = self.conif con : con.close()def create_database(self, database):con = self.concursor = self.cursorif (con or cursor or database) is None: returntry:sql = "CREATE DATABASE IF NOT EXISTS {}".format(database)cursor.execute(sql)returnexcept pymysql.Error as err:print(err)def create_table(self, database, table, table_columns, primary_key):'''CREATE TABLE IF NOT EXISTS gold2102 (DateTime datetime,Open float(10,4),High float(10,4),Low float(10,4),Close float(10,4),TradeVolume BIGINT,HoldVolume BIGINT,Average float(10,4),PRIMARY KEY(DateTime)) ENGINE=innodb DEFAULT CHARSET=UTF8MB4;'''con = self.concursor = self.cursorif (con or cursor or database or table) is None: returntry:sql_statement = "use " + databasecursor.execute(sql_statement)# OHLC: Open, High, Low, Closesql_statement = "CREATE TABLE IF NOT EXISTS {} (\n".format(table)for key, value in table_columns.items():sql_statement += "    {} {}, \n".format(key, value)sql_statement += "    PRIMARY KEY ({})) ENGINE=innodb DEFAULT CHARSET=UTF8MB4;".format(primary_key)print(sql_statement)cursor.execute(sql_statement)except pymysql.Error as err:print(err)def drop_database(self, database):con = self.concursor = self.cursorif (con or cursor or database) is None: returnsql = "DROP DATABASE IF EXISTS " + databasecursor.execute(sql)def drop_table(self, database, table):con = self.concursor = self.cursorif (con or cursor or database or table) is None: returnsql = "use " + databasecursor.execute(sql)sql = "DROP TABLE IF EXISTS " + tablecursor.execute(sql)def insert_item(self, database, table, values):con = self.concursor = self.cursorif (con or cursor or database or table or values) is None: returnsql = "use " + databasecursor.execute(sql)sql = """INSERT INTO {}VALUES{} """.format(table, values)try:print(sql)cursor.execute(sql)con.commit()except pymysql.Error as err:print(err)def insert_items(self, database, table, values_list):con = self.concursor = self.cursorif (con or cursor or database or table or values_list) is None: returnsql = "use " + databasecursor.execute(sql)batch_rows = 10000inserted_rows = 0rest_rows = len(values_list)while rest_rows > 0:rows = batch_rows if rest_rows > batch_rows else rest_rowsprint("rows", rows)print("{} items will be inserted into table {}".format(rows, table))sql = """INSERT INTO {} VALUES """.format(table)idx = 0for values in values_list[inserted_rows : inserted_rows+rows]:if idx == 0:comma = ''else:comma = ','sql += "{} {}".format(comma, values)idx += 1# print(sql)try:passcursor.execute(sql)con.commit()except pymysql.Error as err:print(err)inserted_rows += rowsrest_rows -= rowsdef show_tables(self, database):con = self.concursor = self.cursorif (con or cursor or database) is None: returnsql = "use " + databasecursor.execute(sql)sql = "show tables"cursor.execute(sql)tables = list(cursor.fetchall())print("%d table(s) found: %s" % (len(tables), tables))idx = 0for table in tables:print("Tables[%d] :" % idx, "%s" % table)idx += 1print()return tablesdef show_databases(self):con = self.concursor = self.cursorif (con or cursor) is None: returnsql = "show databases"cursor.execute(sql)databases = list(cursor.fetchall())print("%d databases found: %s" %(len(databases),databases))idx = 0for database in databases:print("Database[%d] :" % idx, "%s" % database)idx += 1print()return databasesdef date(self):con = self.concursor = self.cursorif (con or cursor) is None: returnsql = 'SET @dt = NOW()'cursor.execute(sql)sql = 'SELECT DATE(@dt)'cursor.execute(sql)return cursor.fetchone()def time(self):con = self.concursor = self.cursorif (con or cursor) is None: returnsql = 'SET @dt = NOW()'cursor.execute(sql)sql = 'SELECT TIME(@dt)'cursor.execute(sql)return cursor.fetchone()def version(self):con = self.concursor = self.cursorif (con or cursor) is None: returncursor.execute("SELECT VERSION()")version = cursor.fetchone()return versiondef admin(self, database, table):if (database or table) is None: returntable_columns = {'DateTime': 'datetime','Open': 'float(10,4)','High': 'float(10,4)','Low': 'float(10,4)','Close': 'float(10,4)','TradeVolume': 'BIGINT','HoldVolume': 'BIGINT','Average': 'float(10,4)'}primary_key = 'DateTime'print("database: {}".format(database))print("table: {}".format(table))db.drop_database(database)db.create_database(database)db.show_databases()db.create_table(database, table, table_columns, primary_key)db.show_tables(database)if __name__ == "__main__":database = "futures"table = "gold2102"values_list = [("2020/12/23 02:27", 391.8600, 391.8600, 391.7000, 391.7000, 59.0000, 50041.0000, 391.8169),("2020/12/23 02:28", 391.7000, 391.9200, 391.6800, 391.8400, 75.0000, 50000.0000, 391.8222)]db = mysql(host = "localhost", user = "root", password = "admin")db.admin(database, table)db.insert_items(database, table, values_list)db.close()

PyMySQL安装与使用相关推荐

PyMySQL安装问题解决办法-UnicodeDecodeError：gbk codec cannot decode byte 0xaf
PyMySQL安装问题解决办法-UnicodeDecodeError:gbk codec cannot decode byte 0xaf 依据报错信息,找到site.py 对应文件,分析问题所在行,可 ...
pymysql安装过程
1.打开cmd 2.输入conda list查看anaconde下有无pymysql库没有--> 3.直接输入pip install pymysql安装pymysql库 4.cmd下输入pyt ...
MySQL和pymysql安装 linux ubuntu
MySQL安装简单3步 1. sudo apt-get install mysql-server 2. sudo apt-get install mysql-client 3. sudo apt-g ...
linux pymysql 安装_Python进阶知识全篇-MySQL（PyMySQL）
什么是 PyMySQL? PyMySQL 是在 Python3.x 版本中用于连接 MySQL 服务器的一个库,Python2中则使用mysqldb. PyMySQL 遵循 Python 数据库 AP ...
ubuntu下pymysql安装及python操作
用pymysql代替MySQLdb 在我刚开始学python的时候,用的是python2.7,那时候连接mysql用的库是MySQLdb(很诡异的大小写,初学者经常因为记不住大小写导致"No ...
pymysql安装_openstack stein安装placement
openstack stein install placement 安装组件: placement 注:安装服务均遵循以下步骤: mariadb创库授权 keystone创建角色并关联 keyston ...
pymysql安装_第八章 nova组件安装2
1.在控制节点安装nova组件 apt -y install nova-api nova-placement-api nova-conductor nova-consoleauth nova-sche ...
pymysql安装_pymysql 模块简单使用
关视频可参考 Python3 从入门到精通视频教程千里马:Python3 从入门到精通视频教程zhuanlan.zhihu.com pymysql 模块简单使用安装 pymysql 模块 pip ...
PymySQL安装以及解决报错
Python3.7连接数据库(windows系统) 尝试用python连接数据库并进行访问,网上百度后需要使用mysqldb驱动,但该驱动针对于python之前的版本,python3以上的版本选用的是 ...

PyMySQL安装与使用

PyMySQL安装与使用

安装

使用

导入模块

建立连接

关闭连接

创建数据库

创建数据表

删除数据库

删除数据表

完整代码

PyMySQL安装与使用相关推荐

最新文章

热门文章