Python采用并发查询mysql以及调用API灌数据（一）

Devops海洋的渔夫

发布于 2019-05-31 16:25:43

1.7K0

发布于 2019-05-31 16:25:43

文章被收录于专栏：Devops专栏Devops专栏

实战任务

本次因为服务架构重构，表优化、重构，带来的任务就是需要从原来的mysql数据库中，读取原表数据（部分存在多张关联查询）然后通过调用API的服务方式灌入新的数据库表中（包含mysql、mongodb）。

执行流程如下

那么根据流程所需要的功能，需要以下的实例进行支撑：

1.并发实例

2.查询数据实例

3.执行post请求实例

目标：本章节先实现 2、3两点的实例。

查询数据库实例（PyMySQL连接mysql数据库）

PyMySQL是一个纯Python写的MySQL客户端，它的目标是替代MySQLdb，可以在CPython、PyPy、IronPython和Jython环境下运行。PyMySQL在MIT许可下发布。

PyMySQL的性能和MySQLdb几乎相当，如果对性能要求

不是特别的强，使用PyMySQL将更加方便。

PyMySQL的使用方法和MySQLdb几乎一样。

安装PyMysql

pip install pymysql

编写PyMysql连接mysql相关操作的实例

# -*- coding: utf-8 -*-
import pymysql

config = {
    'host': '你的mysql服务的IP地址',
    'port': 3306,
    'user': 'root',
    'passwd': '你的root密码',
    'charset':'utf8mb4',
    'cursorclass':pymysql.cursors.DictCursor
    }
conn = pymysql.connect(**config)
conn.autocommit(1)
cursor = conn.cursor()

try:
    # 创建数据库
    DB_NAME = 'test_db'
    cursor.execute('DROP DATABASE IF EXISTS %s' %DB_NAME)
    cursor.execute('CREATE DATABASE IF NOT EXISTS %s' %DB_NAME)
    conn.select_db(DB_NAME)

    #创建表
    TABLE_NAME = 'test_user'
    cursor.execute('CREATE TABLE %s(id int primary key,name varchar(30))' %TABLE_NAME)

    # 批量插入纪录
    values = []
    for i in range(20):
        values.append((i,'testuser'+str(i)))
    cursor.executemany('INSERT INTO '+ TABLE_NAME +' values(%s,%s)' ,values)

    # 查询数据条目
    count = cursor.execute('SELECT * FROM %s' %TABLE_NAME)
    print 'total records:', cursor.rowcount

    # 获取表名信息
    desc = cursor.description
    print "%s %3s" % (desc[0][0], desc[1][0])

    cursor.scroll(10,mode='absolute')
    results = cursor.fetchall()
    for result in results:
        print result

except:
    import traceback
    traceback.print_exc()
    # 发生错误时会滚
    conn.rollback()
finally:
    # 关闭游标连接
    cursor.close()
    # 关闭数据库连接
    conn.close()

执行结果:

E:\PycharmProjects\DataProject>python MysqlTest01.py
total records: 20
id name
{u'id': 10, u'name': u'testuser10'}
{u'id': 11, u'name': u'testuser11'}
{u'id': 12, u'name': u'testuser12'}
{u'id': 13, u'name': u'testuser13'}
{u'id': 14, u'name': u'testuser14'}
{u'id': 15, u'name': u'testuser15'}
{u'id': 16, u'name': u'testuser16'}
{u'id': 17, u'name': u'testuser17'}
{u'id': 18, u'name': u'testuser18'}
{u'id': 19, u'name': u'testuser19'}

E:\PycharmProjects\DataProject>

从上面的步骤中，已经可以创建数据库、创建表、插入数据以及查询数据。但是在处理插入中文的时候是存在问题的。

在创建数据库的时候，需要设置好utf的格式。

cursor.execute('CREATE DATABASE IF NOT EXISTS %s DEFAULT CHARACTER SET utf8 COLLATE utf8_general_ci' %DB_NAME )

编写插入以及查询中文数据的实例

# -*- coding: utf-8 -*-
import pymysql

config = {
    'host': '你的mysql服务的IP地址',
    'port': 3306,
    'user': 'root',
    'passwd': '你的mysql服务的root密码',
    'charset':'utf8',
    'cursorclass':pymysql.cursors.DictCursor
    }
conn = pymysql.connect(**config)
conn.autocommit(1)
cursor = conn.cursor()

try:
    # 创建数据库
    DB_NAME = 'test_db'
    cursor.execute('DROP DATABASE IF EXISTS %s' %DB_NAME)
    cursor.execute('CREATE DATABASE IF NOT EXISTS %s DEFAULT CHARACTER SET utf8 COLLATE utf8_general_ci' %DB_NAME )
    conn.select_db(DB_NAME)

    #创建表
    TABLE_NAME = 'test_user'
    cursor.execute('CREATE TABLE %s(id int primary key,name varchar(30))' %TABLE_NAME)

    # 批量插入纪录
    values = []
    for i in range(20):
        # values.append((i,"testuser"+str(i)))
        values.append((i,u"测试用户"+str(i))) # 插入中文数据
    cursor.executemany('INSERT INTO '+ TABLE_NAME +' values(%s,%s)' ,values)

    # 查询数据条目
    count = cursor.execute('SELECT * FROM %s' %TABLE_NAME)
    print 'total records:', cursor.rowcount

    # 获取表名信息
    # desc = cursor.description
    # print "%s %3s" % (desc[0][0], desc[1][0])

    cursor.scroll(10,mode='absolute')
    results = cursor.fetchall()
    for row in results:
        # user_id = row[0]
        print row
        print(row['id'])      # 查询打印用户id
        print(row['name'])    # 查询打印用户名


except:
    import traceback
    traceback.print_exc()
    # 发生错误时会滚
    conn.rollback()
finally:
    # 关闭游标连接
    cursor.close()
    # 关闭数据库连接
    conn.close()

执行如下：

E:\PycharmProjects\DataProject>python MysqlTest01.py
total records: 20
{u'id': 10, u'name': u'\u6d4b\u8bd5\u7528\u623710'}
10
测试用户10
{u'id': 11, u'name': u'\u6d4b\u8bd5\u7528\u623711'}
11
测试用户11
{u'id': 12, u'name': u'\u6d4b\u8bd5\u7528\u623712'}
12
测试用户12
{u'id': 13, u'name': u'\u6d4b\u8bd5\u7528\u623713'}
13
测试用户13

在本次实战中，是不需要做创建库、表、插入数据的操作的，只需要查询对应的数据，然后调用API进行post请求即可，那么下面单独写一个查询的实例。

编写PyMysql的查询实例 - 单独查询用户id以及用户名

# -*- coding: utf-8 -*-
import pymysql

config = {
    'host': '你的mysql服务的IP',
    'port': 3306,
    'user': 'root',
    'passwd': '你的mysql的root密码',
    'charset':'utf8',
    'cursorclass':pymysql.cursors.DictCursor
    }
conn = pymysql.connect(**config)
conn.autocommit(1)
cursor = conn.cursor()

try:
    # 设置使用数据库
    DB_NAME = 'test_db'
    conn.select_db(DB_NAME)

    #定义查询的表
    TABLE_NAME = 'test_user'

    # 查询数据条目
    count = cursor.execute('SELECT * FROM %s' %TABLE_NAME)
    print 'total records:', cursor.rowcount

    # 获取表名信息
    # desc = cursor.description
    # print "%s %3s" % (desc[0][0], desc[1][0])

    cursor.scroll(10,mode='absolute')
    results = cursor.fetchall()
    for row in results:
        # print row
        print("user_id=%s ,user_name=%s " % (row['id'],row['name'])) # 查询用户id\用户名


except:
    import traceback
    traceback.print_exc()
    # 发生错误时会滚
    conn.rollback()
finally:
    # 关闭游标连接
    cursor.close()
    # 关闭数据库连接
    conn.close()

执行如下：

E:\PycharmProjects\DataProject>python MysqlTest02.py
total records: 20
user_id=10 ,user_name=测试用户10
user_id=11 ,user_name=测试用户11
user_id=12 ,user_name=测试用户12
user_id=13 ,user_name=测试用户13
user_id=14 ,user_name=测试用户14
user_id=15 ,user_name=测试用户15
user_id=16 ,user_name=测试用户16
user_id=17 ,user_name=测试用户17
user_id=18 ,user_name=测试用户18
user_id=19 ,user_name=测试用户19

E:\PycharmProjects\DataProject>

写到这里基本上已经满足了查询mysql数据的需求了，那么下一步编写执行POST请求的代码实例。

编写执行POST请求API的代码实例

注意：Python中的请求体需要填写null的时候用None替换。

首先使用POSTMAN请求确认接口是否正常：

按照该上图所示，接口的API服务是能够正常请求的，那么下来用python代码请求一下这个Post请求：

# coding:utf-8
import urllib2,json
url='修改你这边提供服务的API'
data=json.dumps({"userId":"21111111","province":None})                #请求体数据, 注意：python中没有null，需要用None来替换
headers={
     # 修改你这边Header的参数
    'Content-Type':'application/json'
}     #头部
request = urllib2.Request(url,data,headers)     #urllib2用一个Request对象来映射你提出的HTTP请求
print request
response = urllib2.urlopen(request)             #通过调用urlopen并传入Request对象，将返回一个相关请求response对象
print response
message = response.read()                       #这个应答对象如同一个文件对象，所以你可以在Response中调用.read()
print message

执行返回结果如下：

E:\PycharmProjects\DataProject>python PostTest02.py
<urllib2.Request instance at 0x0000000005DED548>
<addinfourl at 98539208L whose fp = <socket._fileobject object at 0x0000000005DB6318>>
{"resultCode":0,"message":null,"data":{"id":"1066914784882040832","userId":21111111,"userCode":null,"province":null,"city":null,"county":null,"town":null,"village":null,....},"accessory":null}

E:\PycharmProjects\DataProject>

从返回结果 "resultCode":0 可以知道，执行的请求成功了。如果是其他返回结果参数，例如"resultCode":500那么就要处理请求失败的情况。

在上面的内容中已经有了基本执行的示例，但是如果需要频繁操作，肯定需要抽象相关的方法类，下一篇章Python采用并发查询mysql以及调用API灌数据（二） - PyMysql操作数据库基本类封装

本文参与腾讯云自媒体分享计划，分享自作者个人站点/博客。

原始发表：2018.11.26 ，如有侵权请联系 cloudcommunity@tencent.com 删除

云数据库 SQL Server