python123平台作业答案第十二周

0.本周知识点预览

Contextlib

Redis发布订阅

RabbitMQ

pymysql

SQLAchemy

1.Contextlib模块

contextlib模块的contextmanager 可以实现用with来管理上下文，类似于 with open('test.txt','r') as f，这样打开文件操作后就可以自动关闭文件。

1.范例一(自定义函数)：

###contextlib (实现with上下文管理)

importcontextlib

list1= [1,2,3]

str1= "lk"###此处必须是这个装饰器

@contextlib.contextmanagerdeffunc(l, s):

l.append(s)try:###执行到yield时,中断跳出函数

yield

finally:print(l)

with func(list1, str1):print(123)print(456)

执行结果如下：

123

456[1, 2, 3, 'lk']

代码解析：以上代码的执行顺序为：

1.加载list1,str1,contextlib.contextmanager装饰器

2.执行with func(list1, str1)

3.执行def func(l, s), l.append(s)

4.try,yield,跳出函数回到with func(list1, str1) 执行print

5.执行完print后回到def func中的yield处，继续往下执行

2.范例二(socket)：

###利用上下文可以处理文件那样处理类似socket,自动关闭连接

importcontextlibimportsocket

@contextlib.contextmanagerdefbase_socket(host, port):

sk=socket.socket()

sk.bind((host, port))

sk.listen(5)print(123)try:yieldskfinally:print(789)

sk.close()

with base_socket("127.0.0.1", 8888) as sock:print(456)

执行结果如下：

123

456

789

代码解析：这个是context对socket的应用，在项目中就可以这么写。执行顺序和上个例子相同。

2.Redis 发布订阅

自定义redis基础类：

## redis 发布订阅

importredisclassRedisHelper:def __init__(self):###创建redis连接对象

self.__conn = redis.Redis(host="127.0.0.1", port=6379)defpublic(self, msg, chan):###publish 方法,把信息发布到频道上,返回消息被传递的订阅者的数量

self.__conn.publish(chan, msg)returnTruedefsubscribe(self, chan):###pubsub 方法的意思是,返回一个发布或者订阅的对象,用这个对象,你就能订阅这个频道,监听给发给这些频道的消息

pub = self.__conn.pubsub()###subscribe 方法: 订阅频道

pub.subscribe(chan)###parse_response 方法:解析从发布者/订阅者命令的响应

pub.parse_response()return pub

发布者代码：

importtest

fabu=test.RedisHelper()whileTrue:

inp= input(">>>")if inp == "exit":break

else:

fabu.public("%s" % inp, "998")

订阅者代码：

importtest

dingyue=test.RedisHelper()whileTrue:###subscribe 方法 -> 订阅频道

data = dingyue.subscribe("998")###parse_response 方法:解析从发布者/订阅者命令的响应

print(data.parse_response())

执行结果：

1、先执行订阅者代码，这时会卡在这里，等待接收消息。

2、后执行发布者代码，这时当发布消息时，订阅者就会收到消息。

3.RabbitMQ

1.未利用exchange

生产者代码：

importpika###第一件事，我们要做的就是与rabbitmq-server建立连接

connection = pika.BlockingConnection(pika.ConnectionParameters(host="127.0.0.1"))

channel=connection.channel()###创建一个队列,假如队列不存在,但是为了确保队列成功创建,C/P两端最好都创建队列###durable=True 消息持久化

channel.queue_declare("hello_lk3", durable=True)###在Rabbitmq中，一个消息不能直接发送给queue，需要经过一个exchange，后续会讲到，现在我们只需将exchange设置为空字符串

channel.basic_publish(exchange='', routing_key="hello_lk1", body="fuck", properties=pika.BasicProperties(delivery_mode=2,))print("[x] sent 'fuck'")

connection.close()

消费者代码：

importpika###第一件事，我们要做的就是与rabbitmq-server建立连接

connection = pika.BlockingConnection(pika.ConnectionParameters(host="127.0.0.1"))###创建一个频道

channel =connection.channel()###创建一个队列,假如队列不存在,但是为了确保队列成功创建,C/P两端最好都创建队列###durable=True 消息持久化

channel.queue_declare("hello_lk3", durable=True)###函数名不必须叫callback,callback函数就是将接收到的消息打印在屏幕上

defcallback(ch, mechod, properties, body):print("[%s] is received" %body)###无限循环监听,调用callback,队列名,no_ack的含义为,当时True时,只要订阅到消息,立刻返回ack,这是TCP层面的,并不能确保消息成功消费###假如no_ack为False时,订阅到消息后要处理成功后才返回ack,这是业务逻辑层面的,确保消费者成功消费消息

channel.basic_consume(callback, queue="hello_lk1", no_ack=True)print("现在开始消费消息...")

channel.start_consuming()

代码执行结果：

生产者：

[x] sent 'fuck'Process finished with exit code 0

消费者：

现在开始消费消息...

[b'fuck'] isreceived

[b'fuck'] is received

代码解析：

1.先执行消费者代码，在执行生产者代码，可以看到如上图所示结果。

2.执行两次消费者代码，会发现生产者每生产个消息，消费者会轮训的来消费。

3.在步骤2中，假如不想让消费者轮训消费而是先来先得的消费，则需要在消费者代码中加入一行：channel.basic_qos(prefetch_count=1) 表示谁来谁取，不再按照奇偶数排列。

2.使用exchange发布订阅(常用)

1.fanout exchange

这是处理逻辑最简单的exchange类型，实际上它没有任何逻辑，它把进入该exchange的消息全部转发给每一个绑定的队列中，如果这个exchange没有队列与之绑定，消息会被丢弃。然后通过exchange发送消息，routing key可以随便填写，因为是fanout类型的exchange，routing key不起作用。

生产者代码：

importpika###第一件事，我们要做的就是与rabbitmq-server建立连接

connection = pika.BlockingConnection(pika.ConnectionParameters(host="127.0.0.1"))

channel=connection.channel()###创建一个exchange,生产者直接向exchange发消息,而不是队列.###type: fanout类型:把进入该exchange的消息全部转发给每一个绑定的队列中，如果这个exchange没有队列与之绑定，消息会被丢弃

channel.exchange_declare(exchange="lk", type="fanout")

message= 'hello'channel.basic_publish(exchange='lk', routing_key='', body=message)print("send MSG: %s" %message)

connection.close()

消费者1代码：

##exchange 版

importpika###第一件事，我们要做的就是与rabbitmq-server建立连接

connection = pika.BlockingConnection(pika.ConnectionParameters(host="127.0.0.1"))###创建一个频道

channel =connection.channel()###创建一个exchange,假如不存在,但是为了确保其成功创建,C/P两端最好都创建exchange###lk 为exchange名###type: fanout类型:把进入该exchange的消息全部转发给每一个绑定的队列中，如果这个exchange没有队列与之绑定，消息会被丢弃

channel.exchange_declare(exchange="lk", type='fanout')##随机创建队列#result = channel.queue_declare(exclusive=True)#queue_name = result.method.queue###指定创建队列

channel.queue_declare("hello_lk5")##绑定队列到exchange

channel.queue_bind(exchange="lk", queue="hello_lk5")###函数名不必须叫callback,callback函数就是将接收到的消息打印在屏幕上

defcallback1(ch, method, propreties, body):print("[x] 收到 %s" %body)

channel.basic_consume(callback1, queue="hello_lk5", no_ack=True)

channel.start_consuming()

消费者2代码：

##exchange 版

importpika###第一件事，我们要做的就是与rabbitmq-server建立连接

connection = pika.BlockingConnection(pika.ConnectionParameters(host="127.0.0.1"))###创建一个频道

channel.exchange_declare(exchange="lk", type='fanout')##随机创建队列#result = channel.queue_declare(exclusive=True)#queue_name = result.method.queue###指定创建队列

channel.queue_declare("hello_lk6")##绑定队列到exchange

channel.queue_bind(exchange="lk", queue="hello_lk6")###函数名必须叫callback,callback函数就是将接收到的消息打印在屏幕上

defcallback1(ch, method, propreties, body):print("[x] 收到 %s" %body)

channel.basic_consume(callback1, queue="hello_lk6", no_ack=True)

channel.start_consuming()

执行结果：

生产者：

send MSG: hello

Process finished with exit code 0

消费者1：

[x] 收到 b'hello'

消费者2：

[x] 收到 b'hello'

代码解析：生产者直接向exchange发消息，这时，消费者创建并绑定队列到exchange上，生产者一旦发布，所有队列都会收到消息。

2.direct exchange

这种类型的交换机Fancout 类型的交换机智能一些，它会根据routing key来决定把消息具体扔到哪个消息队列中。通过exchange发消息的时候会指定一个routing key，只有当routing key和与队列绑定的routing key一样的时候，消息才对发送到对应的消息队列。即，如果与某个队列绑定的routing key叫hello.world，则通过exchange发送的routing key必须也是hello.world，该队列才能接收到消息（可按上述步骤进行验证）。这种情况下，队列之间是互斥关系，一个消息最多只能进入一个队列。

生产者代码：

importpika###第一件事，我们要做的就是与rabbitmq-server建立连接

connection =pika.BlockingConnection(pika.ConnectionParameters(

host='127.0.0.1'))

channel=connection.channel()###创建一个exchange, 类型是direct,

channel.exchange_declare(exchange='direct_logs',

type='direct')###定义关键字severity, 每次发消息时会指定关键字.

severity = "error"

###message 是要发送的消息

message = "123"

###通过定义好的exchange发送消息,关键字也是定好的,发送指定的消息

channel.basic_publish(exchange='direct_logs',

routing_key=severity,

body=message)print("[x] Sent %r:%r" %(severity, message))

connection.close()

消费者1代码：

importpika#import sys

###第一件事，我们要做的就是与rabbitmq-server建立连接

connection =pika.BlockingConnection(pika.ConnectionParameters(

host='127.0.0.1'))###创建一个频道

channel =connection.channel()###创建一个exchange, 类型是direct,

channel.exchange_declare(exchange='direct_logs',

type='direct')###随机创建一个队列

result = channel.queue_declare(exclusive=True)

queue_name=result.method.queue###关联关键字,只要生产者发布的消息关联了以下关键字,订阅者便能在绑定的队列中收到消息

severities = ["info", "waring", "error"]###绑定关键字,队列到exchange.

for severity inseverities:

channel.queue_bind(exchange='direct_logs',

queue=queue_name,

routing_key=severity)print('[*] Waiting for logs. To exit press CTRL+C')###打印订阅到的消息

defcallback(ch, method, properties, body):print("[x] %r:%r" %(method.routing_key, body))###循环监听队列

channel.basic_consume(callback,

queue=queue_name,

no_ack=True)

channel.start_consuming()

消费者2代码：

importpika#import sys

###第一件事，我们要做的就是与rabbitmq-server建立连接

connection =pika.BlockingConnection(pika.ConnectionParameters(

host='127.0.0.1'))###创建一个频道

channel =connection.channel()###创建一个exchange, 类型是direct,

channel.exchange_declare(exchange='direct_logs',

type='direct')###随机创建一个队列

result = channel.queue_declare(exclusive=True)

queue_name=result.method.queue###关联关键字,只要生产者发布的消息关联了以下关键字,订阅者便能在绑定的队列中收到消息

severities = ["error"]###绑定关键字,队列到exchange.

for severity inseverities:

channel.queue_bind(exchange='direct_logs',

queue=queue_name,

routing_key=severity)print('[*] Waiting for logs. To exit press CTRL+C')###打印订阅到的消息

defcallback(ch, method, properties, body):print("[x] %r:%r" %(method.routing_key, body))###循环监听队列

channel.basic_consume(callback,

queue=queue_name,

no_ack=True)

channel.start_consuming()

执行结果：

1.首先执行消费者1和消费者2的代码

2.然后执行生产者的代码

3.可以看到消费者1、2都接收到了生产者的消息。

4.假如生产者发送带有非"error"关键字的消息，则只有消费者1才能收到。

消费者1执行结果：

[*] Waiting for logs. To exit press CTRL+C

[x]'error':b'123'[x]'info':b'123'

消费者2执行结果：

[*] Waiting for logs. To exit press CTRL+C

[x]'error':b'123'

3.Topic exchange

Topic exchange是最灵活的exchange，它会把exchange的routing key与绑定队列的routing key进行模式匹配。Routing key中可以包含和#两种符号，#号可以用来匹配一个或者多个单词，*用来匹配正好一个单词。

生产者代码：

importpika###第一件事，我们要做的就是与rabbitmq-server建立连接

connection =pika.BlockingConnection(pika.ConnectionParameters(

host='127.0.0.1'))

channel=connection.channel()###创建一个exchange, 类型是topic

channel.exchange_declare(exchange='topic_logs',

type='topic')###发布者绑定exchange的关键字,订阅者根据模糊匹配来订阅

routing_key = "lk.haha.python"message= "xxoo"channel.basic_publish(exchange='topic_logs',

routing_key=routing_key,

body=message)print("[x] Sent %r:%r" %(routing_key, message))

connection.close()

消费者代码：

importpika###第一件事，我们要做的就是与rabbitmq-server建立连接

connection =pika.BlockingConnection(pika.ConnectionParameters(

host='127.0.0.1'))

channel=connection.channel()###创建一个exchange, 类型是topic

channel.exchange_declare(exchange='topic_logs',

type='topic')###随机创建一个队列

result = channel.queue_declare(exclusive=True)

queue_name=result.method.queue###订阅者绑定在exchange以及队列的关键字模糊匹配,这里#代表0个或多个单词,* 代表一个单词,假如只写一个#,代表全部匹配.

binding_keys = ["lk.#"]###根据多种匹配来绑定

for binding_key inbinding_keys:

channel.queue_bind(exchange='topic_logs',

queue=queue_name,

routing_key=binding_key)print('[*] Waiting for logs. To exit press CTRL+C')defcallback(ch, method, properties, body):print("[x] %r:%r" %(method.routing_key, body))

channel.basic_consume(callback,

queue=queue_name,

no_ack=True)

channel.start_consuming()

执行结果：

1.先执行消费者代码，后执行生产者代码。

2.因为生产者发送的消息带有关键字lk.haha.python，符合订阅者的绑定逻辑lk.#，所以这个消费者能收到消息。

3.这个用法很方便，可以通过匹配来进行消息的选择接收。

4.Python的SQLAchemy框架

1.MySQL 基础 ----> 请自行百度

2.SQLAchemy

1.pymysql

importpymysql###创建一个MySQL 连接对象

conn = pymysql.connect(host='127.0.0.1', port=3306, user='root', passwd='123', db='s13')###创建一个可以操作MySQL的游标,默认获取结果是元组,当设置cursor=pymysql.cursors.DictCursor,后获取结果为字典

cursor =conn.cursor()#cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)

###执行SQL语句

cursor.execute("select * from t10")#获取第一行数据#row_1 = cursor.fetchone()

#获取前n行数据#row_2 = cursor.fetchmany(3)#获取所有数据

row_3 =cursor.fetchall()print(row_3)###mode='relative',获取结果相对位置移动 mode= 'absolute',获取结果绝对位置移动#cursor.scroll(-2,mode='relative')#row_3 = cursor.fetchall()#print(row_3)

###提交操作,当执行如insert update alter delete drop 等操作后要提交才能生效

conn.commit()###关闭游标

cursor.close()###关闭数据库连接

conn.close()

执行结果：就是数据库的操作结果，不过，库和表都是事先从终端创建好的。

2.SQLAchemy基本操作

from sqlalchemy.ext.declarative importdeclarative_basefrom sqlalchemy importColumn, Integer, String, ForeignKey, UniqueConstraint, Indexfrom sqlalchemy.orm importsessionmaker, relationshipfrom sqlalchemy importcreate_engine###创建一个数据库连接,连接池为5个

engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s13", max_overflow=5)###创建对象的基类,默认就这么写.

Base =declarative_base()#定义个User子类

classUsers(Base):###要创建的表名

__tablename__ = 'users'

###表的结构

id = Column(Integer, primary_key=True)

name= Column(String(32))

extra= Column(String(16))__table_args__ =(

UniqueConstraint('id', 'name', name='uix_id_name'),

Index('ix_id_name', 'name', 'extra'),

)#一对多

classFavor(Base):__tablename__ = 'favor'nid= Column(Integer, primary_key=True)

caption= Column(String(50), default='red', unique=True)classPerson(Base):__tablename__ = 'person'nid= Column(Integer, primary_key=True)

name= Column(String(32), index=True, nullable=True)

favor_id= Column(Integer, ForeignKey("favor.nid"))#多对多

classServerToGroup(Base):__tablename__ = 'servertogroup'nid= Column(Integer, primary_key=True, autoincrement=True)

server_id= Column(Integer, ForeignKey('server.id'))

group_id= Column(Integer, ForeignKey('group.id'))classGroup(Base):__tablename__ = 'group'id= Column(Integer, primary_key=True)

name= Column(String(64), unique=True, nullable=False)classServer(Base):__tablename__ = 'server'id= Column(Integer, primary_key=True, autoincrement=True)

hostname= Column(String(64), unique=True, nullable=False)

port= Column(Integer, default=22)###执行建表操作(create_all),删表操作(drop_all)

Base.metadata.create_all(engine)#Base.metadata.drop_all(engine)

执行结果如下：

mysql>show tables;+---------------+

| Tables_in_s13 |

+---------------+

| favor |

| group |

| person |

| server |

| servertogroup |

| t10 |

| users |

+---------------+

7 rows in set (0.01 sec)

3.SQLAchemy 增删改查

1.利用数据库连接直接SQL语句执行

fromsqlalchemy import create_engine

###创建一个数据库连接,连接池为5个

engine= create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s13", max_overflow=5)

engine.execute(

"INSERT INTO users (id, name, extra) VALUES (1, 'lk', 'haha')"

)

result= engine.execute('select * from users')print(result.fetchall())

执行结果：

[(1, 'lk', 'haha')]

2.利用SQLAchemy内部组件操作

在利用SQLAchemy的子类继承模式创建表之后，创建对象，利用对象执行特定语句。

增：

obj = Users(name="alex0", extra='sb')

session.add(obj)

session.add_all([Users(name="alex1", extra='sb'),

Users(name="alex2", extra='sb'),])

session.commit()

删：

session.query(Users).filter(Users.id > 2).delete()

session.commit()

改：

session.query(Users).filter(Users.id > 2).update({"name" : "099"})

session.query(Users).filter(Users.id> 2).update({Users.name: Users.name + "099"}, synchronize_session=False)

session.query(Users).filter(Users.id> 2).update({"num": Users.num + 1}, synchronize_session="evaluate")

session.commit()

查：

ret = session.query(Users).all()

ret= session.query(Users.name, Users.extra).all()

ret= session.query(Users).filter_by(name='alex').all()

ret= session.query(Users).filter_by(name='alex').first()

其他：

#　条件

ret= session.query(Users).filter_by(name='alex').all()

ret= session.query(Users).filter(Users.id > 1, Users.name == 'eric').all()

ret= session.query(Users).filter(Users.id.between(1, 3), Users.name == 'eric').all()

ret= session.query(Users).filter(Users.id.in_([1,3,4])).all()

ret= session.query(Users).filter(~Users.id.in_([1,3,4])).all()

ret= session.query(Users).filter(Users.id.in_(session.query(Users.id).filter_by(name='eric'))).all()fromsqlalchemy import and_, or_

ret= session.query(Users).filter(and_(Users.id > 3, Users.name == 'eric')).all()

ret= session.query(Users).filter(or_(Users.id < 2, Users.name == 'eric')).all()

ret=session.query(Users).filter(

or_(

Users.id< 2,

and_(Users.name== 'eric', Users.id > 3),

Users.extra!=""

)).all()

# 通配符

ret= session.query(Users).filter(Users.name.like('e%')).all()

ret= session.query(Users).filter(~Users.name.like('e%')).all()

# 限制

ret= session.query(Users)[1:2]# 排序

ret= session.query(Users).order_by(Users.name.desc()).all()

ret= session.query(Users).order_by(Users.name.desc(), Users.id.asc()).all()

# 分组fromsqlalchemy.sql import func

ret= session.query(Users).group_by(Users.extra).all()

ret=session.query(

func.max(Users.id),

func.sum(Users.id),

func.min(Users.id)).group_by(Users.name).all()

ret=session.query(

func.max(Users.id),

func.sum(Users.id),

func.min(Users.id)).group_by(Users.name).having(func.min(Users.id) >2).all()

# 连表

ret= session.query(Users, Favor).filter(Users.id == Favor.nid).all()

ret= session.query(Person).join(Favor).all()

ret= session.query(Person).join(Favor, isouter=True).all()

# 组合

q1= session.query(Users.name).filter(Users.id > 2)

q2= session.query(Favor.caption).filter(Favor.nid < 2)

ret= q1.union(q2).all()

q1= session.query(Users.name).filter(Users.id > 2)

q2= session.query(Favor.caption).filter(Favor.nid < 2)

ret= q1.union_all(q2).all()

python123平台作业答案第十二周_python周报第十二周相关推荐

python123平台作业答案第十二周_【2018年网鼎杯CTF 第二场】红日安全-网鼎杯WriteUp（24日更新：web详解）...
本次比赛主要由红日安全ctf小组奋力拼搏,才可以拿到第二场第四的成绩.感谢他们的付出,才可以让我们看到精彩的wp 1.签到题 2.虚幻题目提示汉信码.使用 binwalk 提取出 9 张图,拼接成如 ...
python123平台作业答案第十一周_马哥2016全新Linux+Python高端运维班第十次作业
系统的INPUT和OUTPUT默认策略为DROP,请完成以下关于iptables的题目:iptables -A INPUT -d 10.18.11.13 -p tcp --dport 22 -j AC ...
python123平台作业答案第十一周_第十一周作业
1.显示统计占用系统内存最多的进程,并排序. [root@centos8 ~]# top 参数 M 按内存大小排列 1.jpg 2.编写脚本,使用for和while分别实现192.168.0.0/24 ...
python123平台作业答案循环结构棋盘放米_python练习集100题(1-20)
题目1:有1.2.3.4四个数,能组成多少个互不相同且无重复的三位数?都是多少?list_num=[1,2,3,4] all_num=[] for i in list_num: for j in li ...
python123平台作业答案进制转换_各种进制转换详解-python
(1)各个进制的符号:b:二进制:o:八进制:d:十进制:x:十六进制在python中,bin(),oct(),hex()返回值均为字符串而且会带有0b,0o,0o前缀 (2)各个进制相互转换 a) ...
北京电大c语言实验作业二,安徽大学C语言实验平台作业答案
实验三 C基本数据类型及运算 1. 编程:根据需要定义相关变量,键盘输入10.0.20.6.5.0三个实数分别赋给相关三个变量,输出该3个变量的和s.乘积p和平均值a. 2. 编程:输入球的半径,计算 ...
python123测验9答案_python123 测验9: Python计算生态纵览 (第9周)
1.‪‬‪‬‪‬‪‬‪‬‮‬‪‬‫‬‪‬‪‬‪‬‪‬‪‬‮‬‫‬‪‬‪‬‪‬‪‬‪‬‪‬‮‬‪‬‮‬‪‬‪‬‪‬‪‬‪‬‮‬‫‬‭‬‪‬‪‬‪‬‪‬‪‬‮‬‪‬‫‬# 以下选项不是Python数据分析 ...
python123程序作业答案说句心里话_作业 -- 几道简单的Python题
1 #!/usr/bin/python 2 #-*- coding: utf-8 -*- 3 4 importrandom5 importsys6 importos7 8 version_error_ ...
python十大语言_Python语言的十大神器
本文主要向大家介绍了Python语言的十大神器,通过具体的内容向大家展示,希望对大家学习Python语言有所帮助. 相传在上古时期,混沌未分,鸿蒙初现,天地间产生十件威力无比的神器,分别是:东皇钟.轩 ...

python123平台作业答案第十二周_python周报第十二周

python123平台作业答案第十二周_python周报第十二周相关推荐

最新文章

热门文章