python常用第三方库(yaml、pymysql、urllib3)

一、常用第三方库 yaml

1、YAML概述

1)一种数据序列化格式
2)用于人类的可读性和与脚本语言的交互
3)一种被认为可以超越 XML、JSON 的配置文件

2、YAML 基本语法规则

1)大小写敏感
2)使用缩进表示层级关系
3)缩进时不允许使用 Tab 键,只允许使用空格
4)缩进的空格数目不重要,只要相同层级的元素左侧对齐即可
5)# 表示注释,从这个字符一直到行尾,都会被解析器忽略

3、YAML 支持的数据结构

1)对象:键值对的集合,用冒号 “:” 表示
2)数组:一组按次序排列的值,前加 “-”
3)纯量:单个的、不可再分的值
  * 字符串
  * 布尔值
  * 整数
  * 浮点数
  * Null
  * 时间
  * 日期
# 编程语言
languages:
  - PHP
  - Java
  - Python 
book:
  Python入门: # 书籍名称
    price: 55.5
    author: Lily
    available: True
    repertory: 20
    date: 2018-02-17
  Java入门:
    price: 60
    author: Lily
    available: False
    repertory: Null
    date: 2018-05-11

4、PyYAML

1)Python 的 YAML 解析器和生成器
2)官网:https://pypi.org/project/PyYAML/
3)安装:pip install pyyaml

5、创建 yaml 文件、读取 yaml 文件

import yaml

"""创建yaml"""
# # 定义python对象
# data = {
#     "client": {"default-character-set": "utf8"},
#     "mysql": {"user": 'root', "password": 123},
#     "custom": {
#         "user1": {"user": "张三", "pwd": 666},
#         "user2": {"user": "李四", "pwd": 999},
#     }
# }
#
# # 用dump方法把python对象转为yaml文档
# # 包含有中文时,allow_unicode需要改为True展示中文,./为在当前目录下创建文件,f为对应打开的文件
# with open('./my.yaml', 'w', encoding='utf-8') as f:
#     yaml.dump(data, f, allow_unicode=True)


"""读取yaml文件"""
# 读取yaml文件内容,转化为python对象
file_path = './my.yaml'

with open(file_path, 'r', encoding='utf-8') as f:
    data = yaml.safe_load(f)
print(data)
print(type(data))
# 读取user1里的第一个姓名
user1_name = data['custom']['user1']['user']
print(user1_name)

二、常用第三方库pymysql

1、pymysql 概述

1)Python 的数据库接口标准是 Python DB-API
2)PyMySQL 是从 Python 连接到 MySQL 数据库服务器的接口
3)PyMySQL 的目标是成为 MySQLdb 的替代品
4)官方文档:http://pymysql.readthedocs.io/

2、pymysql 安装

1)使用 pip 安装
2)使用 Pycharm 界面安装

3、pymysql 连接数据库

1)host:MySQL 服务器地址
2)user:用户名
3)password:密码
4)database:数据库名称
5)charset:编码方式,推荐使用 utf8mb4
# 1.导入库
import pymysql

# 2.建立连接
conn = pymysql.connect(host='服务器地址',
                     user='用户名',
                     password='密码',
                     database='数据库名',
                     charset="utf8mb4")
# 3.关闭连接
conn.close()

4、封装获取连接的函数

import pymysql

# 1.封装建立连接的对象
def get_conn():
    conn = pymysql.connect(
        host="服务器地址",
        user="root",
        password="123456",
        database="数据库名",
        charset="utf8mb4"
    )

    return conn

5、pymysql 实例

1)获取连接对象
    打开
    关闭
2)获取游标对象
    执行 SQL
    查询记录
from py_L3.pymysql.pymysql_conn import get_conn

def test_demo():
    # 1.获取连接对象
    conn = get_conn()
    # 2.获取游标对象
    cursor = conn.cursor()
    # 3.执行SQL
    cursor.execute("SELECT VERSION()")
    # 4.查询结果
    version = cursor.fetchone()
    print(f"数据库的版本是:{version}")
    # 5.关闭连接
    conn.close()

6、pymysql 数据库操作

1)CRUD 操作
    * 创建表
    * 插入记录
    * 查询记录
    * 更新记录
    * 删除记录
2)执行事务
    * 提交:commit
    * 回滚:rollback

7、pymysql 创建表

from py_L3.pymysql.pymysql_conn import get_conn

# 创建表 testcase
def test_create():
    conn = get_conn()  # 获取连接
    cursor = conn.cursor()  # 获取游标

    sql = """
    CREATE TABLE `testcase` (
    `id` int(11) NOT NULL AUTO_INCREMENT,
    `title` varchar(255) COLLATE utf8_bin NOT NULL,
    `expect` varchar(255) COLLATE utf8_bin NOT NULL,
    `owner` varchar(255) COLLATE utf8_bin NOT NULL,
    PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;
    """
    cursor.execute(sql)  # 执行SQL
    conn.close()  # 关闭连接

8、pymysql 插入操作

from py_L3.pymysql.pymysql_conn import get_conn
# 插入记录
def test_insert():

    conn = get_conn()  # 获取连接
    cursor = conn.cursor()  # 获取游标

    sql = """INSERT INTO testcase
    (id, title, expect, owner)
    values (1, 'S11总决赛', '冠军', 'EDG');
    """

    cursor.execute(sql)  # 执行SQL
    conn.commit()  # 提交

9、执行事务

1)提交操作:commit()
2)回滚操作:rollback()
3)try-catch-finally
from py_L3.pymysql.pymysql_conn import get_conn
def test_insert():
    conn = get_conn()  # 获取连接
    cursor = conn.cursor()  # 获取游标

    sql = """INSERT INTO testcase
    (id, title, expect, owner)
    values (2, 'S11全球总决赛', '冠军', 'EDG');
    """
    try:
        cursor.execute(sql)  # 执行SQL
        conn.commit()  # 提交事务
    except:
        conn.rollback()  # 回滚事务
    finally:
        conn.close()  # 关闭连接

10、pymysql 查询操作

1)fetchone():获取单条记录
2)fetchmany(n):获取 n 条记录
3)fetchall():获取所有结果记录
import sys
from py_L3.pymysql.pymysql_conn import get_conn

def test_retrieve():
    conn = get_conn()  # 获取连接
    cursor = conn.cursor()  # 获取游标
    sql = "SELECT * FROM testcase;"
    # 捕获异常
    try:
        cursor.execute(sql)  # 执行SQL
        record = cursor.fetchone()  # 查询记录
        print(record)
    except Exception as e:
        print(sys.exc_info())  # 打印错误信息
    finally:
        conn.close()  # 关闭连接

11、pymysql 更新操作

from py_L3.pymysql.pymysql_conn import get_conn
# 更新数据表的数据
def test_update():
    conn = get_conn()
    cursor = conn.cursor()
    sql = "UPDATE testcase SET owner='hogwarts' WHERE id=2;"
    try:
        cursor.execute(sql)  # 执行SQL
        conn.commit()  # 提交事务
    except:
        conn.rollback()  # 回滚事务
    finally:
        conn.close()  # 关闭连接

12、pymysql 删除操作

from py_L3.pymysql.pymysql_conn import get_conn
# 删除数据表的数据
def test_delete():
    conn = get_conn()  # 获取连接
    cursor = conn.cursor()  # 获取游标
    sql = "DELETE FROM testcase WHERE id=3;"
    try:
        cursor.execute(sql)  # 执行SQL
        conn.commit()  # 提交事务
    except:
        conn.rollback()  # 回滚事务
    finally:
        conn.close()  # 关闭连接

三、常用第三方库urllib3

1、urllib3 概述

1)内置模块:urllib
2)第三方库:requests、urllib3
3)线程安全
4)连接池管理
5)客户端 SSL/TLS 验证
6)支持 HTTP 和 SOCKS 代理
7)官方文档:https://urllib3.readthedocs.io/en/stable/
8)安装:pip install urllib3

2、urllib3 发送 HTTP 请求

1)导入 urllib3 模块
2)创建 PoolManager 实例
3)调用 request() 方法
import urllib3

def test_HTTP():
    # 创建连接池对象,默认会校验证书
    pm = urllib3.PoolManager()
    # 发送HTTP请求
    res = pm.request(method='GET', url="http://httpbin.org/robots.txt")

    print(type(res))

3、urllib3 HTTPResponse 对象

1)status 属性
2)headers 属性
3)data 属性
import urllib3

def test_response():
    # 创建连接池对象
    pm = urllib3.PoolManager()
    # 发送请求
    resp = pm.request(method='GET', url="http://httpbin.org/ip")

    print(resp.status)  # 查看响应状态状态码
    print(resp.headers)  # 查看响应头信息
    print(resp.data)  # 查看响应原始二进制信息

4、urllib3 解析响应内容

1)二进制响应内容解码
2)JSON 字符串
import urllib3
import json

def test_response():
    pm = urllib3.PoolManager()
    resp = pm.request(method='GET', url="http://httpbin.org/ip")

    # 获取二进制形式的响应内容
    raw = resp.data
    print(type(raw), raw)
    # 使用utf-8解码成字符串
    content = raw.decode('utf-8')
    print(type(content), content)
    # 将JSON字符串解析成字典对象
    dict_obj = json.loads(content)
    print(type(dict_obj), dict_obj)
    print(dict_obj['origin'])

5、urllib3 request 请求参数

1)语法:request(method, url, fields, headers, **)
2)必填
    * method:请求方式
    * url:请求地址
3)选填
    * headers:请求头信息
    * fields:请求体数据
    * body:指定请求体类型
    * tiemout:设置超时时间

6、urllib3 定制请求数据

(1)定制请求头信息
(2)使用 headers 参数
import urllib3
import json

def test_headers():
    pm = urllib3.PoolManager()
    url = "http://httpbin.org/get"

    # 定制请求头
    headers = {'School': 'hogwarts'}
    resp = pm.request('GET', url, headers=headers)
(3)定制查询字符串参数
    * fields 参数:适用于GET, HEAD, DELETE 请求
    * 拼接url:适用于POST, PUT请求
import urllib3
import json

# GET/HEAD/DELETE 请求
def test_fields():
    pm = urllib3.PoolManager()
    url = "http://httpbin.org/get"
    fields = {'school': 'hogwarts'}
    resp = pm.request(method='GET', url=url, fields=fields)

# POST/PUT 请求
def test_urlencode():
   # 从内置库urllib的parse模块导入编码方法
    from urllib.parse import urlencode
    pm = urllib3.PoolManager()
    url = "http://httpbin.org/post"
    # POST和PUT请求需要编码后拼接到URL中
    encoded_str = urlencode({'school': 'hogwarts'})
    resp = pm.request('POST', url=url+"?"+encoded_str)
(4)提交 form 表单数据
    * 类型 'Content-Type': 'multipart/form-data
    * 请求方式:POST、PUT
import urllib3
import json

# POST/PUT 请求
def test_form():
    pm = urllib3.PoolManager()
    url = "http://httpbin.org/post"
    fields = {'school': 'hogwarts'}

    # fields数据会自动转成form格式提交
    resp = pm.request('POST', url, fields=fields)
(5)提交 JSON 格式数据
    * 类型:'Content-Type': 'application/json'
    * 请求方式:POST、PUT
import urllib3
import json

def test_json():
    pm = urllib3.PoolManager()
    url = "http://httpbin.org/post"

    # 设定请求体数据类型
    headers={'Content-Type': 'application/json'}

    # JSON文本数据
    json_str = json.dumps({'school': 'hogwarts'})
    resp = pm.request('POST', url, headers=headers, body=json_str)
(6)timeout :设置超时时间
    * 时间单位:秒
    * 值的格式:float 类型
import urllib3

def test_timeout():
    pm = urllib3.PoolManager()
    # 访问这个地址,服务器会在3秒后响应
    url = "http://httpbin.org/delay/3"

    # 设置超时时长
    resp = pm.request(method='GET', url=url, timeout=4.0)
    assert resp.status == 200

7、urllib3 发送 HTTPS 请求

1)HTTPS 请求默认需要校验证书
2)PoolManager 的 cert_reqs 参数
   * "CERT_REQUIRED":需要校验
   * "CERT_NONE":取消校验
import urllib3
import json

def test_HTTPS():
    # 创建不校验证书的连接池对象
    pm_https = urllib3.PoolManager(cert_reqs="CERT_NONE")
    url = "https://httpbin.ceshiren.com/get"

    # 发送HTTPS请求
    resp = pm_https.request(method='GET', url=url)
    print(json.dumps(resp.data.decode('utf-8')))