python 操作三大主流数据库 第十一课

tags:

• python3
• 慕课网

categories:

• python3
• redispy
• Redis 网易新闻实战

文章目录

• python 操作三大主流数据库 第十一课
• 第一节 创建辅助类，导入实战数据
• 创建redis辅助类，redis_news.py
• 2. 批量导入测试数据init_news_redis.py

第一节 创建辅助类，导入实战数据

创建redis辅助类，redis_news.py

import math
import redis

NEWS_FIELDS = (
"title",
"img_url",
"content",
"is_valid",
"news_type",
"created_at",
"updated_at"
)

class RedisNews(object):
def __init__(self):
# 如果返回是二进制类似 b'3\xe6\x9c\x885\xe6\x97\xa5\xe...'需要加decode_responses=True
try:
self.r = redis.StrictRedis(host='localhost',
port=6379, encoding='utf-8',
decode_responses=True,
db=1)
except Exception as e:
print('redis connect faild')

def _news_id(self, int_id):
''' 新闻id '''
return 'news:%d' % int(int_id)

def _news_type(self, news_type):
''' 新闻类型 '''
return 'news_type:%s' % news_type

def _news_list_name(self):
''' 新闻列表名称 '''
return 'news'

def add_new_from_page(self, news_obj):
''' 从页面新增新闻数据 '''

# 记录news_id的数字自增1
int_id = self.r.incr('news_id', amount=1)
# 获取新闻id
news_id = self._news_id(int_id)

# 新闻列表中添加
self.r.lpush(self._news_list_name(), int_id)

# 新闻类型中添加信息
news_type = news_obj['news_type']
self.r.sadd(news_type, int_id)

# 新闻内容中添加信息
self.r.hmset(news_id, news_obj)

def add_news(self, news_obj):
''' 新增新闻数据 '''
# 获取到新闻的id
int_id = int(self.r.incr('news_id'))
# 　拼接新闻数据Hash key(news:2)
news_id = self._news_id(int_id)

# 存储新闻数据(hash)
rest = self.r.hmset(news_id, news_obj)

# 存储新闻的id list
self.r.lpush(self._news_list_name(), int(int_id))

# 存储新闻的类别-新闻id(set)
news_type = self._news_type(news_obj['news_type'])
self.r.sadd(news_type, int_id)
return rest

def add_news_with_transaction(self, news_obj):
''' 使用事务来新增新闻数据 '''
pipe = self.r.pipeline(transaction=True)
int_id = self.r.incr('news_id')
news_id = self._news_id(int_id)

# 使用列表list获取新闻的id
pipe.lpush(self._news_list_name(), int_id)

# 使用hash来保存新闻具体内容
pipe.hmset(news_id, news_obj)

# 使用hash来保存新闻分类信息
news_type = self._news_type(news_obj['news_type'])
pipe.sadd(news_type, int_id)

rest = pipe.execute()
return rest

def get_all_news(self):
''' 获取所有新闻信息 '''

# 获取id列表
id_list = self.r.lrange(self._news_list_name(), 0, -1)
data_list = []

for int_id in id_list:
# 获取具体新闻内容
news_id = self._news_id(int_id)
data = self.r.hgetall(news_id)
data['id'] = int_id
# print(data)
data_list.append(data)

return data_list

def get_news_from_id(self, news_id):
''' 根据新闻id获取新闻内容 '''
news_id = self._news_id(news_id)

# 根据新闻id获取新闻内容
news_obj = self.r.hgetall(news_id)
return news_obj

def get_news_from_cat(self, cat_name):
''' 根据新闻类型获取新闻内容 '''
news_list = []

# 获取新闻类型
news_type = self._news_type(cat_name)
# print(news_type)
# 获取新闻类型集合中新闻id的列表
id_list = self.r.smembers(news_type)
print(id_list)
for int_id in id_list:
# 获取新闻id
news_id = self._news_id(int_id)
# 根据新闻id获取新闻内容
data = self.r.hgetall(news_id)
data['id'] = int_id
news_list.append(data)
return news_list

def update_news(self, pk, news_obj):
''' 新闻的修改 '''
news_id = self._news_id(pk)

# 修改新闻
rest = self.r.hmset(news_id, news_obj)
return rest

def delete_news(self, pk, news_obj):
'''
新闻的删除，物理删除

关于常用的方法可以通过查询redis的命令类型判断是list,string还是hash或者set
1.命令列表定位到具体命令：http://www.redis.cn/commands.html#hash
2.找到命令后，查询api的用法http://redis-py.readthedocs.io/en/latest/

'''

# 获取新闻id
news_id = self._news_id(pk)
# 从新闻列表中删除新闻id
self.r.lrem(self._news_list_name(), 0, pk)
# 从新闻的类型set集合中清理新闻id
news_type = self._news_type(news_obj['news_type'])
self.r.srem(news_type, pk)
# 从新闻的内容hash列表中清理具体的新闻内容NEWS_FIELDS(具体的列信息)
self.r.hdel(news_id, *NEWS_FIELDS)

def paginate(self, page=1, per_page=5):
''' 新闻后台分页 '''
if page is None:
page = 1

data_list = []
# 开始页，结束页面
start = (page - 1) * per_page
end = page * per_page - 1

# 获取所有新闻列表(计算页码使用)
list_ids = self.r.lrange(self._news_list_name(), 0, -1)

# 获取新闻列表
id_list = self.r.lrange(self._news_list_name(), start, end)
# print('id_list%s' % id_list)

for int_id in id_list:
news_id = self._news_id(int_id)
# 根据新闻id获取新闻内容
data = self.r.hgetall(news_id)
data['id'] = int_id
data_list.append(data)
# print('data_list%s' % data_list)
return Pagenation(data_list, page, per_page, list_ids)

def init_news(self, data_list):
''' 批量导入新闻数据 '''
for news_obj in data_list:
rest = self.add_news_with_transaction(news_obj)
print(rest)

class Pagenation(object):
''' 分页类 '''

def __init__(self, data_list, now_page, per_page, list_ids):
self.now_page = now_page
self.data_list = data_list
self.per_page = per_page
self.list_ids = list_ids

@property
def page(self):
''' 当前页 '''
return self.now_page

@property
def items(self):
''' 返回页面数据 '''
return self.data_list

@property
def prev_num(self):
''' 上一页 '''
return self.now_page - 1

@property
def next_num(self):
''' 下一页页码 '''
return self.now_page + 1

@property
def has_prev(self):
''' 是否有上一页 '''
return self.now_page > 1

@property
def has_next(self):
''' 是否有下一页 '''
return self.per_page == len(self.data_list)

def iter_pages(self):
''' 页码 '''
# 获取所有的id长度(即新闻条数)除以每页显示的页面，得到取进一位的整数
total_page = math.ceil(len(self.list_ids) / self.per_page) + 1
# print('total_page=%d' % total_page)
return range(1, total_page)

2. 批量导入测试数据init_news_redis.py

#coding:utf-8

'''
初始化新闻数据
'''

from datetime import datetime

import sys
import os
sys.path.append(os.path.dirname(os.path.dirname(__file__)))

from redis_news import RedisNews

list_news = [
{
"title":"朝鲜特种部队视频公布 展示士兵身体素质与意志",
"img_url":"/static/img/news/01.png",
"content":"在4月15日举行的朝鲜阅兵式上，除了各式展出的导弹，最亮眼的恐怕要数佩戴夜视仪的朝鲜特种部队了。4月19日，俄罗斯卫星网发布了截取自朝鲜官方电视台关于朝鲜特种部队士兵训练与展示的视频。在视频中，尽管训练科目并无太多新意，但是朝鲜士兵展示出了高度惊人的身体素质与顽强意志。",
"is_valid": 1,
"news_type":"推荐",
"created_at": datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
"updated_at": datetime.now().strftime('%Y-%m-%d %H:%M:%S')
},
{
"title":"男子长得像\"祁同伟\"挨打 打人者:为何加害检察官",
"img_url":"/static/img/news/02.png",
"content":"因与热门电视剧中人物长相相近,男子竟然招来一顿拳打脚踢。4月19日,打人男子周某被抓获。半个月前,酒后的周某看到KTV里有一名男子很像电视剧中的反派。二话不说,周某冲上去就问你为什么要加害检察官?男子莫名其妙,回了一句神经病。周某一听气不打一处来,对着男子就是一顿拳打脚踢,嘴里面还念叨着,“叫你加害检察官,我打死你!”随后,周某趁机逃走。受伤男子立即报警,周某被上海警方上网通缉。",
"is_valid": 1,
"news_type":"百家",
"created_at": datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
"updated_at": datetime.now().strftime('%Y-%m-%d %H:%M:%S')
},
{
"title":"导弹来袭怎么办？日本政府呼吁国民堕入地下通道",
"img_url":"/static/img/news/03.JPEG",
"content":"中新网4月21日电 据日媒报道，日本政府本月21日公布了弹道导弹可能落在国内时应采取的应对方法，呼吁民众身处室外时\“尽可能躲入坚固的建筑物或地下通道\”等",
"is_valid": 1,
"news_type":"本地",
"created_at": datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
"updated_at": datetime.now().strftime('%Y-%m-%d %H:%M:%S')
},
{
"title":"美监:朝在建能发射3发以上导弹的3000吨级新潜艇",
"img_url":"/static/img/news/04.JPEG",
"content":"【环球网报道】据韩联社4月21日报道，美国保守媒体《华盛顿自由灯塔》20日引用联合国报告报道称，朝鲜可能对“新浦”级潜艇进行改装，使其可连发多枚潜射导弹，韩国军方负责人21日对此表示，需进一步分析，持谨慎态度",
"is_valid": 1,
"news_type":"推荐",
"created_at": datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
"updated_at": datetime.now().strftime('%Y-%m-%d %H:%M:%S')
},
{
"title":"证监会：前发审委员冯小树违法买卖股票被罚4.99亿",
"img_url":"/static/img/news/05.png",
"content":"证监会新闻发言人张晓军21日表示，中央第七巡视组对证监会开展专项巡视期间，向证监会移交了前深交所工作人员、曾任股票发审委兼职委员冯小树涉嫌违法买卖股票的相关线索。会党委对相关线索高度重视，要求予以彻查。经过调查审理，通过对复杂商业架构的层层剖析，对繁复资金往来情况的抽丝剥茧，证监会查明，冯小树先后以岳母彭某嫦、配偶之妹何某梅名义入股拟上市公司，并在公司上市后抛售股票获利巨额利益，其交易金额累计达到2.51亿元，获利金额达2.48亿元",
"is_valid": 1,
"news_type":"百家",
"created_at": datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
"updated_at": datetime.now().strftime('%Y-%m-%d %H:%M:%S')
},
{
"title":"外交部回应安倍参拜靖国神社:同军国主义划清界限",
"img_url":"/static/img/news/06.jpg",
"content":"新闻图片",
"is_valid": 1,
"news_type":"推荐",
"created_at": datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
"updated_at": datetime.now().strftime('%Y-%m-%d %H:%M:%S')
},
{
"title":"\"萨德\"供地违法？韩民众联名起诉要求撤回供地",
"img_url":"/static/img/news/07.jpg",
"content":"代理本案的“民主社会律师聚会”主张，韩国《国有财产特例限制法》第4条规定，未遵守该法附表中相关法案的国有财产特例无效，该法案附表中并不包括《驻韩美军地位协定》或有关履行《驻韩美军地位协定》的特别法案，因此，韩国政府供地是违反《国有财产特例限制法》向美军提供国有财产特例。",
"is_valid": 1,
"news_type":"推荐",
"created_at": datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
"updated_at": datetime.now().strftime('%Y-%m-%d %H:%M:%S')
},
{
"title":"金正恩:要由朝鲜民族自己谱写祖国统一新历史",
"img_url":"/static/img/news/08.JPEG",
"content":"3月5日，在朝鲜平壤，青瓦台国家安保室室长郑义溶（左）与朝鲜劳动党委员长金正恩握手。新华社平壤3月6日电 据朝中社6日报道，朝鲜最高领导人金正恩5日会见当天抵朝的韩国特使团，双方就北南首脑会晤交换意见并达成共识。报道说，金正恩在听取韩方特使转达的韩国总统文在寅有关南北首脑会晤的意愿后，与韩方交换意见并达成共识，他要求有关部门就此尽快采取相关实际举措。会见时，文在寅总统特使、青瓦台国家安保室长郑义溶向金正恩转交了文在寅的亲笔信。报道说，金正恩与韩方代表团就改善北南关系、保障朝鲜半岛和平稳定进行了开诚布公的交谈，还就缓和朝鲜半岛军事紧张状态、促进北南间多方面对话和接触、合作与交流交换了意见。金正恩说，要由朝鲜民族自己来齐心协力共同推动北南关系发展、谱写祖国统一的新历史，这是朝鲜一贯的原则立场，也是他本人坚定不移的意志。报道说，韩国特使团成员就金正恩向平昌冬奥会派遣高级别代表团等多个大规模代表团、帮助大会取得圆满成功表示感谢。金正恩说，作为血脉相连的同一民族，共同庆祝民族喜事并互相帮助，这次冬奥会是营造北南和解团结与对话良好气氛的重要契机。除郑义溶外，特使团其他成员韩国国家情报院院长徐薰、统一部次官千海成、国家情报院次长金相均和青瓦台国政状况室室长尹建永也参加了会见。朝鲜劳动党中央委员会副委员长金英哲和朝鲜劳动党中央委员会第一副部长金与正会见时在座。据朝中社报道，金正恩5日为韩国特使团成员举行了晚宴，金正恩的夫人李雪主，以及金英哲、金与正等参加晚宴。另据韩国媒体报道，韩国总统府青瓦台发言人金宜谦6日说，5日的会见和晚宴持续4个多小时。特使团将在结束后续会谈后，于6日下午返回首尔。青瓦台消息人士表示，这次会见成果“不令人失望”，韩朝就包括首脑会晤等事项达成一定程度的一致。",
"is_valid": 1,
"news_type":"百家",
"created_at": datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
"updated_at": datetime.now().strftime('%Y-%m-%d %H:%M:%S')
}
]

def main():
redis_news = RedisNews()
rest = redis_news.init_news(list_news)

if __name__ == "__main__":
main()

第二节 网易新闻后台控制flask

import sys
from datetime import datetime
from werkzeug import secure_filename
from flask import Flask, request, render_template, redirect, flash, url_for, abort
from flask_uploads import UploadSet, configure_uploads, IMAGES, patch_request_class

from forms import NewsForm_redis
from redis_news import RedisNews

app = Flask(__name__, template_folder='templates_redis')
app.config['SECRET_KEY'] = 'this is a redis key'
app.config['UPLOADED_PHOTOS_DEST'] = "/data/three_db_python/redis_version01/static/img/news"
query = RedisNews()

photos = UploadSet('photos', IMAGES)
configure_uploads(app, photos)

patch_request_class(app)  # set maximum file size, default is 16MB

@app.route("/", methods=['GET'])
def index():
''' 获取新闻列表 '''
news_list = query.get_all_news()
return render_template("index.html", news_list=news_list)

@app.route("/cat/<name>/", methods=['GET'])
def cat(name):
''' 获取新闻列表 '''
news_list = query.get_news_from_cat(name)
return render_template("cat.html", news_list=news_list)

@app.route("/detail/<int:pk>/", methods=['GET'])
def detail(pk):
''' 获取新闻列表 '''
news_obj = query.get_news_from_id(pk)
return render_template("detail.html", obj=news_obj)

@app.route("/admin/", methods=['GET'])
@app.route("/admin/<int:page>/", methods=['GET'])
def admin(page=None):
''' 获取后台新闻列表 '''
if page is None:
page = 1
page_data = query.paginate(page, 3)
return render_template("admin/index.html", page_data=page_data)

@app.route("/admin/add/", methods=['GET', 'POST'])
def add():
''' 从后台页面添加新闻 '''
form = NewsForm_redis()
news_obj = {}
# 提交增加
if form.validate_on_submit():
# 图片文件名
#filename = photos.save(form.photo.data)

news_obj['title'] = form.title.data
news_obj['news_type'] = form.news_type.data
news_obj['img_url'] = form.img_url.data
news_obj['content'] = form.content.data
news_obj['created_at'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
news_obj['updated_at'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
query.add_new_from_page(news_obj)
flash('添加成功')
return redirect(url_for('admin'))

return render_template("admin/add.html", form=form)

@app.route("/admin/update/<int:pk>/", methods=['GET', 'POST'])
def update(pk):
''' 获取后台新闻列表 '''
# 获取新闻
news_obj = query.get_news_from_id(pk)
if news_obj is None:
abort('no this news')
form = NewsForm_redis(data=news_obj)

# 提交修改
if form.validate_on_submit():
news_obj['title'] = form.title.data
news_obj['news_type'] = form.news_type.data
#news_obj['img_url'] = form.img_url.data
news_obj['content'] = form.content.data
news_obj['updated_at'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')

query.update_news(pk, news_obj)

flash('新闻修改成功')
return redirect(url_for('admin'))
return render_template("admin/update.html", form=form)

@app.route("/admin/delete/<int:pk>/", methods=['GET', 'POST'])
def delete(pk):
''' 删除新闻 '''
news_obj = query.get_news_from_id(pk)
if news_obj:
query.delete_news(pk, news_obj)
return 'yes'

return 'no'

if __name__ == "__main__":
app.run(debug=True)import sys
from datetime import datetime
from werkzeug import secure_filename
from flask import Flask, request, render_template, redirect, flash, url_for, abort
from flask_uploads import UploadSet, configure_uploads, IMAGES, patch_request_class

from forms import NewsForm_redis
from redis_news import RedisNews

app = Flask(__name__, template_folder='templates_redis')
app.config['SECRET_KEY'] = 'this is a redis key'
app.config['UPLOADED_PHOTOS_DEST'] = "/data/three_db_python/redis_version01/static/img/news"
query = RedisNews()

photos = UploadSet('photos', IMAGES)
configure_uploads(app, photos)

patch_request_class(app)  # set maximum file size, default is 16MB

@app.route("/", methods=['GET'])
def index():
''' 获取新闻列表 '''
news_list = query.get_all_news()
return render_template("index.html", news_list=news_list)

@app.route("/cat/<name>/", methods=['GET'])
def cat(name):
''' 获取新闻列表 '''
news_list = query.get_news_from_cat(name)
return render_template("cat.html", news_list=news_list)

@app.route("/detail/<int:pk>/", methods=['GET'])
def detail(pk):
''' 获取新闻列表 '''
news_obj = query.get_news_from_id(pk)
return render_template("detail.html", obj=news_obj)

@app.route("/admin/", methods=['GET'])
@app.route("/admin/<int:page>/", methods=['GET'])
def admin(page=None):
''' 获取后台新闻列表 '''
if page is None:
page = 1
page_data = query.paginate(page, 3)
return render_template("admin/index.html", page_data=page_data)

@app.route("/admin/add/", methods=['GET', 'POST'])
def add():
''' 从后台页面添加新闻 '''
form = NewsForm_redis()
news_obj = {}
# 提交增加
if form.validate_on_submit():
# 图片文件名
#filename = photos.save(form.photo.data)

news_obj['title'] = form.title.data
news_obj['news_type'] = form.news_type.data
news_obj['img_url'] = form.img_url.data
news_obj['content'] = form.content.data
news_obj['created_at'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
news_obj['updated_at'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
query.add_new_from_page(news_obj)
flash('添加成功')
return redirect(url_for('admin'))

return render_template("admin/add.html", form=form)

@app.route("/admin/update/<int:pk>/", methods=['GET', 'POST'])
def update(pk):
''' 获取后台新闻列表 '''
# 获取新闻
news_obj = query.get_news_from_id(pk)
if news_obj is None:
abort('no this news')
form = NewsForm_redis(data=news_obj)

# 提交修改
if form.validate_on_submit():
news_obj['title'] = form.title.data
news_obj['news_type'] = form.news_type.data
#news_obj['img_url'] = form.img_url.data
news_obj['content'] = form.content.data
news_obj['updated_at'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')

query.update_news(pk, news_obj)

flash('新闻修改成功')
return redirect(url_for('admin'))
return render_template("admin/update.html", form=form)

@app.route("/admin/delete/<int:pk>/", methods=['GET', 'POST'])
def delete(pk):
''' 删除新闻 '''
news_obj = query.get_news_from_id(pk)
if news_obj:
query.delete_news(pk, news_obj)
return 'yes'

return 'no'

if __name__ == "__main__":
app.run(debug=True)

• 点赞
• 收藏
• 分享
• 文章举报