Python

Python知识点大杂烩

Posted by iceyao on Thursday, March 21, 2024

Python知识点

什么是yield

yield的函数被称为生成器(generator)。跟普通函数不同的是，生成器是一个返回迭代器的函数，只能用于迭代操作。

def foo():
    print("开始...")
    while True:
        response = yield 1
        print("response = ", response)


def main():
    g = foo()
    print("第一次yield返回值:", next(g))
    print("第二次next")
    print("第二次yield返回值:", next(g))

输出：

开始...
第一次yield返回值: 1
第二次next
response =  None
第二次yield返回值: 1
  1. foo函数中包含yield关键字，所以foo函数不会真正执行，而是得到一个生成器g
  2. 对生成g调用next方法时，foo函数才会真正执行，先执行foo函数中的print(“开始…")，然后进入while循环
  3. 执行遇到yield关键字，先把yield看成是return，return 1之后，程序停止。并没有完成赋值给response的操作，到这里next(g)执行完成
  4. 执行print(“第二次next”)
  5. 执行新的next(g)，执行的位置要从上一个next方法停止的位置开始，即要完成赋值操作给response，因为上一个next已经return了，所以 右边相当于没有值，输出就是response = None
  6. 还没遇到第二次的yield，所以程序还未停止，进入while循环，遇到yield 1，返回打印1

使用send的例子


def foo():
    print("开始...")
    while True:
        response = yield 1
        print("response = ", response)


def main():
    g = foo()
    print("第一次yield返回值:", next(g))
    print("第二次next")
    print("第二次yield返回值:", g.send(2))

输出：

开始...
第一次yield返回值: 1
第二次next
response =  2
第二次yield返回值: 1

前面执行过程跟上面那个例子一样，从g.send(2)开始，程序会从上一个next()停止的下一步操作开始，send的话是会把2赋值给response变量。 其实next()和send()在一定意义上作用是相似的，区别是send()可以传递yield表达式的值进去，而next()不能传递特定的值，只能传递None进去。因此next(c)可以等价于c.send(None)。

生产者-消费者

yield/send实现

def consumer():
    r = ''
    while True:
        n = yield r
        if not n:
            return
        print('消费者: 消费 %s...' % n)
        r = '200 OK'


def producer(c):
    c.send(None)
    n = 0
    while n < 5:
        n = n + 1
        print('生产者: 生产 %s...' % n)
        r = c.send(n)
        print('生产者: 从消费里那里返回: %s' % r)
    c.close()


def main():
    c = consumer()
    producer(c)


if __name__ == "__main__":
    main()
  1. consumer函数定义了一个生成器，它能够消费由生产者发送的值。
  2. 在consumer函数中，while True 创建了一个无限循环，在这个循环中，生成器始终等待来自生产者的值。
  3. n = yield r 这一行既是yield表达式的输出，也是输入的接收点。该生成器一开始会返回空字符串 r=’’，之后每次循环会返回 ‘200 OK’ 给生产者，并且等待下一个来自生产者的send。
  4. 如果生产者发送了None或没有发送值（即 n 为假），消费者将会直接返回，生成器结束。

生产者的producer函数做了以下几件事情：

  1. 使用c.send(None) 启动或"预激活” 消费者生成器，这是开始发送值给生成器之前的必要步骤。
  2. produce 开始在一个循环中生产值，从1到5。
  3. 对于每个新的整数 n，生产者通过调用 c.send(n) 将其发送到消费者，同时接收从消费者返回的值，并将其打印出来。
  4. 生产者在发送了5个值后结束循环，并通过 c.close() 关闭消费者生成器。

main函数会先创建消费者生成器，然后启动生产者函数。最终运行脚本时，会依次打印出生产者生产的值和消费者消费的值，以及消费者返回的结果（‘200 OK’）。运行过程中，生产者和消费者通过send和yield操作进行交互，协同工作完成生产消费任务。

输出：

生产者: 生产 1...
消费者: 消费 1...
生产者: 从消费里那里返回: 200 OK
生产者: 生产 2...
消费者: 消费 2...
生产者: 从消费里那里返回: 200 OK
生产者: 生产 3...
消费者: 消费 3...
生产者: 从消费里那里返回: 200 OK
生产者: 生产 4...
消费者: 消费 4...
生产者: 从消费里那里返回: 200 OK
生产者: 生产 5...
消费者: 消费 5...
生产者: 从消费里那里返回: 200 OK

async/await协程实现

import asyncio
import random
import time


async def consumer(queue, id):
    while True:
        val = await queue.get()
        if val is None:  # 如果接收到None，则认为是结束信号
            # 通知队列任务完成
            queue.task_done()
            break
        print('{} get a val: {}'.format(id, val))
        # 模拟一秒钟处理时间
        await asyncio.sleep(1)
        # 通知队列任务完成
        queue.task_done()


async def producer(queue, id):
    for _ in range(5):
        val = random.randint(1, 10)
        await queue.put(val)
        print('{} put a val: {}'.format(id, val))
        await asyncio.sleep(1)


async def main():
    queue = asyncio.Queue()

    consumers = [asyncio.create_task(
        consumer(queue, f'consumer_{i+1}')) for i in range(2)]
    producers = [asyncio.create_task(
        producer(queue, f'producer_{i+1}')) for i in range(2)]

    # 等待所有生产者结束
    await asyncio.gather(*producers)

    # 发送结束信号给消费者，有几个消费者就发送几个None
    for _ in consumers:
        await queue.put(None)

    # 等待所有任务被消费
    await queue.join()  # 这确保了队列中的所有任务被处理

    # 取消所有消费者
    for c in consumers:
        c.cancel()

    # 等待所有消费者完成取消
    await asyncio.gather(*consumers, return_exceptions=True)


if __name__ == '__main__':
    start_time = time.time()
    asyncio.run(main())
    print('time cost:', time.time() - start_time)

输出：

producer_1 put a val: 7
producer_2 put a val: 5
consumer_1 get a val: 7
consumer_2 get a val: 5
producer_1 put a val: 9
producer_2 put a val: 2
consumer_1 get a val: 9
consumer_2 get a val: 2
producer_1 put a val: 9
producer_2 put a val: 3
consumer_1 get a val: 9
consumer_2 get a val: 3
producer_1 put a val: 5
producer_2 put a val: 1
consumer_1 get a val: 5
consumer_2 get a val: 1
producer_1 put a val: 6
producer_2 put a val: 7
consumer_1 get a val: 6
consumer_2 get a val: 7
time cost: 5.012088060379028

隐式new方法和init方法

  • __new__()方法用来创建实例，它是class的方法，是个静态方法，执行完了需要返回创建的类的实例。
  • __init__()方法用来初始化实例，在实例对象被创建后被调用，是实例对象的方法，通常用于设置实例对象的初始属性。__init__()方法将不返回任何信息。
  • 类中同时出现了__init__()方法和__new__()方法，调用顺序为：先调用__new__()方法，后调用__init__()方法。__new__()方法如果报错，则不会调用__init__()方法。

重写__new__方法

def __new__(cls):
      return super().__new__(cls) # 或return object.__new__(cls)

隐式new方法应用 - 单例模式

class TestCls:
    _xxx = None
    def __new__(cls):
        print('__new__ func called')
        if cls._xxx == None:
            cls._xxx = object.__new__(cls)
            return cls._xxx
        else:
            return cls._xxx
    def __init__(self):
        print('__init__ func called')


t1 = TestCls()
print(t1)
t2 = TestCls()
print(t2)

输出：

__new__ func called
__init__ func called
<__main__.TestCls object at 0x7f8fe75befe0>
__new__ func called
__init__ func called
<__main__.TestCls object at 0x7f8fe75befe0>

可以看出是同个对象，而且隐式new函数先于隐式init函数被调用

functools.wraps

保证被装饰器装饰后的函数还拥有原来的属性，wraps通过partial以及update_wrapper来实现。比如保留原有函数名和doc

没用wraps的装饰器

def decorate(func):
    def wrapper():
        return '<decorate>' + func() + '</decorate>'
    return wrapper


@decorate
def to_decorate():
    return 'to decorate!'


def not_to_decorate():
    return 'not to decorate'


print(to_decorate)
print(not_to_decorate)

输出：

<function decorate.<locals>.wrapper at 0x7fa1e6ddec20>
<function not_to_decorate at 0x7fa1e6ddfd90>

从输出来看，装饰后的函数名发生了变化

用了wraps的装饰器

import functools


def decorate(func):
    @functools.wraps(func)
    def wrapper():
        return '<decorate>' + func() + '</decorate>'
    return wrapper


@decorate
def to_decorate():
    return 'to decorate!'


def not_to_decorate():
    return 'not to decorate'


print(to_decorate)
print(not_to_decorate)

输出：

<function to_decorate at 0x7fd9bfdd2c20>
<function not_to_decorate at 0x7fd9bfdd3d90>

从输出来看，装饰后的函数跟原来的函数保持了一致。知乎上有篇详解functools.wraps原理的文章：https://zhuanlan.zhihu.com/p/45535784

contextmanager上下文管理器

在Python中，@contextmanager是一个装饰器，它允许你使用生成器来创建上下文管理器，而不需要显式地定义一个类并实现__enter__()和__exit__()特殊方法。上下文管理器是Python中的一个特性，主要用于资源的管理和清理，如文件操作、网络连接或数据库会话等，确保无论函数正常结束还是异常退出，资源都能得到正确的释放。

当你使用with语句时，上下文管理器会在进入with块前调用__enter__()方法，然后在退出with块后调用__exit__()方法。这对于确保资源的正确打开和关闭非常有用，即使在发生异常的情况下也能保证资源被妥善处理。

@contextmanager简化了这个过程，你可以定义一个生成器函数，使用yield语句来标记__enter__()和__exit__()之间的代码段。在yield之前的代码相当于__enter__()方法，在yield之后的代码相当于__exit__()方法。

from contextlib import contextmanager

@contextmanager
def managed_file(name):
    try:
        f = open(name, 'r')
        yield f
    finally:
        f.close()

# 使用上下文管理器
with managed_file('example.txt') as f:
    for line in f:
        print(line)

在这个例子中，managed_file函数使用@contextmanager装饰器转换成了一个上下文管理器。当with语句执行时，文件被打开，然后在with块结束后自动关闭，即使在处理文件的过程中发生了异常。这样可以确保文件句柄总是被正确地关闭，避免了资源泄露的问题。

overload装饰器

在Python中，并没有直接支持方法重载（overload）的语言特性，通过一些技巧来模拟方法重载的行为，可以使用typing模块中的overload来提供不同的签名作为类型提示。但实际上运行时只会保留最后一个定义的函数体。

from typing import overload

class MyClass:
    @overload
    def my_function(self, arg: int) -> int:
        ...
    
    @overload
    def my_function(self, arg: str) -> str:
        ...
    
    def my_function(self, arg):
        if isinstance(arg, int):
            return arg * 2
        elif isinstance(arg, str):
            return arg + " world"
        else:
            raise TypeError("Unsupported type")

# 使用示例
obj = MyClass()
print(obj.my_function(5))    # 输出: 10
print(obj.my_function("hello"))  # 输出: hello world

@overload装饰器主要用于文档和类型检查工具，而不是用于实际运行时的行为控制。实际上，只有最后一个定义的my_function会被使用。运行时的行为需要通过条件判断或其他逻辑来实现不同类型参数的处理。

dataclass装饰器

dataclass装饰器是从Python 3.7版本开始引入的，它简化了定义主要用来存储数据的类的过程。dataclass自动为类生成了一些特殊方法，如__init__, repr, __eq__等

一个使用dataclass的例子：

from dataclasses import dataclass, field

@dataclass
class Point:
    x: int
    y: int = field(repr=False)  # y坐标不显示在repr输出中
    z: int = field(default=10, repr=False)  # z坐标默认值为10且不显示在repr输出中
    t: int = 20                     # t坐标默认值为20

# 创建Point实例
p = Point(1, 2)

# 输出实例信息
print(p)  # 默认只显示x坐标，因为y和z的repr设置为False

# 访问实例属性
print(f"x: {p.x}, y: {p.y}, z: {p.z}, t: {p.t}")

# 检查两个实例是否相等
p2 = Point(1, 2)
print(p == p2)  # 输出True，因为x和y相同，而z和t默认相等

在这个例子中，Point类有四个属性：x, y, z, 和 t。

  • y和z的repr参数设置为False，这意味着它们不会出现在__repr__方法的输出中。
  • z和t有默认值，分别为10和20，这使得在创建Point实例时可以省略这两个参数。
  • field()函数用于提供额外的元数据或者改变字段的行为，比如设置默认值或控制__repr__的输出。

当运行这个代码，会看到Point实例的__repr__方法只显示了x坐标，而y和z坐标由于repr=False而不显示。同时，__eq__方法也自动实现了，允许比较两个Point实例是否相等。

pydantic数据校验库

Pydantic是一个 Python 库，用于数据声明和验证，它经常被用来构建API的请求和响应模型，尤其在 FastAPI和其他基于Starlette的框架中非常流行。下面是一些Pydantic 基础用法的例子：

  1. 基本数据类型验证
from pydantic import BaseModel

class User(BaseModel):
    id: int
    name: str
    age: int

user_data = {"id": 1, "name": "John Doe", "age": 30}
user = User(**user_data)
print(user)
  1. 可选字段和默认值
from pydantic import BaseModel

class User(BaseModel):
    id: int
    name: str
    age: int = None  # 可选字段，可以不提供，默认为 None

user_data = {"id": 1, "name": "John Doe"}
user = User(**user_data)
print(user)
  1. 验证错误处理
from pydantic import BaseModel, ValidationError

class User(BaseModel):
    id: int
    name: str
    age: int

try:
    user_data = {"id": "not an int", "name": "John Doe", "age": 30}
    user = User(**user_data)
except ValidationError as e:
    print(e)
  1. 字典转换
from pydantic import BaseModel

class User(BaseModel):
    id: int
    name: str
    age: int

user_data = {"id": 1, "name": "John Doe", "age": 30}
user = User(**user_data)
print(user.dict())  # 转换成字典
  1. JSON 序列化
from pydantic import BaseModel

class User(BaseModel):
    id: int
    name: str
    age: int

user_data = {"id": 1, "name": "John Doe", "age": 30}
user = User(**user_data)
print(user.json())  # 转换成 JSON 字符串
  1. 复杂的数据结构
from pydantic import BaseModel

class Address(BaseModel):
    street: str
    city: str

class User(BaseModel):
    id: int
    name: str
    address: Address

user_data = {
    "id": 1,
    "name": "John Doe",
    "address": {"street": "123 Main St", "city": "New York"}
}
user = User(**user_data)
print(user)

这些例子展示了Pydantic的一些基本功能，包括如何定义数据模型、如何进行数据验证以及如何处理验证错误。在实际应用中，Pydantic还提供了更复杂的功能，如嵌套模型、自定义验证器等

typing类型注解库

typing 库在 Python 中用于提供类型提示，这有助于增强代码的可读性和可维护性，同时也方便了 IDE 和 linter 进行更准确的代码分析和建议。下面是一些 typing 库中常见类型的使用例子：

  1. 基本类型
from typing import List, Dict, Union, Optional

def greet(name: str) -> None:
    print(f"Hello, {name}")

greet("Alice")  # 正确
greet(123)      # 错误，IDE 或 mypy 会警告类型不匹配
  1. 列表和字典
def process_data(data: List[int]) -> List[int]:
    return [x * 2 for x in data]

result = process_data([1, 2, 3])
print(result)  # 输出: [2, 4, 6]

def get_value(dct: Dict[str, int], key: str) -> Optional[int]:
    return dct.get(key)

d = {"a": 1, "b": 2}
print(get_value(d, "a"))  # 输出: 1
print(get_value(d, "c"))  # 输出: None
  1. 联合类型 (Union)
def safe_divide(a: Union[int, float], b: Union[int, float]) -> Union[int, float]:
    if b == 0:
        return None
    return a / b

print(safe_divide(10, 2))  # 输出: 5.0
print(safe_divide(10, 0))  # 输出: None
  1. 可选类型 (Optional)
def get_first_name(full_name: Optional[str]) -> Optional[str]:
    if full_name is None:
        return None
    return full_name.split()[0]

print(get_first_name("Alice Smith"))  # 输出: Alice
print(get_first_name(None))           # 输出: None
  1. 元组 (Tuple)
def point_in_2d(x: float, y: float) -> Tuple[float, float]:
    return (x, y)

p = point_in_2d(1.0, 2.0)
print(p)  # 输出: (1.0, 2.0)
  1. 泛型 (Generic)
from typing import TypeVar, Generic

T = TypeVar('T')

class Stack(Generic[T]):
    def __init__(self):
        self._container = []

    def push(self, item: T) -> None:
        self._container.append(item)

    def pop(self) -> T:
        return self._container.pop()

s = Stack[int]()
s.push(1)
s.push(2)
print(s.pop())  # 输出: 2

这些例子展示了如何使用typing库中的不同类型来注解函数参数和返回值，以及定义泛型类。在实际开发中，使用类型提示可以显著提高代码质量和开发效率。

python远程调试

debugpy是目前的主流python远程调试选择，属于微软开发的，在vscode中已经集成到插件Python Debugger中了. 它有两种模式：

  • vscode端是server，代码端是client
  • vscode端是client，代码端是server

vscode server模式

vscode server端launch.json配置

{
    // Use IntelliSense to learn about possible attributes.
    // Hover to view descriptions of existing attributes.
    // For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
    "version": "0.2.0",
    "configurations": [
        {
            "name": "Python Debugger: Remote Attach",
            "type": "debugpy",
            "request": "attach",
            "listen": {
                "host": "0.0.0.0",
                "port": 5678
            },
            "pathMappings": [
                {
                    "localRoot": "${workspaceFolder}",
                    "remoteRoot": "."
                }
            ]
        }
    ]
}

代码端client设置：在其代码主函数入口加入debugpy connect代码

import debugpy; debugpy.connect(5678)

一切设置完毕后，开始捕获断点： 1.vscode端server启动，并设置断点 2.代码端client运行 3.vscode端server捕获到断点执行处

vscode client模式

vscode launch.json配置

{
    // Use IntelliSense to learn about possible attributes.
    // Hover to view descriptions of existing attributes.
    // For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
    "version": "0.2.0",
    "configurations": [
        {
            "name": "Python Debugger: Remote Attach",
            "type": "debugpy",
            "request": "attach",
            "justMyCode": false,
            "connect": {
                "host": "<远程服务器ip>",
                "port": 5678
            },
            "pathMappings": [
                {
                    "localRoot": "${workspaceFolder}",
                    "remoteRoot": "<远程代码目录>"
                }
            ]
        }
    ]
}

举个例子，远程代码片段

def main():
    print("Hello, Debugger!")


if __name__ == "__main__":
    a = 1
    b = 2
    main()

在远程服务器加载debugpy模块启动，等待vscode连入，还有一种方式是在代码里import debugpy模块

# 这种方式更优雅，在远程服务器上安装debugpy
pip install debugpy
python3 -m debugpy --listen 0.0.0.0:5678 --wait-for-client b.py

在本地vscode就可以捕获到断点了，注意代码文件内容、代码文件名得一致

python常用工具

python内存分析工具 - memray python代码调试帮助工具 - pysnooper

参考链接

「真诚赞赏，手留余香」

爱折腾的工程师

真诚赞赏，手留余香

使用微信扫描二维码完成支付

CATALOG
    FEATURED TAGS
    devops go k8s llm openstack tkestack 练车