回廊识路

Python常用第三方库整理

2/22/2019 Python

本科实习期间写过一段时间python，这里整理了一些常用的第三方库。

# Celery

  • 基于python开发的分布式异步消息任务队列

celery-worker

# Quick Start

from celery import Celery

# 配置backend和broker
app = Celery(
    'tasks',
    backend='redis://localhost:6379/0',
    broker='redis://localhost:6379/0'
)

@app.task
def add(x, y):
    return x + y
1
2
3
4
5
6
7
8
9
10
11
12
celery -A tasks worker --loglevel=info
1
>>> from tasks import add
>>> add.delay(6,6)
<AsyncResult: 8de16f41-3c6c-43e7-9c96-e5273f559aa8>
1
2
3

# Json

方法 功能
json.dumps() 将dict类型的数据转成str
json.loads() 将str类型的数据转成dict
json.dump() 将dict类型的数据转成str，并写入到json文件中
json.load() 从json文件中读取数据

# Logging

# 提供接口

  • logger：提供日志接口，供应用代码使用。logger最常用的操作有两类：配置和发送日志消息。可以通过logging.getLogger(name)获取logger对象，如果不指定name则返回root对象，多次使用相同的name调用getLogger方法返回同一个logger对象。
  • handler：将日志记录（log record）发送到合适的目的地（destination），比如文件、socket等。一个logger对象可以通过addHandler方法添加0到多个handler，每个handler又可以定义不同日志级别，以实现日志分级过滤显示。
  • filter：提供一种优雅的方式决定一个日志记录是否发送到handler。
  • formatter：指定日志记录输出的具体格式。formatter的构造方法需要两个参数：消息的格式字符串和日期字符串，这两个参数都是可选的。

# 用法解析

  1. 初始化logger = logging.getLogger("endlesscode")，getLogger()方法后面最好加上所要日志记录的模块名字，后面的日志格式中的%(name)s 对应的是这里的模块名字。
  2. 设置级别logger.setLevel(logging.DEBUG),Logging中有NOTSET < DEBUG < INFO < WARNING < ERROR < CRITICAL这几种级别，日志会记录设置级别以上的日志。
  3. Handler，常用的是StreamHandler和FileHandler，windows下可以简单理解为一个是console和文件日志，一个打印在CMD窗口上，一个记录在一个文件上。
  4. formatter，定义了最终log信息的顺序,结构和内容，如： '[%(asctime)s] [%(levelname)s] %(message)s', '%Y-%m-%d %H:%M:%S'。
  5. 使用object.debug(message)来记录日志。

# Demo

import logging

class Logger:
    def __init__(self, path, clevel=logging.ERROR, flevel=logging.DEBUG):
        self.logger = logging.getLogger(path)
        self.logger.setLevel(logging.DEBUG)
        fmt = logging.Formatter('[%(asctime)s] [%(levelname)s] %(message)s', '%Y-%m-%d %H:%M:%S')
        # 设置CMD日志
        ch = logging.StreamHandler()
        ch.setFormatter(fmt)
        ch.setLevel(clevel)
        # 设置文件日志
        fh = logging.FileHandler(path)
        fh.setFormatter(fmt)
        fh.setLevel(flevel)
        # handler添加到logger
        self.logger.addHandler(ch)
        self.logger.addHandler(fh)

    def debug(self, message):
        self.logger.debug(message)

    def info(self, message):
        self.logger.info(message)

    def war(self, message):
        self.logger.warn(message)

    def error(self, message):
        self.logger.error(message)

    def cri(self, message):
        self.logger.critical(message)

LOG_FILE = '/path/to/file.log'
logger_info = Logger(LOG_FILE, logging.ERROR, logging.DEBUG)

if __name__ == '__main__':
    logger_info.debug('一个debug信息')
    logger_info.info('一个info信息')
    logger_info.war('一个warning信息')
    logger_info.error('一个error信息')
    logger_info.cri('一个致命critical信息')
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

# Netmiko

# Quick Start

from netmiko import ConnectHandler

cisco_881 = {
    'device_type': 'cisco_ios',
    'ip': '10.10.10.227',
    'username': 'username',
    'password': 'password',
    }

# 创建connect对象
>>> net_connect = ConnectHandler(**cisco_881)
SSH connection established to 10.10.10.227:22
Interactive SSH session established

# 调用find_prompt()方法确认连接状态
>>> net_connect.find_prompt()
u'pynet-rtr1#'


# send_command()发送命令
>>> output = net_connect.send_command("show ip int brief")
>>> print output
Interface           IP-Address      OK? Method Status      Protocol
FastEthernet0       unassigned      YES unset  down        down

# send_config_set()修改配置
>>> config_commands = ['logging buffered 19999']
>>> output = net_connect.send_config_set(config_commands)
>>> print output
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

# 其他方法

method function
net_connect.config_mode() Enter into config mode
net_connect.check_config_mode() Check if you are in config mode, return a boolean
net_connect.exit_config_mode() Exit config mode
net_connect.clear_buffer() Clear the output buffer on the remote device
net_connect.enable() Enter enable mode
net_connect.exit_enable_mode() Exit enable mode
net_connect.find_prompt() Return the current router prompt
net_connect.commit(arguments) Execute a commit action on Juniper and IOS-XR
net_connect.disconnect() Close the SSH connection
net_connect.send_command(arguments) Send command down the SSH channel, return output back
net_connect.send_config_set(arguments) Send a set of configuration commands to remote device
net_connect.send_config_from_file(arguments) Send a set of configuration commands loaded from a file