Harvey Li
首页 标签 归档 关于

自动化运维（二）：Python

2024-03-19 知见

常见魔法方法

考虑下面的类A，定义在test.py文件中：

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
class A:
    def __init__(self, name, data, ds):
        self.name = name
        self.data = data
        self.ds = ds

    def __str__(self):
        return "对象: {}".format(self.name)

    def __repr__(self):
        return "对象: {}".format(self.data)

    def __len__(self):
        return len(self.data)

    def __contains__(self, key):
        print('得到判断的key = %s' % key)
        return key in self.ds

    def __setitem__(self, key, val):
        print('进入__setitem__()方法: <%s, %s>' % (key, val))
        if isinstance(key, int):
            # 如果是数字，操作self.data
            if key >= len(self.data):
                self.data.append(val)
            else:
                self.data[key] = val
        else:
            # 如果是其他，操作self.ds
            self.ds[key] = val

    def __getitem__(self, key):
        print('进入__getitem__()方法: %s' % key)
        if isinstance(key, int) and key < len(self.data):
            return self.data[key]
        elif key in self.ds:
            return self.ds[key]
        return None

实例化A类得到a对象：

1
2
>>> from test import A
>>> a = A('harvey', ['abc', 'xxxx', 'xyz'], {'k1': 'v1', 'k2': 'v2', 'k3': 'v3'})

__repr__()与__str__()

1
2
3
4
5
6
>>> a
对象: ['abc', 'xxxx', 'xyz']
>>> print(A)
<class 'test.A'>
>>> print(a)
对象: harvey

__repr__()和__str__()分别改变了对象a在交互模式下以及print()函数的输出结果。在交互模式下直接输出对象a，将搜索对象中的__repr__()方法，并将该方法的返回值作为输出。如果是使用print()函数打印输出，那么Python解释器会先查找对象中的 __str__()方法，将其返回结果作为输出打印，如果没有__str__()方法，会继续查找__repr__()方法的返回值输出。如果都没有，则打印默认的输出。

__len__()

__len__()方法用于支持len(对象)操作。

1
2
>>> len(a)
3

对一个自定义的类，如果它没有 __len__()方法，就无法使用len(A对象)语句。

1
2
3
4
>>> len(a)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
TypeError: object of type 'A' has no len()

__contains__()

定义魔法方法__contains__()后，该类的对象将支持 for key in xxx 这样的语句：

1
2
3
4
5
6
>>> 'k1' in a
得到判断的key = k1
True
>>> 'kx' in a
得到判断的key = kx
False

__getitem__()与__setitem__()

让对象可以使用索引方式调用。使用表达式A对象[1] 或者 A对象['k1']时，将自动进入__getitem__()魔法方法中；当使用赋值语句 A对象['kx'] = 'xxxxx'时，将进入 __setitem__()方法处理：

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
>>> a[1]
进入__getitem__()方法: 1
'xxxx'
>>> a[2]
进入__getitem__()方法: 2
'xyz'
>>> a[3] # 超出self.data的索引范围，返回为None
进入__getitem__()方法: 3
>>> a['k1']
进入__getitem__()方法: k1
'v1'
>>> a['k2']
进入__getitem__()方法: k2
'v2'
>>> a['kx']
进入__getitem__()方法: kx
# 调用__setitem__()方法
>>> a[1] = 'hello'
进入__setitem__()方法: <1, hello>
>>> a.data
['abc', 'hello', 'xyz']
>>> a[10] = '末尾添加'
进入__setitem__()方法: <10, 末尾添加>
>>> a.data
['abc', 'hello', 'xyz', '末尾添加']
>>> a['kx'] = 'vx'
进入__setitem__()方法: <kx, vx>
>>> a.ds
{'k1': 'v1', 'k2': 'v2', 'k3': 'v3', 'kx': 'vx'}

Pymysql

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
# 安装 pip install pymysql

from pymysql.connections import Connection

# 创建连接
conn = Connection(host='192.168.26.1', port=3306, user='root', password='admin', database='hive')

# 获取游标
cursor = conn.cursor()
print('cursor object = {} '.format(cursor))
# cursor object = <pymysql.cursors.Cursor object at 0xffffab408880> 

# 执行操作（SQL）
res = cursor.execute('select * from tbls')  # Returns: Number of affected rows.
print('受影响行数:', res)  # 返回一个int
## 逐行获取
print('第1条结果：', cursor.fetchone())  # 游标前进1行
print('第2条结果：', cursor.fetchone())
print('取后5条数据：', cursor.fetchmany(5))  # 游标前进5行
print('取剩余数据：', cursor.fetchall())  # 游标移动到末尾 len(self._rows)
## 获取全部
res = cursor.execute('select * from tbls')
print('获取全部结果：{}'.format(cursor._rows))  # cursor._rows元组

# 关闭连接
cursor.close()
conn.close()

Requests

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
# urllib3的简单使用
from urllib.parse import urlparse
import urllib3

test_url = 'HTTP://192.168.26.140:80'
parsed = urlparse(test_url)
# ParseResult(scheme='http', netloc='192.168.26.140:80', path='', params='', query='', fragment='')
req_url = parsed.geturl()  # 重新解析以规范格式
# 'https://192.168.26.140:80'
pool_manager = urllib3.PoolManager()
conn = pool_manager.connection_from_url(req_url)
resp = conn.urlopen('GET', req_url)
# <urllib3.response.HTTPResponse object at 0xffff80284b50>
print(f'请求状态码={resp.status}，请求结果={resp.data}')


# request
import requests
response = requests.get(test_url)
print(f'requests请求结果: {response.status_code}, {response.text}')

文档：https://requests.readthedocs.io/projects/cn/zh-cn/latest/user/quickstart.html

getattr()典型应用

日志打印

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
import logging

def print_log(msg, level="debug"):
    level = level.lower()
    if level == "debug":
        logging.debug(msg)
    elif level == "info":
        logging.info(msg)
    elif level == "warning":
        logging.warning(msg)
    elif level == "error":
        logging.error(msg)

# 默认只打印warning以及之上级别的日志，这里修改为打印debug之上的日志
logging.basicConfig(level=logging.DEBUG)

print('开始你的打印')
print_log('DEBUG，测试')
print_log('INFO,  测试', level='info')
print_log('WARN,  测试', level='warning')
print_log('ERROR,  测试', level='error')

结果如下：

1
2
3
4
5
6
[root@center ~]# python test.py 
开始你的打印
DEBUG:root:DEBUG，测试
INFO:root:INFO,  测试
WARNING:root:WARN,  测试
ERROR:root:ERROR,  测试

以上代码非常浪费if...else，观感不好。但注意到，这里传入的level值与logging的属性名是一致的，此时就可以使用getattr()函数来回去要对应调用的日志打印函数。这样就可以大幅简化if语句：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
import logging

def print_log_new(msg, level="debug"):
    try:
        log_func = getattr(logging, level)
        log_func(msg)
    except AttributeError:
        print(msg)

logging.basicConfig(level=logging.DEBUG)


print('开始你的打印')
print_log_new('DEBUG，测试')
print_log_new('INFO,  测试', level='info')
print_log_new('WARN,  测试', level='warning')
print_log_new('ERROR,  测试', level='error')

也能取得一样的结果。

返回结果

主动设置能符合getattr()操作的语句，可以简化后续的语句。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
class Result:
    def set_success_result(self, data):
        print('设置成功结果')
    def set_failure_result(self, data):
        print('设置失败结果')

def handle(flag=False):
    if flag:
        return True, "hello"
    return False, "world"

result = Result()
ret, data = handle(False)
if ret:
    result.set_success_result(data)
else:
    result.set_failure_result(data)

处理函数handle()的传入参数为true时，表示处理正确，希望能调用Result对象的set_success_result()方法封装结果并返回（这里用一个print代替操作）；当传入参数为false时，表示处理错误，希望能调用Result对象的set_failure_result()方法封装结果并返回。后面有出现了对返回结果的判断，然后才决定调用哪个方法。

注意到，不同结果的方法名只有一点细微的差别，都叫set_xxx_result()。那么只需要handler()方法返回这个不同的值，就可以不加判断了。

1
2
3
4
5
6
7
8
def handle_new(flag=False):
    if flag:
        return "success", "hello"
    return "failure", "world"

ret, data = handle_new(True)
callback = getattr(result, f"set_{ret}_result") # 通过getattr()获取对应的方法
callback(data)  # 调用结果

Django源码中的典型应用

Django 的一个非常经典的视图类的实现，如下：

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
from django.http import HttpResponse
from django.views import View

class HelloView(View):

    def get(self, request, *args, **kwargs):  
        return HttpResponse('get\n')

    def post(self, request, *args, **kwargs):  
        return HttpResponse('post\n')

    def put(self, request, *args, **kwargs):  
        return HttpResponse('put\n')

    def delete(self, request, *args, **kwargs):  
        return HttpResponse('delete\n')

    # 注意，给CBV加上@csrf_exempt注解，需要加到dispatch方法上，后续会详解介绍这个函数的作用
    @csrf_exempt
    def dispatch(self, request, *args, **kwargs):
        return super(HelloView, self).dispatch(request, *args, **kwargs)

urlpatterns = [
    path('admin/', admin.site.urls),
    path('hello/', HelloView.as_view()),
]

只需理解，当客户端的HTTP请求（路径为hello/）为GET请求时，处理的视图方法为HelloView类中定义的get()方法，而同一路径的PUT请求对应着HelloView类中定义的put()方法。那Django框架内部是如何做到这一点的呢？Django中View类的源码，内容如下：

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
44
45
46
47
48
49
50
51
52
53
54
55
56
class View:
    """
    Intentionally simple parent class for all views. Only implements
    dispatch-by-method and simple sanity checking.
    """
    http_method_names = ['get', 'post', 'put', 'patch', 'delete', 'head', 'options', 'trace']

    # ...

    @classonlymethod
    def as_view(cls, **initkwargs):
        """Main entry point for a request-response process."""
        for key in initkwargs:
            if key in cls.http_method_names:
                raise TypeError(
                    'The method name %s is not accepted as a keyword argument '
                    'to %s().' % (key, cls.__name__)
                )
            if not hasattr(cls, key):
                raise TypeError("%s() received an invalid keyword %r. as_view "
                                "only accepts arguments that are already "
                                "attributes of the class." % (cls.__name__, key))

        def view(request, *args, **kwargs):
            self = cls(**initkwargs)
            self.setup(request, *args, **kwargs)
            if not hasattr(self, 'request'):
                raise AttributeError(
                    "%s instance has no 'request' attribute. Did you override "
                    "setup() and forget to call super()?" % cls.__name__
                )
            return self.dispatch(request, *args, **kwargs)
        view.view_class = cls
        view.view_initkwargs = initkwargs

        # take name and docstring from class
        update_wrapper(view, cls, updated=())

        # and possible attributes set by decorators
        # like csrf_exempt from dispatch
        update_wrapper(view, cls.dispatch, assigned=())
        return view

    # ...

    def dispatch(self, request, *args, **kwargs):
        # Try to dispatch to the right method; if a method doesn't exist,
        # defer to the error handler. Also defer to the error handler if the
        # request method isn't on the approved list.
        if request.method.lower() in self.http_method_names:  # 最核心的部分
            handler = getattr(self, request.method.lower(), self.http_method_not_allowed)
        else:
            handler = self.http_method_not_allowed
        return handler(request, *args, **kwargs)

    # ...

上面的源码中非常多地方用到了hasattr()、getattr()函数，其中实现请求自动映射方法的最关键地方是dispatch()方法中使用的getattr()语句，它将请求的方式值（如GET、POST、PUT等）转成小写后，然后获取视图对象中该请求的方法handler，然后使用该handler去处理对应的请求，这就是Django中对getattr()函数的一个典型应用场景。

Psutil

…

本文作者：liyijie

本文链接：https://liyijie.cn/2024/ops-automation-2/

文章默认以 署名-非商业性使用-相同方式共享 授权。

更新于 2024-03-19

# Linux# 自动化运维# Shell
上一篇
linux-network
下一篇
SQL常用命令