0.本周知识点预览

• Contextlib
• Redis发布订阅
• RabbitMQ
• pymysql
• SQLAchemy

 

1.Contextlib模块

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

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

###contextlib (实现with上下文管理)import contextliblist1 = [1,2,3]str1 = "lk" ###此处必须是这个装饰器@contextlib.contextmanagerdef func(l, s):    l.append(s)    try:        ###执行到yield时,中断跳出函数        yield    finally:        print(l)with func(list1, str1):    print(123)    print(456)

执行结果如下：

123456[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,自动关闭连接import contextlibimport socket@contextlib.contextmanagerdef base_socket(host, port):    sk = socket.socket()    sk.bind((host, port))    sk.listen(5)    print(123)    try:        yield sk    finally:        print(789)        sk.close()with base_socket("127.0.0.1", 8888) as sock:    print(456)

执行结果如下：

123456789

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

2.Redis 发布订阅

自定义redis基础类：

## redis 发布订阅import redisclass RedisHelper:    def __init__(self):        ###创建redis连接对象        self.__conn = redis.Redis(host="127.0.0.1", port=6379)    def public(self, msg, chan):        ###publish 方法,把信息发布到频道上,返回消息被传递的订阅者的数量        self.__conn.publish(chan, msg)        return True    def subscribe(self, chan):        ###pubsub 方法的意思是,返回一个发布或者订阅的对象,用这个对象,你就能订阅这个频道,监听给发给这些频道的消息        pub = self.__conn.pubsub()        ###subscribe 方法: 订阅频道        pub.subscribe(chan)        ###parse_response 方法:解析从发布者/订阅者命令的响应        pub.parse_response()        return pub

发布者代码：

import testfabu = test.RedisHelper()while True:    inp = input(">>> ")    if inp == "exit":        break    else:        fabu.public("%s" % inp, "998")

订阅者代码：

import testdingyue = test.RedisHelper()while True:    ###subscribe 方法 -> 订阅频道    data = dingyue.subscribe("998")    ###parse_response 方法:解析从发布者/订阅者命令的响应    print(data.parse_response())

执行结果：

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

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

3.RabbitMQ

1.未利用exchange

生产者代码：

import pika###第一件事，我们要做的就是与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()

消费者代码：

import pika###第一件事，我们要做的就是与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函数就是将接收到的消息打印在屏幕上def callback(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'] is received[b'fuck'] is received[b'fuck'] is received[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不起作用。

生产者代码：

import pika###第一件事，我们要做的就是与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 版import pika###第一件事，我们要做的就是与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")##绑定队列到exchangechannel.queue_bind(exchange="lk", queue="hello_lk5")###函数名不必须叫callback,callback函数就是将接收到的消息打印在屏幕上def callback1(ch, method, propreties, body):    print("[x] 收到 %s" % body)channel.basic_consume(callback1, queue="hello_lk5", no_ack=True)channel.start_consuming()

消费者2代码：

##exchange 版import pika###第一件事，我们要做的就是与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_lk6")##绑定队列到exchangechannel.queue_bind(exchange="lk", queue="hello_lk6")###函数名必须叫callback,callback函数就是将接收到的消息打印在屏幕上def callback1(ch, method, propreties, body):    print("[x] 收到 %s" % body)channel.basic_consume(callback1, queue="hello_lk6", no_ack=True)channel.start_consuming()

执行结果：

生产者：

send MSG: helloProcess 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，该队列才能接收到消息（可按上述步骤进行验证）。这种情况下，队列之间是互斥关系，一个消息最多只能进入一个队列。

生产者代码：

import pika###第一件事，我们要做的就是与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代码：

 

import pika# 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 in severities:    channel.queue_bind(exchange='direct_logs',                       queue=queue_name,                       routing_key=severity)print(' [*] Waiting for logs. To exit press CTRL+C')###打印订阅到的消息def callback(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代码：

import pika# 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 in severities:    channel.queue_bind(exchange='direct_logs',                       queue=queue_name,                       routing_key=severity)print(' [*] Waiting for logs. To exit press CTRL+C')###打印订阅到的消息def callback(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中可以包含 和#两种符号，#号可以用来匹配一个或者多个单词，*用来匹配正好一个单词。

生产者代码：

import pika###第一件事，我们要做的就是与rabbitmq-server建立连接connection = pika.BlockingConnection(pika.ConnectionParameters(        host='127.0.0.1'))channel = connection.channel()###创建一个exchange, 类型是topicchannel.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()

消费者代码：

import pika###第一件事，我们要做的就是与rabbitmq-server建立连接connection = pika.BlockingConnection(pika.ConnectionParameters(        host='127.0.0.1'))channel = connection.channel()###创建一个exchange, 类型是topicchannel.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 in binding_keys:    channel.queue_bind(exchange='topic_logs',                       queue=queue_name,                       routing_key=binding_key)print(' [*] Waiting for logs. To exit press CTRL+C')def callback(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

import pymysql###创建一个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 import declarative_basefrom sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Indexfrom sqlalchemy.orm import sessionmaker, relationshipfrom sqlalchemy import create_engine###创建一个数据库连接,连接池为5个engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s13", max_overflow=5)###创建对象的基类,默认就这么写.Base = declarative_base()# 定义个User子类class Users(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'),    )# 一对多class Favor(Base):    __tablename__ = 'favor'    nid = Column(Integer, primary_key=True)    caption = Column(String(50), default='red', unique=True)class Person(Base):    __tablename__ = 'person'    nid = Column(Integer, primary_key=True)    name = Column(String(32), index=True, nullable=True)    favor_id = Column(Integer, ForeignKey("favor.nid"))# 多对多class ServerToGroup(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'))class Group(Base):    __tablename__ = 'group'    id = Column(Integer, primary_key=True)    name = Column(String(64), unique=True, nullable=False)class Server(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语句执行

from sqlalchemy 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()from sqlalchemy 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()# 分组from sqlalchemy.sql import funcret = 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()