前言
在这一节中，我们主要介绍如何使用python操作MySQL数据库。
准备
MySQL数据库使用的是上一节中的docker容器 “test-mysql”.
Python 操作 MySQL
我们使用的IDE是 “神奇” 的 pycharm：
1. 首先新建一个python的项目，并且安装 “mysql-connector-python”。
“mysql-connector-python” 是MySQL官方对于python的数据驱动，感兴趣的童鞋可以移步这里： https://dev.mysql.com/doc/connector-python/en/
2. 创建 MySQLUtil.py
基于上一节的介绍，我们的任务是要读取FTP上的文本文件，从中解析出天气数据，并且插入到数据库表中（持久化）。因此，我们将基本的数据库操作都封装到这个python 文件中。
3. 准备工作 -- 在第一次连接的时候需要DDL操作（创建相应的数据库和表）； 之后的的操作是DML（增/删/改/查）
# -*- coding: utf-8 -*- from __future__ import print_function import mysql.connector
from mysql.connector import errorcode config_prepare = {
'user': 'root',
'password': 'password',
'host': '127.0.0.1',
#'database': 'weather_db',
'raise_on_warnings': True,
} config_op = {
'user': 'root',
'password': 'password',
'host': '127.0.0.1',
'database': 'weather_db',
#'raise_on_warnings': True,
} def preparation(db_name='weather_db', table_name='weather'):
try:
cnx = mysql.connector.connect(**config)
except mysql.connector.Error as err:
print(err)
return (-1) # try use db or create it
try:
cursor = cnx.cursor()
cursor.execute("CREATE DATABASE IF NOT EXISTS {} DEFAULT CHARACTER SET 'UTF8MB4'".format(db_name))
cnx.database = db_name
except mysql.connector.Error as err:
print(err)
#return (-2) # Create table if not exist
try:
cnx.database = db_name
cursor = cnx.cursor()
createTableSql = """CREATE TABLE IF NOT EXISTS `weather` ( \
`weather_timestamp` varchar(32) NOT NULL, \
`weather_data` varchar(256) NOT NULL, \
PRIMARY KEY (`weather_timestamp`) \
) ENGINE=InnoDB;""" cursor.execute(createTableSql)
cursor.close()
except mysql.connector.errorcode as err:
print(err.msg)
return (-3) print ("Preparation OK") return (0)
4. 插入或更新数据
def insert_or_update_weather_data(datas):
sql = "INSERT INTO weather(weather_timestamp, weather_data) " \
"VALUES(%s,%s)" \
"ON DUPLICATE KEY UPDATE weather_data = VALUES(weather_data);" try:
cnx = mysql.connector.connect(**config_op) cursor = cnx.cursor()
cursor.executemany(sql, datas) cnx.commit() except mysql.connector.errorcode as e:
print('Error:', e)
return (-1) finally:
cursor.close()
cnx.close() return (0)
PS: 由于天气数据是由供应商提供，提供的方式是 每日两次（早上8点；下午5点）放到FTP上面，每次提供7天的天气预报。 因此，当我们读取天气数据的时候，有一段时间的数据是重合的。因此采用插入或更新 （表中没有天气数据时：插入； 有数据时：更新）的方式来处理。
5. 根据时间条件，读取天气数据
def query_weather_data(begin_time_stamp, end_time_stamp):
sql = "SELECT weather_timestamp, weather_data FROM weather " \
"WHERE weather_timestamp BETWEEN %s AND %s;"
try:
cnx = mysql.connector.connect(**config_op)
cursor = cnx.cursor()
# The second parameter must be a tuple that contains all the delimeters
cursor.execute(sql, (begin_time_stamp, end_time_stamp))
rows = cursor.fetchall()
return rows
except mysql.connector.errorcode as e:
print('Error:', e)
return []
finally:
cursor.close()
cnx.close()
6. 最后附上测试代码
if __name__ == '__main__':
print("=== Unit test begin ===") # Test prepareation
nRet = preparation() # Test insert or update datas in 'weather' table
datas1 = (
('', '0.0 0.0 0.0 16.461 95.163 6.038 97.493 1013.791'),
('', '0.0 0.0 0.0 16.347 95.532 6.046 97.606 1013.713'),
('', '0.0 0.0 0.0 16.233 95.901 6.055 97.719 1013.634'),
('', '0.0 0.0 0.0 16.139 96.269 6.063 97.832 1013.556'),
)
nRet1 = insert_or_update_weather_data(datas1) datas2 = ( ('', '{data1:0.0, data2:0.0, data3:0.0, data4:16.233, data5:95.901, data6:6.055, data7:97.719, data8:1013.030}'),
('', '{data1:0.0, data2:0.0, data3:0.0, data4:11.111, data5:93.901, data6:6.099, data7:97.700, data8:1013.045}'),
)
nRet2 = insert_or_update_weather_data(datas2) # Test query data from 'weather' table
ret_list = query_weather_data('', '')
if len(ret_list) == 0:
print ("No data queried out.")
else:
for timestamp, data in ret_list:
print ('timestamp:%s data:%s' %(timestamp, data)) print("=== Unit test done ===")
小结
至此，我们操作MySQL的工具代码就全部编写完成。 在下一节中，我们将着重FTP部分的说明。
感谢大家的收看，欢迎积极留言讨论~~