python + docker, 实现天气数据 从FTP获取以及持久化-- python操作MySQL数据库

Posted 2020-11-19 TonyZhang24

前言

在这一节中，我们主要介绍如何使用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 = (
        (\'201804280000\', \'0.0    0.0    0.0    16.461    95.163    6.038    97.493    1013.791\'),
        (\'201804280015\', \'0.0    0.0    0.0    16.347    95.532    6.046    97.606    1013.713\'),
        (\'201804280030\', \'0.0    0.0    0.0    16.233    95.901    6.055    97.719    1013.634\'),
        (\'201804280045\', \'0.0    0.0    0.0    16.139    96.269    6.063    97.832    1013.556\'),
    )
    nRet1 = insert_or_update_weather_data(datas1)

    datas2 = (

        (\'201804280030\', \'{data1:0.0, data2:0.0, data3:0.0, data4:16.233, data5:95.901, data6:6.055, data7:97.719, data8:1013.030}\'),
        (\'201804280045\', \'{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(\'201804270000\', \'201804280040\')
    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部分的说明。

 

感谢大家的收看，欢迎积极留言讨论~~