一、常用第三方库 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')))
