（一）Redis
1、數據庫分為關系型數據庫和非關系型數據庫：
（1）關系型數據庫分為：MySQL、Oracle、SQL Server、SQLite...
數據庫
表
SQL語句
（2）非關系型數據庫分為：NoSQL（Redis、MongoDB...）
key-value
Redis：存在內存里面，做緩存用。
MongoDB：放在磁盤里面。
2、Redis（Remote Dictionary Server）：（摘自：http://www.redis.cn/）
Redis 是一個開源（BSD許可）的，內存中的數據結構存儲系統，它可以用作數據庫、緩存和消息中間件。 它支持多種類型的數據結構，如 字符串（strings）， 散列（hashes）， 列表（lists）， 集合（sets）， 有序集合（sorted sets） 與范圍查詢， bitmaps， hyperloglogs 和 地理空間（geospatial） 索引半徑查詢。 Redis 內置了 復制（replication），LUA腳本（Lua scripting）， LRU驅動事件（LRU eviction），事務（transactions） 和不同級別的 磁盤持久化（persistence）， 並通過 Redis哨兵（Sentinel）和自動 分區（Cluster）提供高可用性（high availability）。
3、Redis與其他key-value緩存產品的特點：
（1）Redis支持數據的持久化，將內存中的數據保存在磁盤中，重啟后再次可以加載使用；
（2）Redis不僅支持簡單的key-value類型的數據，還支持list、set、zset、hash數據結構存儲；
（3）Redis支持數據備份，master-slave模式的數據備份。
4、根據自己的需求到官網下載安裝：https://redis.io/download
5、Redis 命令參考：http://doc.redisfans.com/
6、數據結構：
（1）String：字符串，最基本的數據類型，最大能存儲512MB。
（2）Hash：散列，是一個string類型的field和value的映射表，特別適合用於存儲對象。（存儲、讀取、修改用戶屬性）
（3）List：列表，是簡單的字符串類別，按照插入順序排序。
（4）Set：集合，是string類型的無序集合，通過hash表實現的。
（5）Sorted Set：有序集合，是string類型元素的集合，且不允許重復的成員。
7、連接Redis：如圖所示，填寫自己的配置：
（1）配置信息填寫：
（2）測試連接：
（3）查看當前庫里的信息，都是key-value 結構：
8、使用Python來操作Redis-String類型，相當於字典類型
（1）Python中使用connection連接Redis
import redis r = redis.Redis(host='118.24.3.40', password='HK139bc&*', port=6379, db=2, decode_responses=True )
（2）Python中利用跳板機ssh遠程連接redis
from sshtunnel import SSHTunnelForwarder # ssh連接庫 import redis # redis模塊 server = SSHTunnelForwarder( ssh_address_or_host= , # ssh地址 ssh_username= , # ssh連接的用戶名 ssh_password= , # ssh連接的用戶名 remote_bind_address=('遠程機器地址', 端口號)) server.start() r=redis.Redis(host='redis地址', port=server.local_bind_port, decode_responses=True) server.close() #關閉連接
（3）string類型：添加數據
import redis r = redis.Redis(host='118.24.3.40', password='HK139bc&*', port=6379, db=2, decode_responses=True ) #string r.set('iphone','{"price":6999,"count":50}')
執行這個程序后，去Redis客戶端查看是否有該數據存在：
（4）string類型：獲取數據
r.set('students','{"name":John,"age":15,"hobby":"playing Games"}') print(r.get('students'))
執行結果為：
（5）string類型：刪除數據
import redis r = redis.Redis(host='118.24.3.40', password='HK139bc&*', port=6379, db=2, decode_responses=True ) r.set('礦泉水','{"price":10.5,"count":20}') data = r.get('礦泉水') r.delete('礦泉水')
（6）string類型：修改數據
import redis r = redis.Redis(host='118.24.3.40', password='HK139bc&*', port=6379, db=2, decode_responses=True ) r.set('礦泉水','{"price":10.5,"count":20}') r.set('礦泉水','xxxx')
（7）bytes：字節類型（上傳一個圖片或者下載一個圖片會遇到這個字節類型）
# bytes類型通過wb寫入 f = open('a.jpg','wb') f.write('xxx')
bytes類型和字符串類型相互轉換
data = r.get('礦泉水') data.decode() # bytes類型轉換成字符串 s='hello' s.encode() # 字符串轉換成bytes類型
（8）Python中設置過期時間：（指定某個key的過期時間）
expire_time = 60 * 60 * 24 r.set('Bob_session','sdgx312vsdrq',expire_time)
Redis客戶端過期時間顯示：
9、使用Python來操作Redis-hash類型，相當於二維字典
（1）Hash添加數據
r.hset('salary','Ann','{"money":2500}') r.hset('salary','Sarah','{"money":5500}') r.hset('salary','Sun','{"money":3860}') r.hset('salary','Beans','{"money":7800}')
Redis客戶端查看新增的數據：
（2）Hash獲取數據
# 獲取單個數據 print(r.hget('salary','Ann')) # 獲取所有的數據 int(r.hgetall('salary'))
（3）Hash-當含有bytes類型，需要變成正常的字典類型
d = r.hgetall('students') new_d = {} for k,v in d.items(): new_d[k.decode()] = v.decode() print(new_d)
從根源上解決Bytes類型問題：
r = redis.Redis(host='118.24.3.40', password='HK139bc&*', port=6379, db=2, decode_responses=True # 解決Bytes類型問題 )
（4）Hash-修改數據
（5）Hash-刪除數據
r.hdel('salary','Bob')
（6）設置過期時間
# 指定某個key的過期時間 r.expire('salary',1000)
10、String和Hush（Redis客戶端展示）
（1）String
（2）Hash
（3）從Redis取數據：Key是什么，Key的類型是什么
（4）刪掉大Key值：
r.delete('students')
11、Redis其他命令：
# 獲取當前數據庫有多少key print(r.keys()) # 獲取具體的某個開頭的key進行匹配 print(r.keys('s*')) #查看key的類型 print(r.type('students')) # 清空當前數據庫里面所有的key print(r.flushdb()) # 清空所有數據庫里面所有的key print(r.flushall()) #判斷Key存不存在 print(r.exists('students'))
12、Redis-管道：批量操作
（1）Python中建立管道，批量執行操作
p= r.pipeline() #建立管道 p.hset('salary','Ann','{"money":2500}') p.hset('salary','Sarah','{"money":5500}') p.hset('salary','Sun','{"money":3860}') p.hset('salary','Beans','{"money":7800}') # 執行，返回一個list，這個list里面是每個命令執行的結果 p.execute()
（2）使用管道和不用管道對比時間：
import redis import time r = redis.Redis(host='118.24.3.40', password='HK139bc&*', port=6379, db=2, decode_responses=True ) # 不用管道： start_time =time.time() for i in range(100): r.set('key%s'%i,'%s'%i) print('不用管道的時間',time.time() - start_time) # 使用管道： start_time =time.time() p = r.pipeline() for i in range(100): p.set('pipeline_key%s'%i,'%s'%i) p.execute() print('用管道的時間',time.time() - start_time)
執行最后的結果為：
13、Python里新增數據帶有文件夾，key里有冒號：就會有文件夾
r.set('product:water','{"count":1,"price":5}') r.set('product:apple','{"count":1,"price":15}') r.set('product:melon','{"count":1,"price":35}')
在Redis客戶端中可以查看到添加的文件夾里數據信息
14、當前服務器需要向另外一個服務器進行Redis數據遷移
編程分析：
r = redis.Redis(host='118.24.3.40', password='HK139bc&*', port=6379, db=14, decode_responses=True ) r2 = redis.Redis(host='118.24.3.40', password='HK139bc&*', port=6378, db=14, decode_responses=True ) p = r2.pipeline() for k in r.keys(): key_type = r.type(k) if key_type == 'string': value = r.get(k) p.set(k,value) elif key_type =='hash': hash_data = r.hgetall(k) # {'XX':XXX} for field,data in hash_data.items(): p.hset(k,field,data) p.execute()
15、若需要多個數據庫操作，需要循環數據庫：（一般用不到）
for i in range(17): r = redis.Redis(host='118.24.3.40', password='HK139bc&*', port=6379, db=i, decode_responses=True ) r2 = redis.Redis(host='118.24.3.40', password='HK139bc&*', port=6379, db=i, decode_responses=True )
（二）加密和解密
1、加密，md5是一種不可逆，同時也是常見的加密方式，需要導入hashlib模塊：
（1）實例1代碼：
import hashlib s = '123456' # 對字符串進行加密 m = hashlib.md5(s) # 加密后的結果用二進制表示 result = m.hexdigest() print(result)
（2）執行結果報錯，這是因為MD5必須傳Bytes類型的：hashlib.md5()
（3）解決方法：將字符串轉換成bytes類型
#導入hashlib模塊 import hashlib # 定義字符串 s = '123456' # 將字符串轉換成bytes類型 s =s.encode() # 對字符串進行加密 m = hashlib.md5(s) # 加密后的結果用二進制表示 result = m.hexdigest() print(result)
執行結果為：
（4）同樣的字符串，MD5出來的結果都一樣
（5）加鹽，例如設置用戶表
import hashlib # 密碼+隨機字符串（隨機字符串就叫鹽值） s = '123=123' + 'djsjks51@#$$' # 轉換成bytes s = s.encode() m = hashlib.md5(s) # bytes，不可逆 result = m.hexdigest() print(result)
（6）定義函數
def my_md5(s): s = str(s) s = s.encode() m = hashlib.md5(s) result = m.hexdigest() return result
2、加密主要有兩種方式：對稱加密和非對稱加密
（1）對稱加密：對稱加密算法在加密和解密時使用的是同一個秘鑰。
對稱加密的模式是：甲方選擇某一種加密規則，對信息進行加密；乙方使用同一種規則，對信息進行解密；
客戶端和服務端進行通信，采用對稱加密，若使用同一個秘鑰很容易破解，若使用不同的秘鑰，秘鑰的管理和傳輸成本又較高；
（2）非對稱加密：非對稱加密算法需要兩個秘鑰來進行加密和解密，分別為：公開密碼（公鑰）和私有秘鑰（私鑰）。
乙方生成兩把密鑰（公鑰和私鑰）。公鑰是公開的，任何人都可以獲得，私鑰是保密的。
甲方獲取乙方的公鑰，然后用它對信息加密。
（3）http和https
常規的http請求，明文傳播；
https可以認為是：http + TLS ，TLS是傳輸層加密協議，前身是SSL協議
3、解密（撞庫解密）
（1）拖庫與撞庫：
拖庫是指黑客盜取了網站的數據庫。撞庫是指黑客用拖庫獲得的用戶名和密碼在其它網站批量嘗試登陸，進而盜取更有價值的東西。由於一些用戶在多個網站用相同的用戶名和密碼，所以撞庫是有一定成功率的。要想撞庫，必須得知道密碼的明文，也就是用戶真正輸入的密碼。
（2）MD5實際上只能破解比較簡單的密碼
4、Base64加密：是可以破解的
（1）例如百度搜索“動畫片”url中里面是轉譯密文
https://www.baidu.com/s?wd=%E5%8A%A8%E7%94%BB%E7%89%87&rsv_spt=1&rsv_iqid=0xcc0d32ae00192d4f&issp=1&f=8&rsv_bp=1&rsv_idx=2&ie=utf-8&rqlang=cn&tn=baiduhome_pg&rsv_enter=0&rsv_dl=tb&oq=%25E5%258A%25A0%25E5%25AF%2586%25E6%2592%259E%25E5%25BA%2593&rsv_t=298fJqJmONwDQA1vi18GP009wC%2Fh67nlOij54DtqzQ7X6efhzWNd7AcKIyB6FO0SBNyM&rsv_btype=t&inputT=10055&rsv_pq=b07987030011a206&rsv_sug3=80&rsv_sug1=32&rsv_sug7=101&bs=%E5%8A%A0%E5%AF%86%E6%92%9E%E5%BA%93
（2）Base64使用：
import base64 # 一般用於傳輸數據過程中 # 加密 s = 'sjfkjdkjxh1145646你好' r = base64.b64encode(s.encode()) result =r.decode() print(result) # 解密 r = base64.b64decode('c2pma2pka2p4aDExNDU2NDbkvaDlpb0=') print(r.decode)
（三）mock接口開發：開發接口，模擬數據
1、Python中可安裝flash模塊或fastapi模塊：
# 建議安裝FastAPI接口 pip install fastapi # 若安裝了FastAPI，還需要安裝一個ASGI服務器，用於生產如Uvicorn或Hypercorn pip install uvicorn # 或者安裝Flask pip install Flask
2、FastAPI （摘自：https://fastapi.tiangolo.com/）
FastAPI是一種現代、快速（高性能）的Web框架，快速編碼，更少的錯誤，直觀簡易，短而健壯，基於API的開放標准：OpenAPI（舊稱Swagger）和JSON Schema。
3、如何mock接口開發：
（1）例如銀行姓名需要查詢產品、開戶接口
# Product { "code":0, "data":[ {"id":1,"product_name":"XXX"}, {"id":2,"product_name":"XXX"}, {"id":3,"product_name":"XXX"}, {"id":4,"product_name":"XXX"}, {"id":5,"product_name":"XXX"} ] }
（2）使用fastapi編寫接口
import fastapi # 自動生成一個API接口文檔 import uvicorn # 啟動一個服務 server = fastapi.FastAPI() @server.get('/login') def login(username:str,password:str): return {'username':username,'password':password} uvicorn.run(server,port=8800,debug=True)
執行結果如下：
（3）復制url在瀏覽器打開，添加參數：http://127.0.0.1:8800/login
（4）在http://127.0.0.1:8800/login，再次補充其他參數：http://127.0.0.1:8800/login?username=Ann&password=123456
（5）定義一個不加參數的接口：
@server.get('/test') def test(): return {'msg':'hello world!'} uvicorn.run(server,port=8800,debug=True)
在瀏覽器打開輸入url：http://127.0.0.1:8800/test，就可以看到無參數的接口信息
（6）查看接口文檔：http://127.0.0.1:8800/docs（根據設置的端口進行查看）
（7）在接口文檔中修改參數進行執行
執行結果如下：
（8）定義一個product接口：
import fastapi import uvicorn server = fastapi.FastAPI() @server.get('/product') def test(): return { 'code':0, 'data':[ {'product_name': '基金1號', 'status': 0}, {'product_name': '基金2號', 'status': 2}, {'product_name': '基金3號', 'status': 1}, {'product_name': '基金4號', 'status': 0}, {'product_name': '基金5號', 'status': 1} ] }
執行結果是：
注： http://127.0.0.1指的是本機的IP地址
若想別人訪問你的地址需要加上host：，例如：
# 自己訪問自己，無需加host uvicorn.run(server,port=8800,debug=True) # 若別人訪問你的地址，需要加上host uvicorn.run(server,port=8800,debug=True,host='0.0.0.0') # http://127.0.0.1:8800/product # http://192.168.1.xx:8800/product
（9）定義一個支付pay的接口
import fastapi import uvicorn server = fastapi.FastAPI() @server.get('/pay') def pay(money:float,status:str): if status == '0': return {'code':1, 'status':'fail'} elif status=='1': return {'code':0, 'status':'success','balance':money} uvicorn.run(server,port=8800,debug=True)
打開瀏覽器，url中輸入：http://127.0.0.1:8800/pay?money=1&status=0，如圖所示：
pay接口中，status加上默認值1：
@server.get('/pay') # status加上默認值1 def pay(money:float,status='1'): if status == '0': return {'code':1, 'status':'fail'} elif status=='1': return {'code':0, 'status':'success','balance':money} uvicorn.run(server,port=8800,debug=True)
接口文檔中顯示默認值1：
（10）定義一個注冊接口
@server.post('/register') def register(username:str,password:str,cpassword:str): if username.strip() and password.strip() and cpassword.strip(): if password.strip() != cpassword.strip(): return {'code':-1, 'msg':'兩次輸入的密碼不一致!'} else: sql = 'select * from app_myuser where username="%s";'%username if tools.execute_sql(sql): return {'code':-1, 'msg':'用戶已經存在！'} else: p = tools.my_md5(password) insert_sql = 'insert into app_myuser (username,password) values ("%s","%s");'%(username,p) tools.execute_sql(insert_sql) return {'code':0, 'msg':'注冊成功！'} else: return {'code':-1, 'msg':'必填參數不能為空!'}
前往postman應用中，進行接口測試：
（11）定義一個登錄接口：
@server.post('/login') def login(username:str=Form(...),password:str=Form(...)): if username.strip() and password.strip(): p = tools.my_md5(password) sql = 'select * from app_myuser where username="%s" and password="%s";' % (username,p) if tools.execute_sql(sql): return {'code': 0, 'msg': '登錄成功！'} else: return {'code':-1, 'msg':'輸入的賬號/密碼不存在！'} else: return {'code':-1, 'msg':'必填參數不能為空!'}
4、flask：開發接口
（1）安裝flask模塊：
pip install flask
（2）用flask定義一個登錄接口：
# flask 是一個輕量級web開發框架 import flask import tools # 因為flask返回的是一個字符串，故需要導入JSON模塊 import json server = flask.Flask(__name__) @server.route('/login',methods=['post','get']) def login(): username = flask.request.values.get('username') password = flask.request.values.get('password') # flask.json.get('xxxx') # 如果入參是json類型的話，使用這樣的方式 # flask.request.cookies.get('xxx') # 獲取cookie里面的數據 # flask.request.headers.get('xx') # flask.request.files.get("xxx") # 文件 if username.strip() and password.strip(): p = tools.my_md5(password) query_sql = 'select * from app_myuser where username= "%s" and passwd="%s";' % (username, p) print(query_sql) if tools.execute_sql(query_sql): return json.dumps({'code': '0', 'msg': '登錄成功'},ensure_ascii=False) else: return json.dumps({'code': '-1', 'msg': '輸入的用戶名/密碼錯誤'}) else: return json.dumps({'code': '-1', 'msg': '不能為空'})
server.run(host='0.0.0.0',port=8999,debug=True)
（3）定義一個注冊接口：
import flask import tools # 因為flask返回的是一個字符串，故需要導入JSON模塊 import json server = flask.Flask(__name__) @server.route('/reg',methods=['post','get']) def reg(): username = flask.request.values.get('username') password = flask.request.values.get('password') cpassword = flask.request.values.get('cpassword') if username.strip() and password.strip() and cpassword.strip(): if password.strip() != cpassword.strip(): return json.dumps({'code': -1, 'msg': '兩次輸入的密碼不一樣'}) else: sql='select * from app_myuser where username="%s";'%username if tools.execute_sql(sql): return json.dumps({'code':-1,'msg':'用戶已經存在'}) else: p = tools.my_md5(password) insert_sql = 'insert into app_myuser (username,passwd) value ("%s","%s");'%(username,p) tools.execute_sql(insert_sql) return json.dumps({'code':0,'msg':'注冊成功！'},ensure_ascii=False) else: return json.dumps({'code':-1,'msg':'必填參數不能為空'}) server.run(host='0.0.0.0',port=8999,debug=True)
（四）網絡請求：做接口自動化
1、Python自帶的模塊發送網絡請求：urllib
（1）安裝urllib模塊
import urllib
（2）urllib模塊：get請求
from urllib import request import json url = 'http://127.0.0.1:8999/login?username=Bob&password=123456' # get請求 req = request.urlopen(url) dic = json.loads(req.read().decode()) print(dic)
（3）urllib模塊：post請求
# post請求 from urllib import request from urllib.parse import urlencodeimport json url = 'http://127.0.0.1:8999/login' data ={'username':'Ann01','password':'123456'} req = request.urlopen(url,urlencode(data).encode()) dic = json.loads(req.read().decode()) print(dic)
2、requests模塊
（1）接口測試的請求方式：get、post、傳cookie、傳文件、傳json、傳headers
（2）安裝requests模塊
pip install requests
（3）使用requests查看請求
import requests import json url = 'http://127.0.0.1:8999/login' data ={'username':'Ann01','password':'123456'} r = requests.get(url,data) # 字典類型 print(r.json) # 字符串格式 print(r.text) # bytes類型的 print(r.content) #返回的狀態碼:200、404、500、502 print(r.status_code)
（4）get請求方式（根據使用情況選擇不同的方法）：
# 字典格式，為了獲取某個字段的值：d.get('XXX') r.json # 字符串格式：接口返回結果不處理，直接存入庫里或者json文件里面 r.text # bytes類型，例如下載一個圖片 r.content #返回的狀態碼 r.status_code
（5）post請求方式：
r = requests.post(url,data) print(r.json()) print(r.text) print(r.content) print(r.status_code)
（6）post請求url中，有時會遇到url帶有參數，這時候需要：
r = requests.post(url,data,params={"version":1.0)
# params是把參數傳到url后頭的
（7）傳cookie：
# url=http://www.nnzhp.cn/ # cookie:PHPSESSID=8a97d6ac860319bf067a674b8b5a8e34 cookie = {'PHPSESSID':'8a97d6ac860319bf067a674b8b5a8e34'} r = requests.post(url,data=data,params={"version":1.0},cookies=cookie)
（8）傳headers：
headers = {'User-Agent':'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/85.0.4183.102 Safari/537.36','cookie':'PHPSESSID':'8a97d6ac860319bf067a674b8b5a8e34'}
r = requests.post(url,data=data,params={"version":1.0},headers=headers)
（9）傳json和傳文件
# 傳json r = requests.post(url,json=data) print(r.json()) # 傳文件 url = 'http://api.nnzhp.cn/api/file/file_upload' data = {'file':open('XXX.xls','rb')} r = requests.post(url,files=data)
（10）下載文件
#下載文件 r = requests.get('https://ss0.bdstatic.com/70cFuHSh_Q1YnxGkpoWK1HF6hhy/it/u=3086278442,1750390944&fm=26&gp=0.jpg') f = open('dog.jpg','wb') print(r.content) f.close()
注：有時會遇到https的報錯，這時需要：
r = requests.get('https://ss0.bdstatic.com/70cFuHSh_Q1YnxGkpoWK1HF6hhy/it/u=3086278442,1750390944&fm=26&gp=0.jpg',verify=False)