python学习笔记-Day12 (上下文管理、redis发布订阅、rabbitmq、pymysql模块、SQLAchemy)

上下文管理

import contextlib

# 上下文管理

@contextlib.contextmanager
def worker_state(state_list, worker_thread):
    """
    :param state_list:
    :param worker_thread:
    :return:
    """
    state_list.append(worker_thread) # 2. 进入执行函数体
    try:
        yield                       # 3. 遇到yield，弹出函数执行with代码块
    finally:
        state_list.remove(worker_thread)  # 5. 执行，然后结束

free_list = []
current_thread = ‘alex‘
with worker_state(free_list, current_thread): # 1. 首先进入with， 进入worker_state函数
    print(123)                          # 4. 执行此代码后，回到函数
# 上下文管理，实现自动关闭socket

import contextlib
import socket

@contextlib.contextmanager
def context_socket(host,port):
    sk = socket.socket()   # 2. 执行函数体内容
    sk.bind((host,port))
    sk.listen(5)
    try:
        yield sk          # 3. yield 返回with代码块执行
    finally:
        sk.close()         # 5. 继续执行，结束

with context_socket(‘127.0.0.1‘, 8888) as sock:   # 1. 执行context_socket函数
    print(sock)            # 4. 执行with代码块后，返回函数

redis的发布与订阅

# 发布

import redis

class RedisHelper:

    def __init__(self):
        self.__conn = redis.Redis(host=‘192.168.11.87‘)

    def public(self,msg,chan):
        self.__conn.publish(chan,msg)   # 调用发布函数，向订阅者发布
        return True

    def subscribe(self,chan):
        pub = self.__conn.pubsub()
        pub.subscribe(chan)
        pub.parse_response()
        return pub

if __name__ == ‘__main__‘:
    obj = RedisHelper()
    obj.public(‘this is test‘, ‘fm8888.7‘)
# 订阅
import publish # 导入为上代码的类

obj = publish.RedisHelper()
data = obj.subscribe(‘fm8888.7‘)
print(data.parse_response())

rabbitmq基本使用

############ 基础使用

# 生产者
import pika

# 创建一个连接
connection = pika.BlockingConnection(pika.ConnectionParameters(host=‘192.168.31.98‘))

# 创建一个频道
channel = connection.channel()

# 定义一个队列
channel.queue_declare(queue=‘rock‘)

channel.basic_publish(exchange=‘‘,
                      routing_key=‘rock‘,   # 往这个队列发消息
                      body=‘hello world!‘)
print(" [x] Sent ‘hello world‘")
connection.close()

print(‘ [*] Waiting for messages. To exit press CTRL+c ‘)
channel.start_consuming()

=================================================================================================================

############ 消费者
import pika

# 创建连接
connection = pika.BlockingConnection(pika.ConnectionParameters(host=‘192.168.31.98‘))

# 创建频道
channel = connection.channel()

# 定义一个队列
channel.queue_declare(queue=‘rock‘)

def callback(ch, method, properties,body):  # ch: channel   method:  队列名字   properties:  连接上rabbiatmq基本属性   body: 取出队列的内容
    print(‘[x] Received %r‘ % body)

channel.basic_consume(callback,queue=‘rock‘,no_ack=True)  # 当我在队列rock中取到数据时， 我就会执行callback函数，并且我还会给callback函数传入4个参数

print(‘ [*] Waiting for messages. To exit press CTRL+c ‘)
channel.start_consuming()

rabbitmq: 消息不丢失

######## 生产者
import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(host=‘192.168.11.87‘))

channel = connection.channel()

channel.exchange_declare(exchange=‘rock_now‘,
                         type=‘direct‘)

severity = ‘info‘
message = ‘123‘
channel.basic_publish(exchange=‘rock_now‘,
                      routing_key=severity,
                      body=message,
                      propperties=pika.BasicProperties(
                          delivery_mode=2,  # 生产者设置持久化，保证生产者发送队列时，宕机不会丢失
                      ) )
print(" [x] Sent %r:%r" % (severity,message))
connection.close()

========================================

######## 消费者
import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(host=‘192.168.11.87‘))

channel = connection.channel()

channel.exchange_declare(exchange=‘rock_now‘,
                         type=‘direct‘) # 指定关键字，接收有此关键字的队列
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
severities = [‘error‘, ‘info‘, ‘warning‘]
for severity in severities:
    channel.queue_bind(exchange=‘rock_now‘,
                       queue=queue_name,
                       routing_key=severity)
print(‘ [*] Waiting for messages. To exit press CTRL+c ‘)

def callback(ch, method, properties,body):
    print(‘[x] Received %r‘ % body)

channel.basic_consume(callback,
                      queue=queue_name,
                      no_ack=False)   # 消费者在接受队里消息后，发送ack确认接收

channel.start_consuming()

rabbitmq: 获取顺序

# 消费者
# 因为是获取，只针对消费者

import pika

# 当有多个消费者，共同取数据时，默认python是“按照顺序取”
# 也就是，如果有三个消费者，去共同取队列数据，
# 第一个消费者，取得是1,4,7,10...， 第二个消费者，取得是2,5,8,11...， 第三个消费者，取的是3,6,9,12...
# 虽然，消费者之间并不会因为顺序而阻塞， 但是各个消费者还是会按照他们的先后顺序取跳着取数据
# 下边，我们修改这种默认的配置

connection = pika.BlockingConnection(pika.ConnectionParameters(host=‘192.168.31.98‘))
channel = connection.channel()

# make message persistent
channel.queue_declare(queue=‘hello‘)

def callback(ch, method, properties, body):
    print(" [x] Received %r" % body)
    import time
    time.sleep(10)
    print(‘ok‘)
    ch.basic_ack(delivery_tag = method.delivery_tag)

channel.basic_qos(prefetch_count=1)   # 此设置为多个消费者，不跳着取值

channel.basic_consume(callback,
                      queue=‘hello‘,
                      no_ack=False)

print(‘ [*] Waiting for messages. To exit press CTRL+C‘)
channel.start_consuming()

rabbitmq: fanout类型exchange

######### 生产者
import pika

# 创建连接
connection = pika.BlockingConnection(pika.ConnectionParameters(host=‘192.168.31.98‘))

# 创建频道
channel = connection.channel()

# 创建exchang, fanout 类型的exchange发送或接受所有在此exchange中的队列
channel.exchange_declare(exchange=‘rock‘, type=‘fanout‘)

message = ‘rock now‘
channel.basic_publish(exchange=‘rock‘,  # 往这个exchange组里所有队列发送
                      routing_key=‘‘,
                      body=message)
print(" [x] Sent %r" % message)
connection.close()

========================================

######## 消费者
import pika

# 创建连接
connection = pika.BlockingConnection(pika.ConnectionParameters(host=‘192.168.11.87‘))

# 创建频道
channel = connection.channel()

# 创建exchang, fanout 类型的exchange发送或接受所有在此exchange中的队列
channel.exchange_declare(exchange=‘rock‘, type=‘fanout‘)

# 随机创建队列
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue

# 绑定
# 这里当我们运行多次此程序时，我们随机创建的队列肯定不一致，而我们绑定的是一个exchange，
# 所以，这里我们运行多次此程序，然后运行一次生产者，即可
channel.queue_bind(exchange=‘rock‘, queue=queue_name)

print(‘ [*] Waiting for messages. To exit press CTRL+c ‘)

def callback(ch,method,properties,body):
    print(‘ [x] %r‘ % body )

channel.basic_consume(callback, queue=queue_name, no_ack=True)

channel.start_consuming()

rabbitmq: direct类型exchange

######## 消费者
import pika

# 创建连接
connection = pika.BlockingConnection(pika.ConnectionParameters(host=‘192.168.11.87‘))

# 创建频道
channel = connection.channel()

# 创建exchange， direct类型的exchange可使程序根据关键字的信息发送或接收相应队列的内容
channel.exchange_declare(exchange=‘rock_now‘,
                         type=‘direct‘) # 指定关键字，接收有此关键字的队列

# 创建随机队列
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue

severities = [‘error‘, ‘info‘, ‘warning‘]
for severity in severities:
    channel.queue_bind(exchange=‘rock_now‘,
                       queue=queue_name,
                       routing_key=severity)  # 指定关键字，找对应关键字队列接收
print(‘ [*] Waiting for messages. To exit press CTRL+c ‘)

def callback(ch, method, properties,body):
    print(‘[x] Received %r‘ % body)

channel.basic_consume(callback,queue=queue_name,no_ack=True)

channel.start_consuming()

========================================

######## 生产者
import pika

# 创建连接
connection = pika.BlockingConnection(pika.ConnectionParameters(host=‘192.168.11.87‘))

# 创建频道
channel = connection.channel()

# 绑定exchang
channel.exchange_declare(exchange=‘rock_now‘,
                         type=‘direct‘)

severity = ‘info‘
message = ‘123‘
channel.basic_publish(exchange=‘rock_now‘,
                      routing_key=severity, # 指定关键字，发送到关键字对应的队列
                      body=message)
print(" [x] Sent %r:%r" % (severity,message))
connection.close()

rabbitmq: topic类型exchange

exchange type = topic

在topic类型下，可以让队列绑定几个模糊的关键字，之后发送者将数据发送到exchange，exchange将传入”路由值“和 ”关键字“进行匹配，匹配成功，则将数据发送到指定队列。

  • # 表示可以匹配 0 个 或 多个 单词
  • *  表示只能匹配 一个 单词
发送者路由值              队列中
old.boy.python          old.*  -- 不匹配
old.boy.python          old.#  -- 匹配

在使用上，参考关键字匹配，只不过是传入值带*或#，topic会模糊匹配，而关键字不会~

pymysql 模块

#基础操作

import pymysql

# 创建连接
conn = pymysql.connect(host=‘192.168.31.98‘, port=3306, user=‘root‘, passwd=‘123qwe‘, db=‘test‘)

# 创建游标
cursor = conn.cursor()

# 执行SQL，并返回收影响行数
effect_row = cursor.execute("update hosts set host = ‘1.1.1.2‘")

# 执行SQL，并返回受影响行数
effect_row = cursor.execute("update hosts set host = ‘1.1.1.2‘ where nid > %s", (1,))  # 支持字符串拼接

# 执行SQL，并返回受影响行数
effect_row = cursor.executemany("insert into hosts(host,color_id)values(%s,%s)", [("1.1.1.11",1),("1.1.1.11",2)])   # 支持多个字符串拼接

# 提交，不然无法保存新建或者修改的数据
conn.commit()

# 关闭游标
cursor.close()
# 关闭连接
conn.close()
# 自增ID

import pymysql

conn = pymysql.connect(host=‘127.0.0.1‘, port=3306, user=‘root‘, passwd=‘123‘, db=‘t1‘)
cursor = conn.cursor()
cursor.executemany("insert into hosts(host,color_id)values(%s,%s)", [("1.1.1.11",1),("1.1.1.11",2)])
conn.commit()
cursor.close()
conn.close()

# 获取最新自增ID
new_id = cursor.lastrowid
# 在上代码中，我们使用executemany，添加多条后，获取最新的自增ID为最后一个
# 获取查询数据 

import pymysql

conn = pymysql.connect(host=‘127.0.0.1‘, port=3306, user=‘root‘, passwd=‘123‘, db=‘t1‘)
cursor = conn.cursor()
cursor.execute("select * from hosts")

# 获取第一行数据
row_1 = cursor.fetchone()

# 获取前n行数据，设置为3，则获取3行数据
# row_2 = cursor.fetchmany(3)
# 获取所有数据
# row_3 = cursor.fetchall()

conn.commit()
cursor.close()
conn.close()
# 移动游标
-----------------------------------------------------------------
cursor.scroll(1,mode=‘relative‘)  # 相对当前位置移动
cursor.scroll(2,mode=‘absolute‘) # 相对绝对位置移动
-----------------------------------------------------------------

import pymysql

conn = pymysql.connect(host=‘127.0.0.1‘, port=3306, user=‘root‘, passwd=‘123‘, db=‘t1‘)
cursor = conn.cursor()
cursor.execute("select * from hosts")

# 获取所有数据
row_3 = cursor.fetchall()

# 获取所有数据后，我将游标上移2位
cursor.scroll(-2,mode=‘relative‘)

# 重新获取所有数据，就从游标处开始获取所有
row_3 = cursor.fetchall()

conn.commit()
cursor.close()
conn.close()
# fetch 数据类型

import pymysql

conn = pymysql.connect(host=‘127.0.0.1‘, port=3306, user=‘root‘, passwd=‘123‘, db=‘t1‘)

# 游标设置为字典类型
cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)
r = cursor.execute("call p1()")

result = cursor.fetchone()

conn.commit()
cursor.close()
conn.close()

SQLAchemy （ORM）使用：创建与删除

######### 创建表

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine

# 创建连接
engine = create_engine("mysql+pymysql://root:[email protected]:3306/test", max_overflow=5)

Base = declarative_base()  # 规定这么写

# 创建单表
class Users(Base):   # 所有你自定义的类必须继承你刚刚创建的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‘),
    )

Base.metadata.create_all(engine)  # 这里，Base会找到所有它的子类，根据子类执行创建表
######### 一对多创建表

# 一对多
class Favor(Base):
    __tablename__ = ‘favor‘  # 表名定义
    nid = Column(Integer, primary_key=True)  # 自增ID
    caption = Column(String(50), default=‘red‘, unique=True) # 字符串最多50，unique=True不允许重复，default默认值red

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"))   # ForeignKey("favor.nid"),与创建favor表的nid列做一个外键

Base.metadata.create_all(engine)
######## 多对多建表

# 多对多
class ServerToGroup(Base):  # 第三张表存下两个表的关系
    __tablename__ = ‘servertogroup‘
    nid = Column(Integer, primary_key=True, autoincrement=True)
    # 这里同时外键两个表的ID列，这样建立了下两个表（类）的关系
    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)

Base.metadata.create_all(engine)
# 创建与删除

Base.metadata.create_all(engine)   # 创建所有表
Base.metadata.drop_all(engine)     # 删除所有表

SQLAchemy （ORM）使用：操作表

# 增
engine = create_engine("mysql+pymysql://root:[email protected]:3306/test", max_overflow=5)

Base = declarative_base()  # 规定这么写

class Users(Base):   # 所有你自定义的类必须继承你刚刚创建的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‘),
    )

obj = Users(name="alex0", extra=‘sb‘)  # 需要在哪个表增加数据，就将哪个表的类封装一个对象
seeion.add(obj)  # 添加

seeion.commit() # 提交
时间： 2024-09-26 22:31:15

python学习笔记-Day12 (上下文管理、redis发布订阅、rabbitmq、pymysql模块、SQLAchemy)的相关文章

Redis学习笔记07Redis命令之(6) 发布订阅

1.1.1. subscribe 订阅一个或多个频道.当所订阅的频道上发布了消息时,本客户端连接将收到这个消息. redis.coe2coe.me:6379> subscribe abc Reading messages... (press Ctrl-C to quit) 1) "subscribe" 2) "abc" 3) (integer) 1 成功订阅了一个频道. 也可以一次订阅多个频道: redis.coe2coe.me:6379> subscr

Python学习笔记——基础篇【第六周】——Subprocess模块

执行系统命令 可以执行shell命令的相关模块和函数有: os.system os.spawn* os.popen*          --废弃 popen2.*           --废弃 commands.*      --废弃,3.x中被移除 1 import commands 2 3 result = commands.getoutput('cmd') 4 result = commands.getstatus('cmd') 5 result = commands.getstatuso

Python学习笔记——基础篇【第六周】——hashlib模块

常用模块之hashlib模块 用于加密相关的操作,3.x里代替了md5模块和sha模块,主要提供 SHA1, SHA224, SHA256, SHA384, SHA512 ,MD5 算法 import md5 hash = md5.new() hash.update('admin') print hash.hexdigest() MD5-废弃 import sha hash = sha.new() hash.update('admin') print hash.hexdigest() sha-废

Python学习笔记——基础篇【第六周】——logging模块

常用模块之logging 用于便捷记录日志且线程安全的模块 import logging logging.basicConfig(filename='log.log', format='%(asctime)s - %(name)s - %(levelname)s -%(module)s: %(message)s', datefmt='%Y-%m-%d %H:%M:%S %p', level=10) logging.debug('debug') logging.info('info') loggi

Python学习笔记——基础篇【第五周】——os模块和sys模块

OS模块 提供对操作系统进行调用的接口 os.getcwd() 获取当前工作目录,即当前python脚本工作的目录路径 os.chdir("dirname") 改变当前脚本工作目录:相当于shell下cd os.curdir 返回当前目录: ('.') os.pardir 获取当前目录的父目录字符串名:('..') os.makedirs('dirname1/dirname2') 可生成多层递归目录 os.removedirs('dirname1') 若目录为空,则删除,并递归到上一级

Python学习笔记——基础篇【第五周】——常用模块学习

模块介绍 本节大纲: 模块介绍 time &datetime模块   (时间模块) random   (随机数模块) os   (系统交互模块) sys shutil   (文件拷贝模块) json & picle shelve   (封装模块) xml处理 yaml处理 configparser hashlib subprocess logging模块 模块,用一砣代码实现了某个功能的代码集合. 类似于函数式编程和面向过程编程,函数式编程则完成一个功能,其他代码用来调用即可,提供了代码的

Python学习笔记（三十二）常用内建模块（1）— datetime

>>> from datetime import datetime >>> now = datetime.now() # 获取当前datetime >>> print(now) 2015-05-18 16:28:07.198690 >>> print(type(now)) <class 'datetime.datetime'> 获取指定日期和时间 要指定某个日期和时间,我们直接用参数构造一个datetime: >&g

python学习笔记-Day5（反射、特殊方法、OS模块）

反射 # 反射(或叫自省): 有时候我们会碰到这样的需求,需要执行对象的某个方法,或是需要对对象的某个字段赋值,而方法名或是字段名在编码代码时并不能确定,需要通过参数传递字符串的形式输入 # 这就是我们反射功能: 通过字符串形式去调用对象 # 例, 我有一个导向页面commons.py def login(): print('登录') def home(): print('HOME PAGE') def logout(): print('logout !!!') # 我需要根据用户输入,去执行对

Python学习笔记——基础篇【第六周】——shutil模块

常用模块之shutil 高级的 文件.文件夹.压缩包 处理模块 shutil.copyfileobj(fsrc, fdst[, length]) 将文件内容拷贝到另一个文件中,可以部分内容 1 def copyfileobj(fsrc, fdst, length=16*1024): 2 """copy data from file-like object fsrc to file-like object fdst""" 3 while 1: 4