python
auggie
  2022-10-28 11:56:04 2022-10-28 11:56   2022-11-09 20:20:26    1.5k Words  6 Mins

背景

入职字节已经小半年了，对业务已经轻车熟路。导致可以有很多时间在工位上学习，由于岗位是 python，所以打算来一个 python 进阶。

最近学习了一些 golang 的特性，所以先学学 python 对 golang 特性的一些实现。

协程 & 装饰器

goroutine 的实现是非常方便和优雅的。虽然 python coroutine 实现繁琐了一点，但是还是有的。

协程对于IO密集型的程序的提速还是非常明显的。

使用 async 关键字可以定义协程函数

1
2
3
async def add(x):
    await asyncio.sleep(2)
    return x + 1

协程函数不能像普通函数一样直接 func() 调用，这样只会得到一个协程对象 

1
<coroutine object main at 0x1032987b0>

协程函数需要运行的时候，需要注册到事件循环中。

1
loop = asyncio.get_event_loop() # 获取事件循环

可以使用 await 关键字调用其他的协程函数。

  • await + 可等待对象（协程对象，Future，Task对象（IO等待））
  • 等待到对象的返回结果，才会继续执行后续代码
1
2
3
4
5
async def add(x):
    await asyncio.sleep(2)
    return x + 1

result = await asyncio.gather(*[add(i) for i in range(3)])

全部代码：

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
import asyncio
import time
import functools

def timer(f):
    @functools.wraps(f)
    async def inner(*args, **kwargs):
        start = time.time()
        await f(*args, **kwargs)
        print(time.time() - start)

    return inner


async def add(x):
    await asyncio.sleep(2)
    return x + 1


@timer
async def main():
    r = await asyncio.gather(*[add(i) for i in range(3)])
    print(r)


if __name__ == '__main__':
    asyncio.get_event_loop().run_until_complete(main())

一个控制 coroutine 并发量的装饰器👇

1
2
3
4
5
6
7
8
9
def concurrency_limit(limit):
    sem = asyncio.Semaphore(limit)
    def executor(func):
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            async with sem:
                return await func(*args, **kwargs)
        return wrapper
    return executor

asyncio

asyncio

python 异步 IO

asyncio 是用来编写 并发 代码的库，使用 async/await 语法。

asyncio 被用作多个提供高性能 Python 异步框架的基础，包括网络和网站服务，数据库连接库，分布式任务队列等等。

asyncio 往往是构建 IO 密集型和高层级 结构化 网络代码的最佳选择。

pytest

python 的一个第三方自动化测试框架，功能十分强大，类似于 go test

pytest 框架的约束：

  • 所有的单测文件名都需要满足test_*.py格式或***_test.py**格式。
  • 在单测文件中，测试类以Test开头，并且不能带有 init 方法(注意：定义class时，需要以T开头，不然pytest是不会去运行该class的)
  • 在单测类中，可以包含一个或多个test_开头的函数。
  • 在执行pytest命令时，会自动从当前目录及子目录中寻找符合上述约束的测试函数来执行。

-s 显示测试函数中的 print

-vv 查看详细 test

-q 查看简略 test

-pdb 失败时跳转到 pdb

–html 生成错误报告

指定测试函数：

1
2
pytest test_mod.py::test_func
pytest test_mod.py::TestClass::test_method

setup 和 teardown

pytest setup & teardown

  • setup_method，teardown_method 在测试类的每一个测试方法前后都执行一遍
  • setup_class，teardown_class 在测试类的前后都执行一遍
  • setup_function，teardown_function 在测试函数的前后执行一遍，不会影响测试类
  • setup_module，teardown_module 模块级测试

pytest.ini

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
[pytest]
# test路径
testpaths = .
# 命令行参数
addopts = -s --html=./report.html -vv
# 搜索文件名
python_files = test_*.py
# 搜索的类名
python_classes = Test_*
#搜索的函数名
python_functions = test_*
# 分组
markers = 
    g1: group1
    g2: group2
1
2
3
# 一些好用的 pytest 插件
pip install pytest-html
pip install pytest-sugar
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
# 跳过测试函数
@pytest.mark.skipif(func() == 1,reason = "跳过该函数")

# 标记为预期失败
@pytest.mark.xfail(func() == 1, reason="标注为预期失败")

# 函数参数赋值
@pytest.mark.parametrize("a,b",[(1,2),(0,3)]) # 参数a,b均被赋予两个值，函数会运行两遍
def test_a(self,a,b): # 参数必须和parametrize里面的参数一致
    print("test data:a=%d,b=%d"%(a,b))

## 或者使用函数返回值的形式
@pytest.mark.parametrize("a,b",return_test_data())

# 分组
@pytest.mark.组名

# 分组测试
pytest -m 组名

pytest blog 1

pytest blog 2

locust

轻量级 python 压测框架

on_site() 方法，用于对每一个 user 进行初始化。

wait_time 成员，让一个用户在执行完上一个 task 之后等待一段时间

  • between(start, end) 等待一个随机值
  • constant(second) 等待一个固定值

task 装饰器

tag 装饰器，用于标识每一个 task 的 tag，可以指定 --tags [tag] 用于载入测试，或者 --exclude-tags tag3 tag4 排斥 tag3 tag4.

验证 response 的正确性

使用 catch_response 参数和 success() & failure(msg="") 方法

1
2
3
4
5
with self.client.get("/", catch_response=True) as response:
    if response.text != "Success":
        response.failure("Got wrong response")
    elif response.elapsed.total_seconds() > 0.5:
        response.failure("Request took too long")
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
from locust import HttpUser, between, task, TaskSet, events

@events.test_start.add_listener
def on_test_start(**kwargs):
    print('test start')


@events.test_stop.add_listener
def on_test_stop(**kwargs):
    print('test end')

class Test(TaskSet):
    def on_start(self):
        print("user start")
    
    def on_stop(self):
        print("user stop")

    @task
    def test(self):
        url = '/api/v1/tts?reqid=$4f1d09bc-5b6a-11ed-9b4c-0a8d64818ea5'
        data = '{"app": {"appid": "sa_tts", "token": "access_token"}'
        with self.client.post(url=url, data=data, catch_response=True) as response:
            if response.text is not None:
                response.success()
            else:
                response.failure("text is none")



class WebsiteUser(HttpUser):
    wait_time = between(1, 1)
    tasks = [Test]
1
locust -f locustfile.py --host=http://speech.byted.org

networkx

网络介数中心性计算

networkx tutorial

CSDN networkx

创建图

1
2
3
4
5
G = nx.Graph()          # 无向图
G = nx.DiGraph()        # 有向图
G = nx.MultiGraph()     # 多重无向图
G = nx.MultiDigraph()   # 多重有向图
G.clear()               # 清空图

加点

1
2
3
4
5
6
7
8
G.add_node(1)
G.add_nodes_from([2, 3])

# add attribute (node, node_attribute_dict)
G.add_nodes_from([
    (4, {"color": "red"}),
    (5, {"color": "green"}),
])

加边

1
2
3
4
5
6
7
G.add_edge(1, 2)             # default edge data=1
G.add_edge(2, 3, weight=0.9) # specify edge data
 
elist = [(1, 2), (2, 3), (1, 4), (4, 2)]
G.add_edges_from(elist)
elist = [('a', 'b', 5.0), ('b', 'c', 3.0), ('a', 'c', 1.0), ('c', 'd', 7.3)]
G.add_weighted_edges_from(elist)

基本信息获取

1
2
3
4
5
obj.nodes
obj.edges
obj.degree

obj.degree([2, 4]) # 查看 2，4 节点的度

绘图

1
2
3
import matplotlib.pyplot as plt
nx.draw(G)
plt.show()
  • Post title：python
  • Post author：auggie
  • Create time：2022-10-28 11:56:04
  • Post link：https://ruanjiancheng.github.io/2022/10/28/python/
  • Copyright Notice：All articles in this blog are licensed under BY-NC-SA unless stating additionally.